基于 LangChain 的海量 API 动态检索与调用架构

1. 核心痛点与解决思路

问题：当系统接入成百上千个 API（如 ERP、CRM、SaaS 接口）时，直接将其全部塞入 LLM 的上下文（Context Window）会导致：

1. 上下文溢出：超过 Token 限制。
2. 注意力分散：LLM 在海量选项中无法精准选择，幻觉率急剧上升。
3. 成本高昂：每次请求都携带数万 Token 的无用信息。

解决方案：采用RAG for Tools（工具检索增强）模式。

• 离线阶段：解析 API 文档，建立向量索引。
• 在线阶段：根据用户意图检索 Top-N 相关 API -> 动态绑定到 LLM -> 执行多步推理。

2. 整体架构设计

graph TD subgraph "准备阶段 (Offline)" Swagger[Swagger/OpenAPI 文档] --> Parser[自定义解析器 & 清洗] Parser -->|生成 Pydantic Schema| Tools[结构化工具定义] Tools -->|Embedding 描述信息| VectorDB[(向量数据库 / FAISS)] end subgraph "运行阶段 (Runtime)" User[用户指令: '给 Alice 发红包'] --> Retriever[检索层] VectorDB -->|检索 Top-5 相关工具| SelectedTools[动态工具集] SelectedTools --> Binder[动态绑定层 (LLM.bind_tools)] Binder --> Agent[智能体执行循环 (Agent Loop)] Agent -->|思考 1| Step1[调用 search_user API] Step1 -->|返回 ID| Agent Agent -->|思考 2| Step2[调用 send_bonus API] Step2 -->|返回结果| Final[最终响应] end

3. 关键实现步骤

步骤一：Swagger 解析与结构化 (The Parser)

目标：将冗余的 Swagger JSON 转换为 LLM 易懂、且具备严格类型检查的StructuredTool。

• 核心动作：
  1. Schema 瘦身：只保留 Endpoint、Method、以及关键参数描述。剔除无用的 HTTP 状态码和复杂的嵌套 Response 定义。
  2. Pydantic 建模：利用 Python 的pydantic库动态创建参数模型，确保 LLM 传参符合类型规范（如 Integer 不传 String）。
  3. Enum 注入：将枚举值明确写入描述，防止 LLM 编造参数。

步骤二：工具向量化 (The Indexing)

目标：让系统能够根据自然语言理解 API 的功能。

• 核心动作：
  1. 提取 API 的description（功能描述）作为向量化的文本内容。
  2. 将tool_name和 Swagger 的tags（分类）存为 Metadata，以便后续过滤。
  3. 存入 FAISS、Pinecone 或 Milvus 等向量库。

步骤三：动态检索与绑定 (Retrieval & Binding)

目标：在运行时（Runtime）动态构建 Prompt。

• 逻辑流程：
  User Query->Vector Search->Get Top-K Tools->LLM.bind_tools(Top-K)

4. 核心代码实现 (Python + LangChain v0.3)

以下代码展示了从“解析”到“检索”再到“Agent执行”的完整闭环。

