4.1 工具调用与集成 — LangGraph Agent 工具集成

本节导读：掌握LangGraph中智能工具调用的核心机制，从基础工具注册到高级动态工具选择，完整构建能与外部世界交互的Agent系统。

学习目标

• 理解LangGraph中工具注册和调用的基本概念
• 掌握StructuredTool和BaseTool的使用方法
• 实现工具动态选择和条件调用机制
• 构建具备错误恢复能力的工具调用流程
• 优化工具性能和响应质量

核心概念

LangGraph的工具集成系统是一个开放式的工具生态系统，让AI智能体能够调用外部API、函数和数据库等资源。与LangChain的工具系统相比，LangGraph更强调工具的状态感知和执行控制，工具调用可以持久化到检查点，支持故障恢复和人工干预。

工具系统的架构层次

环境准备 / 前置知识

• Python 3.8+
• LangGraph >= 0.2.0
• LangChain >= 0.2.0
• 基础的异步编程知识
• RESTful API 调用经验

分步实战

步骤1：基础工具注册

首先创建一个基础工具注册和调用系统：

from typing import Dict, Any, Optional, Union
from langgraph.graph import StateGraph
from langgraph.prebuilt import ToolNode
from langgraph.graph.message import AIMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from langchain_core.tools.structured import StructuredTool
from langchain_core.pydantic_v1 import BaseModel, Field
import asyncio

# 定义工具输入参数模型
class SearchInput(BaseModel):
    query: str = Field(description="搜索查询关键词")
    max_results: int = Field(default=10, description="最大返回结果数")

class WebSearchTool(BaseModel):
    """网络搜索工具"""
    name: str = "web_search"
    description: str = "使用搜索引擎进行网络搜索"
    args_schema: type[SearchInput] = SearchInput

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

# 另一个工具：数学计算
class CalculatorInput(BaseModel):
    expression: str = Field(description="数学表达式")

@tool
def calculator(expression: str) -> str:
    """数学计算工具"""
    try:
        result = eval(expression)
        return f"计算结果: {expression} = {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"

# 定义Agent状态
class AgentState(dict):
    messages: list = []

# 创建状态图
def create_agent_with_tools():
    # 工具列表
    tools = [web_search, calculator]
    
    # 创建工具节点
    tool_node = ToolNode(tools)
    
    # 创建状态图
    workflow = StateGraph(AgentState)
    
    # 添加节点
    workflow.add_node("agent", call_agent)
    workflow.add_node("tools", tool_node)
    
    # 添加边
    workflow.add_edge("agent", "tools")
    workflow.add_edge("tools", "agent")
    
    return workflow.compile()

def call_agent(state: AgentState):
    """Agent节点函数"""
    messages = state["messages"]
    
    # 创建提示
    prompt = """
    你是一个强大的AI助手，可以使用以下工具：
    1. web_search: 网络搜索
    2. calculator: 数学计算
    
    根据用户需求选择合适的工具并调用。
    """
    
    # 模拟AI响应
    if len(messages) == 0:
        return {"messages": [AIMessage(content="你好！我可以帮你进行网络搜索和数学计算。请告诉我需要什么帮助？")]}
    
    # 简单的Agent逻辑
    user_msg = messages[-1]
    if isinstance(user_msg, HumanMessage):
        content = user_msg.content.lower()
        
        if "搜索" in content or "search" in content:
            return {"messages": [AIMessage(content="需要使用web_search工具。")]}
        elif "计算" in content or "calculate" in content or "math" in content:
            return {"messages": [AIMessage(content="需要使用calculator工具。")]}
        else:
            return {"messages": [AIMessage(content="我需要更多信息来帮助你。")]}}
    
    return {"messages": [AIMessage(content="我理解了你的需求。")]}

# 创建Agent
agent = create_agent_with_tools()

