实现智能体调用海量api

针对成百上千个数据 API 接口的场景，单纯依靠 LLM 的上下文（Context Window）硬塞是行不通的。目前业界公认的主流方案是“RAG for Tools”（基于检索的工具增强生成）。

其核心思想是：“先检索，再调用”。只在运行时动态地将最相关的 3-5 个 API 挂载给 LLM，从而保证准确率并节省 Token。

以下是一个完整的、可落地的技术方案，包含架构设计、核心代码实现和优化策略。

一、 整体架构设计

我们将系统分为两个阶段：离线准备阶段和在线运行阶段。

graph TD subgraph "1. 离线准备 (Indexing)" Swagger[Swagger/OpenAPI 文档] --> Cleaner[文档清洗与解析] Cleaner -->|生成| ToolDefs[标准工具定义 (JSON/Pydantic)] ToolDefs -->|Embedding| VectorDB[(向量数据库\nMilvus/FAISS)] end subgraph "2. 在线运行 (Runtime)" User[用户提问: '查下上月销售额'] --> Router[意图识别/路由器] Router -->|语义搜索| VectorDB VectorDB -->|返回 Top-K| Candidates[候选 API 列表] Candidates -->|动态挂载| Agent[LLM 智能体] User --> Agent Agent -->|1. 决策调用| APICall[生成 API 请求参数] APICall -->|2. HTTP 请求| Backend[数据中台/后端 API] Backend -->|3. 返回 JSON| Agent Agent -->|4. 数据分析与总结| FinalResponse[最终回答] end

二、 核心实现步骤

1. 技术栈推荐

• 编排框架: LangChain (Python 版 v0.2/v0.3)
• 大模型: GPT-4o, Claude 3.5 Sonnet (Tool Calling 能力强), 或 Qwen-2.5-72B (私有化推荐)
• 向量库: FAISS (轻量), Milvus (生产级)
• API 描述标准: OpenAPI (Swagger)

2. 代码实现详解

这里提供一个基于LangChain的核心实现 Demo。

第一步：定义 API 并建立索引 (Mock 数据)

在真实场景中，你需要写脚本遍历 Swagger JSON，提取 path, description, parameters。

fromlangchain_core.toolsimportStructuredToolfromlangchain_core.documentsimportDocumentfromlangchain_community.vectorstoresimportFAISSfromlangchain_openaiimportOpenAIEmbeddings,ChatOpenAIfromlangchain.agentsimportcreate_tool_calling_agent,AgentExecutorfromlangchain_core.promptsimportChatPromptTemplatefrompydanticimportBaseModel,Field# --- 1. 定义数据 API 的 Schema (模拟从 Swagger 解析而来) ---classSalesQuerySchema(BaseModel):start_date:str=Field(description="开始日期, 格式 YYYY-MM-DD")end_date:str=Field(description="结束日期, 格式 YYYY-MM-DD")defquery_sales_data(start_date:str,end_date:str):"""查询指定时间范围内的公司销售总额。"""# 模拟真实 HTTP 请求return{"total_sales":100000,"currency":"CNY","period":f"{start_date}to{end_date}"}classUserInfoSchema(BaseModel):user_id:str=Field(description="用户ID")defget_user_profile(user_id:str):"""根据 ID 获取用户的详细画像和等级信息。"""return{"id":user_id,"name":"Alice","level":"VIP"}# 假设你有 1000 个这样的工具all_tools_map={"query_sales_data":StructuredTool.from_function(func=query_sales_data,name="query_sales_data",description="查询指定时间范围内的公司销售总额",args_schema=SalesQuerySchema),"get_user_profile":StructuredTool.from_function(func=get_user_profile,name="get_user_profile",description="根据 ID 获取用户的详细画像和等级信息",args_schema=UserInfoSchema),# ... 更多工具 ...}# --- 2. 建立向量索引 (Indexing) ---# 将工具的功能描述向量化docs=[]forname,toolinall_tools_map.items():docs.append(Document(page_content=tool.description,# 检索的关键：描述必须准确metadata={"tool_name":name}))embeddings=OpenAIEmbeddings()# 使用 FAISS 建立索引vector_store=FAISS.from_documents(docs,embeddings)retriever=vector_store.as_retriever(search_kwargs={"k":3})# 每次只找最相关的3个

