Anything LLM 文档检索增强实战：命令行接口开发指南

1次阅读

没有评论

共计 2923 个字符，预计需要花费 8 分钟才能阅读完成。

Anything LLM 作为一款强大的语言模型工具，其默认的 Web 界面虽然直观易用，但在自动化工作流中却显得力不从心。特别是当我们需要批量处理文档检索或集成到现有系统中时，缺乏命令行支持成为了一个明显的瓶颈。

交互方式受限 ：Web 界面难以实现自动化脚本调用
批量操作困难 ：无法高效处理大量文档的连续检索请求
集成成本高 ：难以与其他工具链无缝衔接

为解决上述问题，我们设计了一个基于 API 的命令行接口方案，核心架构如下：

客户端层 ：处理用户输入和结果展示
API 交互层 ：封装 Anything LLM 的 HTTP 接口
数据处理层 ：转换检索结果格式
缓存层 ：优化重复查询性能

推荐使用 Python 3.8+ 环境，安装依赖：

pip install requests python-dotenv tabulate

创建 .env 文件存储 API 密钥：

API_KEY=your_anything_llm_api_key
BASE_URL=https://api.anythingllm.com/v1

以下是 Python 实现的认证封装类：

import os
from dotenv import load_dotenv
import requests

class AnythingLLMClient:
    def __init__(self):
        load_dotenv()
        self.api_key = os.getenv('API_KEY')
        self.base_url = os.getenv('BASE_URL')

    def _make_request(self, endpoint, params=None):
        headers = {'Authorization': f'Bearer {self.api_key}'}
        response = requests.get(f'{self.base_url}/{endpoint}',
            headers=headers,
            params=params,
            timeout=30
        )
        response.raise_for_status()
        return response.json()

实现结果格式化输出的关键方法：

from tabulate import tabulate

class ResultFormatter:
    @staticmethod
    def format_as_table(results):
        headers = ['Rank', 'Document', 'Relevance']
        rows = []
        for idx, item in enumerate(results['items'], 1):
            rows.append([
                idx,
                item['metadata']['title'][:50] + '...',
                f"{item['score']:.2f}%"
            ])
        return tabulate(rows, headers=headers, tablefmt='grid')

整合核心功能的命令行工具：

import argparse
import logging
from typing import Dict, Any

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class AnythingLLMCLI:
    def __init__(self):
        self.client = AnythingLLMClient()

    def search(self, query: str, top_k: int = 5) -> Dict[str, Any]:
        params = {
            'q': query,
            'limit': top_k
        }
        try:
            results = self.client._make_request('search', params)
            return ResultFormatter.format_as_table(results)
        except Exception as e:
            logger.error(f"Search failed: {str(e)}")
            raise

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('query', help='Search query text')
    parser.add_argument('--top', type=int, default=5, help='Number of results')
    args = parser.parse_args()

    cli = AnythingLLMCLI()
    print(cli.search(args.query, args.top))

请求批处理 ：将多个查询合并为单个 API 调用
本地缓存 ：使用磁盘或内存缓存高频查询结果
并发请求 ：对独立查询采用多线程处理

from concurrent.futures import ThreadPoolExecutor
import hashlib
import json
import os

class CachedClient(AnythingLLMClient):
    def __init__(self, cache_dir='.cache'):
        super().__init__()
        os.makedirs(cache_dir, exist_ok=True)
        self.cache_dir = cache_dir

    def _get_cache_key(self, params):
        return hashlib.md5(json.dumps(params).encode()).hexdigest()

    def _make_request(self, endpoint, params=None):
        cache_key = self._get_cache_key(params)
        cache_file = os.path.join(self.cache_dir, f'{cache_key}.json')

        if os.path.exists(cache_file):
            with open(cache_file) as f:
                return json.load(f)

        result = super()._make_request(endpoint, params)
        with open(cache_file, 'w') as f:
            json.dump(result, f)
        return result