# 测试
def test_basic_tool_usage():
    """测试基础工具使用"""
    print("=== 测试基础工具注册 ===")
    
    # 测试网络搜索
    result = web_search("LangGraph教程", max_results=5)
    print(f"网络搜索结果: {result}")
    
    # 测试数学计算
    result = calculator("2 + 2 * 3")
    print(f"数学计算结果: {result}")
    
    # 测试错误处理
    result = calculator("2 / 0")
    print(f"错误处理: {result}")

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

步骤2：动态工具选择系统

构建一个能够根据任务需求动态选择工具的系统：

from enum import Enum
from typing import List, Callable
import json
from dataclasses import dataclass

class ToolCategory(Enum):
    """工具分类"""
    SEARCH = "search"
    CALCULATION = "calculation"
    DATA_PROCESSING = "data_processing"
    COMMUNICATION = "communication"
    UTILITY = "utility"

@dataclass
class ToolMetadata:
    """工具元数据"""
    name: str
    description: str
    category: ToolCategory
    difficulty_level: str
    response_time: float
    success_rate: float

class DynamicToolSelector:
    """动态工具选择器"""
    
    def __init__(self):
        self.tools: Dict[str, Any] = {}
        self.tool_metadata: Dict[str, ToolMetadata] = {}
        
    def register_tool(self, tool: Any, metadata: ToolMetadata):
        """注册工具"""
        self.tools[tool.name] = tool
        self.tool_metadata[tool.name] = metadata
        
    def get_tools_by_category(self, category: ToolCategory) -> List[str]:
        """按类别获取工具"""
        return [name for name, meta in self.tool_metadata.items() 
                if meta.category == category]
    
    def select_best_tool(self, task: str, available_tools: List[str]) -> str:
        """根据任务选择最佳工具"""
        task_lower = task.lower()
        
        # 简单的匹配逻辑
        if any(keyword in task_lower for keyword in ["搜索", "search", "find"]):
            category_tools = self.get_tools_by_category(ToolCategory.SEARCH)
            return category_tools[0] if category_tools else None
            
        elif any(keyword in task_lower for keyword in ["计算", "calculate", "math"]):
            category_tools = self.get_tools_by_category(ToolCategory.CALCULATION)
            return category_tools[0] if category_tools else None
            
        elif any(keyword in task_lower for keyword in ["处理", "process", "analysis"]):
            category_tools = self.get_tools_by_category(ToolCategory.DATA_PROCESSING)
            return category_tools[0] if category_tools else None
            
        # 默认返回第一个可用工具
        return available_tools[0] if available_tools else None
    
    def get_tool_info(self, tool_name: str) -> ToolMetadata:
        """获取工具信息"""
        return self.tool_metadata.get(tool_name)

# 创建动态工具选择系统
selector = DynamicToolSelector()

# 注册工具
selector.register_tool(
    web_search, 
    ToolMetadata(
        name="web_search",
        description="网络搜索工具",
        category=ToolCategory.SEARCH,
        difficulty_level="简单",
        response_time=2.0,
        success_rate=0.95
    )
)

selector.register_tool(
    calculator,
    ToolMetadata(
        name="calculator",
        description="数学计算工具",
        category=ToolCategory.CALCULATION,
        difficulty_level="简单",
        response_time=0.5,
        success_rate=0.99
    )
)

def test_dynamic_selection():
    """测试动态工具选择"""
    print("\n=== 测试动态工具选择 ===")
    
    # 测试不同任务
    tasks = [
        "搜索最新的AI技术发展",
        "计算 5 * 7 + 3 的结果",
        "分析销售数据",
        "发送邮件给客户"
    ]
    
    for task in tasks:
        selected_tool = selector.select_best_tool(task, list(selector.tools.keys()))
        if selected_tool:
            metadata = selector.get_tool_info(selected_tool)
            print(f"任务: {task}")
            print(f"选择工具: {selected_tool} ({metadata.description})")
            print(f"难度: {metadata.difficulty_level}, 响应时间: {metadata.response_time}s")
            print("---")
        else:
            print(f"任务: {task} - 没有找到合适的工具")

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