第二步：构建“检索-执行”智能体

这是运行时的逻辑：用户提问 -> 检索 -> 动态组装 Agent。

# --- 3. 运行时逻辑 ---defrun_data_agent(user_query:str):print(f"--- 用户提问:{user_query}---")# 3.1 检索阶段：找到相关工具retrieved_docs=retriever.invoke(user_query)# 根据名字找回真实的工具对象selected_tools=[all_tools_map[d.metadata["tool_name"]]fordinretrieved_docs]print(f"Step 1: 检索到的相关接口 ->{[t.namefortinselected_tools]}")ifnotselected_tools:return"抱歉，没有找到相关的数据接口。"# 3.2 动态构建 Agentllm=ChatOpenAI(model="gpt-4o",temperature=0)prompt=ChatPromptTemplate.from_messages([("system","你是一个数据分析助手。当前时间是 2025-12-23。请根据用户的要求，选择合适的工具获取数据。"),("human","{input}"),("placeholder","{agent_scratchpad}"),])# 将检索到的 tools 绑定给 LLMagent=create_tool_calling_agent(llm,selected_tools,prompt)agent_executor=AgentExecutor(agent=agent,tools=selected_tools,verbose=True)# 3.3 执行与回答result=agent_executor.invoke({"input":user_query})returnresult["output"]# --- 测试 ---# 用户问模糊的时间，LLM 需要结合 System Prompt 里的当前时间推算 API 参数response=run_data_agent("帮我查一下上个月的销售额")print(f"Final Answer:{response}")

三、 关键难点与解决方案

在处理“成百上千”个接口时，你会遇到准确率和性能问题，以下是进阶优化方案：

1. 提高检索准确率 (The Precision Problem)

如果只用简单的语义检索，可能会出现查错接口的情况（比如“查询订单”和“查询退单”描述很像）。

• 方案 A：元数据过滤 (Metadata Filtering)
  • 在 Swagger 中提取tags(如Finance,HR,SupplyChain)。
  • 先用一个轻量级 LLM 判断用户意图属于哪个领域（例如判断为Finance）。
  • 并在向量检索时加上过滤条件filter={"tag": "Finance"}。
• 方案 B：混合检索 (Hybrid Search)
  • 结合BM25 (关键词匹配)和Vector (语义匹配)。如果用户准确说出了接口名或特有的参数名（如SKU_ID），关键词匹配权重更高。

2. 处理复杂参数 (The Parameter Problem)

用户提问通常很模糊（“查那个谁的订单”），而 API 需要精准的 ID。

• 方案：多步推理 (ReAct)
  • Agent 发现参数缺失时，不要瞎填。
  • 可以在 Prompt 中设定：“如果缺少参数（如 user_id），请先调用搜索接口（search_user）获取，或者反问用户。”

3. 降低 Token 消耗 (The Context Problem)

Swagger 文档通常极其冗长。

• 方案：Schema 瘦身
  • 不要把完整的 Swagger JSON 喂给 LLM。
  • 编写脚本清洗 Swagger，只保留description,parameters(name, type, required), 和return的简要描述。剔除 404/500 等错误码定义。

4. 安全性控制 (Safety)

• 方案：只读原则
  • 对于数据问数场景，建议严格过滤 GET 请求。
  • 对于 POST/DELETE 等修改操作，必须在 Agent 执行前增加“人机确认 (Human-in-the-loop)”环节，让用户点击“确认执行”。

四、 推荐的数据流转图

用户: "上个月北京地区的空调销售额是多少？" | v [意图分析] -> 提取关键词: "销售额", "上个月", "北京", "空调" | [向量检索] -> 在 1000 个 API 中找到 Top 3: 1. get_sales_summary(date_range, region, category) <-- 命中 2. get_inventory(product_id) 3. get_weather(city) | [LLM 推理] -> 选中 API 1 -> 转换参数: date_range="2025-11-01 to 2025-11-30" (根据当前时间推算) region="Beijing" category="Air Conditioner" | [API 执行] -> 调用后端接口 -> 返回 JSON { "amount": 500000 } | [LLM 生成] -> "上个月（11月）北京地区空调销售额为 50 万元。"

总结

要实现这个系统，不要试图训练一个能记住所有 API 的模型。LangChain + 向量数据库 + 动态工具挂载是目前最成熟、性价比最高的方案。