如何在LangGraph Agent中集成外部API工具？

<think>我们正在讨论如何在LangGraph Agent中集成外部API工具。根据引用[1]和引用[2]，我们可以知道集成外部工具（如SearxNG）到LangGraphAgent中的关键步骤。结合引用[1]中提到的SearxNG集成方法，以及引用[2]中展示的LangGraph Agent与外部工具交互的架构，我们可以总结出以下步骤： 1. **创建工具包装器**：将外部API封装成LangChain工具（例如使用`BaseTool`类）。 2. **在LangGraph中注册工具**：通过`ToolNode`将工具集成到工作流中。 3. **配置工具调用逻辑**：在工作流中添加条件分支，决定何时调用工具。下面我们详细展开：###步骤1：创建外部API的工具包装器首先，我们需要将外部API封装成LangChain可以识别的工具。以集成SearxNG为例（如引用[1]所述），我们可以使用`SearxSearchWrapper`：```pythonfromlangchain.toolsimport BaseTool,Toolfromlangchain.utilitiesimport SearxSearchWrapperclass CustomSearxTool(BaseTool):name ="searx_search"description= "使用SearxNG搜索引擎进行网络搜索"def_run(self, query: str,**kwargs):#实例化SearxSearchWrappersearx= SearxSearchWrapper(searx_host="http://localhost:8888")returnsearx.run(query)async def_arun(self, query: str, **kwargs):#异步调用（可选）return awaitsearx.arun(query)``` ###步骤2：在LangGraph工作流中集成工具在LangGraph中，我们可以使用`ToolNode`来管理工具调用（引用[1]提到）： ```pythonfrom langgraph.graphimport StateGraphfrom langgraph.prebuilt importToolNode#定义状态结构class AgentState(TypedDict):user_input:strtool_outputs: dict={}iteration:int =0#初始化工作流workflow= StateGraph(AgentState) #创建工具节点tools= [CustomSearxTool(),CalculatorTool()]#假设CalculatorTool是另一个工具tool_node =ToolNode(tools) #添加工具节点到工作流workflow.add_node("tools", tool_node)``` ###步骤3：配置条件工具调用我们需要一个条件判断节点来决定是否调用工具（引用[2]中的条件分支）：```python#添加LLM节点（用于决定是否调用工具） workflow.add_node("llm", self.llm_node) #条件边：根据LLM输出决定下一步defshould_use_tool(state: AgentState)-> str:llm_output =state["llm_output"]#如果LLM输出包含工具调用指令，则转向工具节点if "tool_call"in llm_output:return "tools"else:return "end"workflow.add_conditional_edges("llm",should_use_tool,{"tools": "tools","end":END}) #从工具节点返回到LLM节点（以便处理工具结果） workflow.add_edge("tools","llm")``` ###步骤4：处理工具输出在工具调用后，我们需要将工具输出整合到状态中，并传递给LLM节点： ```pythondef llm_node(state:AgentState)-> dict:#如果有工具输出，将其添加到LLM的输入中ifstate.get("tool_outputs"):tool_results= "\n".join([f"{k}: {v}"for k,v instate["tool_outputs"].items()])prompt =f"工具输出：{tool_results}\n请根据以上信息回答用户问题：{state['user_input']}"else:prompt= state['user_input']#调用LLM（这里使用DeepSeek，如引用[3]）llm_response= deepseek_model.invoke(prompt)return {"llm_output":llm_response} ```###完整工作流示例```mermaidgraphTDA[输入节点] -->B(LLM节点)B--> C{是否调用工具?}C-->|是|D[工具节点]C -->|否| E[输出节点]D--> B```###最佳实践1. **错误处理**：在工具调用中捕获异常并返回错误信息[^3]：```pythondef_run(self,query:str,**kwargs):try:#调用APIexcept Exceptionas e:return f"工具调用失败: {str(e)}"```2.**异步支持**：对于耗时工具实现`_arun`方法，避免阻塞主线程[^1]。 3. **工具描述优化**：确保工具的描述清晰，帮助LLM准确理解工具用途[^2]：```pythondescription= "使用SearxNG搜索互联网。输入应为搜索查询字符串。"```###集成效果通过以上步骤，LangGraph Agent可以： -根据LLM决策动态调用外部API工具-将工具输出整合到后续处理流程-支持复杂工作流（如多工具顺序调用）[^1]§§相关问题§§1.如何为LangGraphAgent设计工具调用失败的重试机制？ 2.在LangGraph中如何实现多个工具的顺序调用？ 3.如何限制LangGraphAgent的工具调用权限？</think>### 在LangGraph Agent中集成外部API工具的完整方法 #### 一、核心集成步骤 1. **创建工具包装器** 将外部API封装成LangChain兼容的工具类： ```python from langchain.tools import BaseTool import requests class APITool(BaseTool): name = "external_api" description = "调用外部API服务" def _run(self, input: str) -> str: # 调用API示例 response = requests.post( "https://api.example.com/endpoint", json={"query": input}, headers={"Authorization": "Bearer YOUR_API_KEY"} ) return response.json()["result"] async def _arun(self, input: str) -> str: # 异步实现 return await self._run(input) ``` 2. **在LangGraph中注册工具** 使用`ToolNode`将工具集成到工作流： ```python from langgraph.graph import StateGraph from langgraph.prebuilt import ToolNode # 定义状态结构 class AgentState(TypedDict): user_input: str api_response: str = None workflow = StateGraph(AgentState) # 创建工具节点 api_tool = APITool() tool_node = ToolNode([api_tool]) workflow.add_node("api_tool", tool_node) ``` 3. **配置条件调用逻辑** 添加决策节点控制工具调用时机： ```python def should_use_api(state: AgentState) -> str: # 根据输入内容决定是否调用API if "查询" in state["user_input"]: return "invoke_api" return "direct_response" workflow.add_conditional_edges( "decision_node", should_use_api, { "invoke_api": "api_tool", "direct_response": "response_node" } ) ``` #### 二、高级集成模式（参考引用[1][2]） 1. **多工具集成架构** ```mermaid graph LR A[用户输入] --> B(LangGraph Agent) B --> C{决策节点} C -->|需要搜索| D[SearxNG工具] C -->|需要计算| E[计算器工具] C -->|需要API| F[外部API工具] D --> G[结果整合] E --> G F --> G G --> H[最终响应] ``` 2. **SearxNG集成示例**（引用[1]） ```python from langchain_community.tools import SearxSearchResults searx_tool = SearxSearchResults( searx_host="http://localhost:8080", # 自托管实例 engines=["google", "bing"], categories=["news", "web"] ) ``` 3. **MCP架构集成**（引用[2]） ```python # 连接到MCP服务器 mcp_tool = MCPTool( server_url="http://mcp-server/api", plugin_registry={ "weather": "get_weather", "stock": "get_stock_price" } ) ``` #### 三、最佳实践 1. **错误处理机制** ```python def _run(self, input: str) -> str: try: # API调用逻辑 except requests.exceptions.RequestException as e: return f"API错误: {str(e)}" except KeyError: return "API响应格式异常" ``` 2. **异步优化** ```python async def invoke_api(state: AgentState): # 并发调用多个API results = await asyncio.gather( api_tool.arun(state["query1"]), weather_tool.arun(state["query2"]) ) return {"api_responses": results} ``` 3. **安全控制** ```python # 在工具节点添加权限检查 def tool_node(state: AgentState): if not user_has_permission(state["user_id"], "api_access"): raise PermissionError("无API访问权限") return tool.run(state["input"]) ``` #### 四、调试与监控 1. **添加追踪钩子** ```python workflow.add_checkpoint( "api_call_checkpoint", lambda state: print(f"API调用状态: {state}") ) ``` 2. **性能监控** ```python import time def timed_api_call(input): start = time.perf_counter() result = api_tool.run(input) latency = time.perf_counter() - start log_metric("api_latency", latency) return result ``` ### 集成效果 1. **动态工作流**：根据输入内容智能路由到不同API工具[^1] 2. **错误隔离**：单个工具故障不影响整个工作流[^3] 3. **权限控制**：基于用户角色的细粒度工具访问[^2]