importrequestsfromlangchain_core.toolsimportStructuredToolfromlangchain_core.documentsimportDocumentfromlangchain_community.vectorstoresimportFAISSfromlangchain_openaiimportOpenAIEmbeddings,ChatOpenAIfromlangchain.agentsimportAgentExecutor,create_tool_calling_agentfromlangchain_core.promptsimportChatPromptTemplatefrompydanticimportcreate_model# ==========================================# 1. 模拟 Swagger 解析器 (解析 + Pydantic 建模)# ==========================================defgenerate_tools_from_meta(api_definitions):tools=[]forapiinapi_definitions:# 动态创建参数模型 (关键：保证精准度)param_model=create_model(f"{api['name']}_args",**api['parameters'])# 定义执行函数def_func(**kwargs):# 实际场景中这里是 requests.get/postreturnf"调用{api['name']}成功，参数:{kwargs}"# 封装为 LangChain Tooltools.append(StructuredTool.from_function(func=_func,name=api['name'],description=api['description'],# 向量检索的核心依据args_schema=param_model))returntools# 模拟 API 定义 (实际应从 Swagger JSON 读取)mock_apis=[{"name":"search_user","description":"根据姓名查找用户ID和信息","parameters":{"name":(str,...)}},{"name":"get_user_balance","description":"查询用户的钱包余额","parameters":{"user_id":(str,...)}},{"name":"send_bonus","description":"给指定用户ID发放奖金","parameters":{"user_id":(str,...),"amount":(int,...)}},{"name":"get_weather","description":"查询天气","parameters":{"city":(str,...)}},# ... 假设这里还有 900 个其他 API ...]all_tools=generate_tools_from_meta(mock_apis)tool_map={t.name:tfortinall_tools}# 方便后续通过名字找回对象# ==========================================# 2. 建立向量索引 (Indexing)# ==========================================docs=[Document(page_content=t.description,metadata={"tool_name":t.name})fortinall_tools]embeddings=OpenAIEmbeddings()vector_store=FAISS.from_documents(docs,embeddings)retriever=vector_store.as_retriever(search_kwargs={"k":3})# 每次只取 Top 3# ==========================================# 3. 运行时逻辑：检索 + Agent 执行# ==========================================defrun_agent_with_retrieval(user_query):print(f"--- 用户指令:{user_query}---")# 3.1 检索阶段retrieved_docs=retriever.invoke(user_query)selected_tools=[tool_map[d.metadata["tool_name"]]fordinretrieved_docs]print(f" [检索命中]:{[t.namefortinselected_tools]}")# 3.2 绑定阶段llm=ChatOpenAI(model="gpt-4o",temperature=0)# 3.3 构建 Agent (处理多步依赖)# ReAct / Tool Calling Promptprompt=ChatPromptTemplate.from_messages([("system","你是一个助手，请使用提供的工具解决问题。如果需要，你可以分多步执行。"),("human","{input}"),("placeholder","{agent_scratchpad}"),# 记忆中间步骤 (如 Step1 查到的 ID)])agent=create_tool_calling_agent(llm,selected_tools,prompt)agent_executor=AgentExecutor(agent=agent,tools=selected_tools,verbose=True)# 3.4 执行agent_executor.invoke({"input":user_query})# ==========================================# 4. 测试用例# ==========================================# 场景：多步调用 (先查人，再发钱)# 预期检索：search_user, send_bonusrun_agent_with_retrieval("给 Alice 发 100 块钱奖金")

5. 专家级优化建议 (Best Practices)

为了在生产环境中达到 99% 的可用性，建议实施以下策略：

A. 提升检索精准度

1. 混合检索 (Hybrid Search)：结合关键词匹配（BM25）和向量检索。防止因专有名词（如 API 里的特殊字段名）被向量化模糊掉。
2. 元数据过滤 (Metadata Filtering)：
   • 利用 Swagger 的Tags。
   • 先用一个轻量级 LLM 判断领域（如 “这是财务相关的问题”），然后在向量检索时加 Filterfilter={"category": "Finance"}。

B. 提升参数填充成功率

1. Pydantic 校验回环：
   • 当 LLM 传错参（如类型错误）时，Python 端捕获ValidationError。
   • 不要直接崩溃，而是将错误信息（“Error: ‘amount’ must be an integer”）作为 Observation 返回给 Agent。
   • Agent 看到错误后会进行Self-Correction (自我修正)并重新发起调用。
2. Schema 描述优化：
   • 在description中给出 Example。例如：“用于查询用户，输入参数示例：user_id=‘u123’”。

C. 处理上下文依赖

1. 对话历史注入：检索工具时，不要只用当前的 Query。如果用户上一句说了 “我是管理员”，下一句说 “列出所有票据”，检索器需要知道 “管理员” 这个上下文才能匹配到高权限 API。建议总结历史对话后再进行检索。