河池市网站建设_网站建设公司_云服务器_seo优化
2025/12/22 23:15:25 网站建设 项目流程

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。建议总结历史对话后再进行检索。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询