1. 核心痛点与解决思路
问题:当系统接入成百上千个 API(如 ERP、CRM、SaaS 接口)时,直接将其全部塞入 LLM 的上下文(Context Window)会导致:
- 上下文溢出:超过 Token 限制。
- 注意力分散:LLM 在海量选项中无法精准选择,幻觉率急剧上升。
- 成本高昂:每次请求都携带数万 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[最终响应] end3. 关键实现步骤
步骤一:Swagger 解析与结构化 (The Parser)
目标:将冗余的 Swagger JSON 转换为 LLM 易懂、且具备严格类型检查的StructuredTool。
- 核心动作:
- Schema 瘦身:只保留 Endpoint、Method、以及关键参数描述。剔除无用的 HTTP 状态码和复杂的嵌套 Response 定义。
- Pydantic 建模:利用 Python 的
pydantic库动态创建参数模型,确保 LLM 传参符合类型规范(如 Integer 不传 String)。 - Enum 注入:将枚举值明确写入描述,防止 LLM 编造参数。
步骤二:工具向量化 (The Indexing)
目标:让系统能够根据自然语言理解 API 的功能。
- 核心动作:
- 提取 API 的
description(功能描述)作为向量化的文本内容。 - 将
tool_name和 Swagger 的tags(分类)存为 Metadata,以便后续过滤。 - 存入 FAISS、Pinecone 或 Milvus 等向量库。
- 提取 API 的
步骤三:动态检索与绑定 (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. 提升检索精准度
- 混合检索 (Hybrid Search):结合关键词匹配(BM25)和向量检索。防止因专有名词(如 API 里的特殊字段名)被向量化模糊掉。
- 元数据过滤 (Metadata Filtering):
- 利用 Swagger 的
Tags。 - 先用一个轻量级 LLM 判断领域(如 “这是财务相关的问题”),然后在向量检索时加 Filter
filter={"category": "Finance"}。
- 利用 Swagger 的
B. 提升参数填充成功率
- Pydantic 校验回环:
- 当 LLM 传错参(如类型错误)时,Python 端捕获
ValidationError。 - 不要直接崩溃,而是将错误信息(“Error: ‘amount’ must be an integer”)作为 Observation 返回给 Agent。
- Agent 看到错误后会进行Self-Correction (自我修正)并重新发起调用。
- 当 LLM 传错参(如类型错误)时,Python 端捕获
- Schema 描述优化:
- 在
description中给出 Example。例如:“用于查询用户,输入参数示例:user_id=‘u123’”。
- 在
C. 处理上下文依赖
- 对话历史注入:检索工具时,不要只用当前的 Query。如果用户上一句说了 “我是管理员”,下一句说 “列出所有票据”,检索器需要知道 “管理员” 这个上下文才能匹配到高权限 API。建议总结历史对话后再进行检索。