4.1 工具调用与集成 (终)

完整示例

下面是一个完整的工具集成系统示例，包含所有上述功能：

"""
LangGraph 工具集成完整示例
实现一个智能搜索和计算系统
"""

from typing import Dict, List, Any, Optional
from langgraph.graph import StateGraph
from langgraph.prebuilt import ToolNode
from langgraph.graph.message import AIMessage, HumanMessage
from langchain_core.tools import tool
from dataclasses import dataclass
from enum import Enum
import time
import json

# 工具定义
@tool
def web_search(query: str, max_results: int = 10) -> str:
    """网络搜索工具"""
    # 模拟API调用
    return f"搜索 '{query}' 返回 {max_results} 条结果"

@tool
def calculator(expression: str) -> str:
    """数学计算工具"""
    try:
        # 安全地评估表达式
        allowed_chars = set('0123456789+-*/(). ')
        if not all(c in allowed_chars for c in expression):
            return "非法字符，只允许数字和基本运算符"
        
        result = eval(expression)
        return f"计算结果: {expression} = {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"

@tool
def data_analyzer(data: str, operation: str = "summary") -> str:
    """数据分析工具"""
    try:
        # 解析JSON数据
        if data.strip().startswith('['):
            parsed_data = json.loads(data)
            if operation == "summary":
                return f"数据分析摘要: 共 {len(parsed_data)} 项数据"
            elif operation == "count":
                return f"数据计数: {len(parsed_data)} 项"
            else:
                return f"未知的操作: {operation}"
        else:
            return "数据格式不正确，需要JSON格式"
    except Exception as e:
        return f"数据分析错误: {str(e)}"

# Agent状态定义
class AgentState(dict):
    messages: list = []
    tool_results: Dict[str, Any] = {}

# 工具管理器
class ToolManager:
    def __init__(self):
        self.tools = {
            "web_search": web_search,
            "calculator": calculator,
            "data_analyzer": data_analyzer
        }
        
    def get_tool(self, name: str):
        return self.tools.get(name)
    
    def list_tools(self) -> List[str]:
        return list(self.tools.keys())

# Agent系统
class LangGraphToolAgent:
    def __init__(self):
        self.tool_manager = ToolManager()
        self.tool_node = ToolNode(self.tool_manager.tools)
        self.graph = self._create_graph()
        
    def _create_graph(self) -> StateGraph:
        """创建状态图"""
        workflow = StateGraph(AgentState)
        
        # 添加节点
        workflow.add_node("agent", self._agent_node)
        workflow.add_node("tools", self.tool_node)
        
        # 添加边
        workflow.add_edge("agent", "tools")
        workflow.add_edge("tools", "agent")
        
        return workflow.compile()
    
    def _agent_node(self, state: AgentState) -> Dict:
        """Agent节点"""
        messages = state["messages"]
        
        if not messages:
            return {
                "messages": [AIMessage(
                    content="你好！我是AI助手，可以帮你进行网络搜索、数学计算和数据分析。\n"
                           "可用的工具：\n"
                           "- web_search: 网络搜索\n"
                           "- calculator: 数学计算\n"
                           "- data_analyzer: 数据分析\n\n"
                           "请告诉我你需要什么帮助？"
                )]
            }
        
        # 获取用户输入
        last_message = messages[-1]
        if isinstance(last_message, HumanMessage):
            content = last_message.content.lower()
            tool_results = state.get("tool_results", {})
            
            # 根据用户输入选择合适的工具
            if "搜索" in content or "search" in content:
                return {
                    "messages": [AIMessage(content="我将使用网络搜索工具来帮您搜索。")],
                    "tool_results": tool_results
                }
            elif any(word in content for word in ["计算", "计算器", "calculator", "math", "数学"]):
                return {
                    "messages": [AIMessage(content="我将使用计算器工具来帮您计算。")],
                    "tool_results": tool_results
                }
            elif "分析" in content or "analysis" in content or "data" in content:
                return {
                    "messages": [AIMessage(content="我将使用数据分析工具来帮您分析数据。")],
                    "tool_results": tool_results
                }
            else:
                return {
                    "messages": [AIMessage(
                        content="我理解您的需求。请告诉我您想要搜索什么、计算什么，或者提供需要分析的数据。"
                    )],
                    "tool_results": tool_results
                }
        
        return {"messages": messages, "tool_results": tool_results}

# 使用示例
def main():
    print("=== LangGraph 工具集成示例 ===")
    
    # 创建Agent
    agent = LangGraphToolAgent()
    
    # 测试工具
    print("\n--- 工具测试 ---")
    tool_manager = ToolManager()
    
    # 测试网络搜索
    result = web_search("LangGraph教程", max_results=5)
    print(f"网络搜索: {result}")
    
    # 测试数学计算
    result = calculator("2 + 2 * 3")
    print(f"数学计算: {result}")
    
    # 测试数据分析
    test_data = '[{"name": "AI", "value": 100}, {"name": "ML", "value": 80}, {"name": "DL", "value": 60}]'
    result = data_analyzer(test_data, "summary")
    print(f"数据分析: {result}")
    
    print("\n=== 系统就绪 ===")
    print("工具已注册并测试完成，Agent系统可以使用了！")

if __name__ == "__main__":
    main()
​

常见问题 FAQ

Q1：如何选择合适的工具？

A： 工具选择应该基于以下几个因素：

1. 任务类型匹配：根据任务的具体需求选择工具，如搜索任务选择web_search，计算任务选择calculator
2. 性能要求：考虑工具的响应时间、准确率和可靠性
3. 依赖资源：评估工具所需的外部依赖和资源消耗
4. 错误处理：选择具备完善错误处理机制的工具

Q2：工具调用失败时如何处理？

A： 采用多层次错误处理策略：

1. 重试机制：实现指数退避重试，避免立即重试导致的问题加剧
2. 超时控制：设置合理的超时时间，防止长时间阻塞
3. 降级处理：当主工具失败时，使用备选工具或缓存数据
4. 错误分类：根据错误类型选择不同的处理策略

Q3：如何优化工具调用的性能？

A： 性能优化可以从多个维度入手：

1. 并行调用：使用异步并发执行多个工具调用
2. 缓存机制：对频繁调用的工具结果进行缓存
3. 工具预热：对耗时较长的工具进行预热加载
4. 负载均衡：在多个同类工具间进行负载分配

Q4：工具调用安全如何保证？

A： 工具安全需要考虑：

1. 输入验证：对所有工具输入进行严格的格式和内容验证
2. 权限控制：实现工具调用权限管理和访问控制
3. 审计日志：记录所有工具调用的详细信息用于审计
4. 沙箱隔离：对高风险工具进行沙箱隔离执行

Q5：如何实现工具间的数据传递？

A： 工具间数据传递可以通过以下方式实现：

1. 状态管理：使用Agent状态在不同工具间共享数据
2. 参数传递：将前一个工具的结果作为后一个工具的输入
3. 全局存储：使用数据库或缓存系统存储中间结果
4. 消息队列：通过消息队列实现异步工具间的数据传递

最佳实践与避坑

• 工具注册规范：为每个工具提供清晰的文档和参数说明
• 错误处理完整：每个工具都应该具备完善的错误处理机制
• 性能监控：实时监控工具调用的性能指标，及时发现性能问题
• 版本管理：对工具进行版本管理，支持向后兼容
• 测试覆盖：为工具系统编写全面的单元测试和集成测试

本节小结

通过本节的学习，我们掌握了LangGraph工具集成的核心概念和实现方法。从基础的工具注册到动态工具选择，从执行监控到错误恢复，再到高级工具编排，我们构建了一个完整的工具集成系统。这个系统不仅能够处理简单的工具调用，还能支持复杂的工具链和条件执行，为构建强大的AI智能体奠定了基础。

下一节我们将深入探讨条件边与决策逻辑，学习如何让工具调用更加智能和灵活。

延伸阅读

• LangGraph官方文档 - 工具集成
• LangChain工具系统指南
• 异步工具调用最佳实践
• 工具编排和调度系统设计
• 错误处理和容错机制

关键词：LangGraph, 工具集成, 工具调用, 工具编排, 智能工具, 动态选择, 错误恢复, 并发执行, 工具监控, 技术教程, 实战指南
难度：进阶
预计阅读：45分钟