LangGraph工具系统：构建智能代理的核心模块

1. LangGraph工具系统概述

LangGraph作为LangChain生态中的工作流编排框架，其工具系统是构建智能代理的核心模块。工具（Tool）在这里特指可被语言模型调用的功能单元，它们将自然语言指令转化为具体操作，实现从认知到执行的闭环。与普通API调用不同，LangGraph工具系统具备三大特征：

• 状态感知：工具能访问会话上下文、短期记忆和持久化存储
• 执行控制：支持直接返回、结构化输出和状态修改等多种响应模式
• 动态编排：可根据运行时条件灵活调整工具集

典型的工具应用场景包括：

• 数据查询（如天气API、数据库查询）
• 系统操作（文件读写、服务启停）
• 数学计算（单位换算、公式求解）
• 业务处理（订单查询、支付操作）

2. 基础工具定义与使用

2.1 最小化工具示例

最简单的工具只需用@tool装饰器标记函数：

python复制from langchain.tools import tool

@tool
def get_time() -> str:
    """返回当前服务器时间"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

关键要素说明：

1. 函数文档字符串（docstring）作为工具描述，会被语言模型用于理解功能
2. 返回类型声明决定输出处理方式（字符串将直接展示，字典会结构化解析）
3. 工具名称默认使用函数名，可通过@tool("自定义名称")覆盖

2.2 参数化工具开发

实际工具通常需要参数输入，LangGraph支持两种参数定义方式：

方式一：Pydantic模型验证

python复制from pydantic import BaseModel, Field
from langchain.tools import tool

class CalculatorInput(BaseModel):
    a: float = Field(description="第一个运算数")
    b: float = Field(description="第二个运算数")
    op: str = Field(description="运算符", enum=["+", "-", "*", "/"])

@tool(args_schema=CalculatorInput)
def calculate(a: float, b: float, op: str) -> float:
    """执行基础四则运算"""
    if op == "+": return a + b
    elif op == "-": return a - b
    elif op == "*": return a * b
    elif op == "/": return a / b
    else: raise ValueError("不支持的运算符")

方式二：JSON Schema直接定义

python复制from langchain.tools import tool

calculator_schema = {
    "type": "object",
    "properties": {
        "a": {"type": "number"},
        "b": {"type": "number"},
        "op": {"type": "string", "enum": ["+", "-", "*", "/"]}
    },
    "required": ["a", "b", "op"]
}

@tool(args_schema=calculator_schema)
def calculate(a: float, b: float, op: str) -> float:
    """执行基础四则运算"""
    # 实现同上...

参数设计注意事项：

• 每个参数必须包含类型和描述信息
• 使用enum限定可选值范围能显著提升模型调用准确率
• 复杂参数建议拆分为多个简单工具，避免模型理解困难

3. 运行时上下文访问

3.1 状态管理基础

LangGraph工具可通过ToolRuntime访问三类运行时数据：

3.2 状态读写实战

读取会话状态示例：

python复制from langchain.tools import tool, ToolRuntime
from langchain.messages import HumanMessage

@tool
def summarize_chat(runtime: ToolRuntime) -> str:
    """总结当前会话的讨论要点"""
    messages = runtime.state["messages"]
    user_msgs = [m.content for m in messages if isinstance(m, HumanMessage)]
    return f"已讨论{len(user_msgs)}个用户问题，最近问题：{user_msgs[-1][:50]}..."

修改状态示例：

python复制from langchain.tools import tool, ToolRuntime
from langgraph.types import Command
from langchain.messages import ToolMessage

@tool
def set_theme(theme: str, runtime: ToolRuntime) -> Command:
    """设置界面主题颜色"""
    return Command(
        update={
            "user_settings": {"theme": theme},
            "messages": [
                ToolMessage(
                    content=f"主题已切换为{theme}",
                    tool_call_id=runtime.tool_call_id
                )
            ]
        }
    )

状态更新最佳实践：

1. 通过Command返回状态修改指令
2. 包含ToolMessage让模型感知操作结果
3. 对可能并发修改的字段定义reducer函数

4. 高级工具模式

4.1 多模态工具开发

现代语言模型支持处理图文混排内容，工具可以返回复合型结果：

python复制from langchain.tools import tool

@tool
def get_product_info(product_id: str) -> list:
    """获取商品详情（含图片）"""
    return [
        {"type": "text", "text": f"商品ID：{product_id}"},
        {"type": "text", "text": "价格：¥299.00"},
        {"type": "image", "url": f"https://cdn.com/{product_id}.jpg"},
        {"type": "text", "text": "库存状态：有货"}
    ]

支持的多媒体类型包括：

• text：纯文本内容
• image：图片URL（需模型支持图片理解）
• audio：音频URL（需模型支持语音处理）
• video：视频URL（需模型支持视频理解）

4.2 动态工具注册

通过中间件实现运行时工具动态加载：

python复制from langchain.tools import tool
from langchain.agents.middleware import AgentMiddleware

class PluginToolsMiddleware(AgentMiddleware):
    def __init__(self, plugin_dir):
        self.plugin_dir = plugin_dir
        self.loaded_tools = {}
        
    def wrap_model_call(self, request, handler):
        # 按需加载插件工具
        if "use_plugins" in request.runtime.state:
            self._load_plugins()
            tools = list(request.tools) + list(self.loaded_tools.values())
            return handler(request.override(tools=tools))
        return handler(request)
    
    def _load_plugins(self):
        # 实际项目应从插件目录动态加载
        if not self.loaded_tools:
            @tool
            def plugin_demo(query: str) -> str:
                """示例插件工具"""
                return f"处理结果: {query.upper()}"
            self.loaded_tools["plugin_demo"] = plugin_demo

5. 生产环境实践

5.1 错误处理规范

健壮的工具实现应包含错误处理逻辑：

python复制from langchain.tools import tool
from langchain.messages import ToolMessage

@tool
def safe_divide(a: float, b: float) -> float:
    """安全除法运算"""
    try:
        return a / b
    except ZeroDivisionError:
        return ToolMessage(
            content="除数不能为零",
            tool_call_id=runtime.tool_call_id,
            is_error=True
        )

推荐错误处理策略：

1. 使用ToolMessage返回结构化错误信息
2. 为可重试错误设置retry_policy
3. 记录错误日志到监控系统

5.2 性能优化技巧

工具冷启动优化：

python复制from langchain.tools import tool
from functools import lru_cache

@lru_cache(maxsize=100)
@tool
def heavy_computation(x: float) -> float:
    """耗时计算函数"""
    import time
    time.sleep(3)  # 模拟复杂计算
    return x ** 2

异步工具实现：

python复制from langchain.tools import tool
import aiohttp

@tool
async def fetch_webpage(url: str) -> str:
    """异步获取网页内容"""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

6. 调试与测试

6.1 单元测试方案

使用unittest进行工具测试：

python复制import unittest
from your_module import calculate

class TestTools(unittest.TestCase):
    def test_calculate(self):
        self.assertEqual(calculate(2, 3, "+"), 5)
        self.assertEqual(calculate(5, 2, "-"), 3)
        with self.assertRaises(ValueError):
            calculate(1, 1, "x")  # 测试异常输入

6.2 集成测试方案

模拟完整调用链路：

python复制from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

def test_tool_integration():
    agent = create_agent(
        ChatOpenAI(model="gpt-4"),
        tools=[calculate],
    )
    response = agent.invoke({
        "messages": [{
            "role": "user", 
            "content": "请计算3.14乘以100"
        }]
    })
    assert "314" in response["output"]

测试要点检查表：

• [ ] 参数验证是否生效
• [ ] 错误处理是否合规
• [ ] 性能是否达标
• [ ] 状态修改是否符合预期
• [ ] 多工具协同是否正常

7. 安全规范

7.1 输入验证原则

所有工具必须实现严格的输入验证：

python复制from langchain.tools import tool
from pydantic import BaseModel, validator

class SafeInput(BaseModel):
    query: str
    
    @validator('query')
    def check_query(cls, v):
        if len(v) > 1000:
            raise ValueError("输入过长")
        if "<script>" in v.lower():
            raise ValueError("检测到危险输入")
        return v

@tool(args_schema=SafeInput)
def safe_search(query: str) -> str:
    """安全搜索工具"""
    # 实现逻辑...

7.2 权限控制方案

基于角色的工具访问控制：

python复制from langchain.agents.middleware import wrap_model_call
from dataclasses import dataclass

@dataclass
class UserContext:
    role: str  # admin/user/guest

@wrap_model_call
def role_based_tools(request, handler):
    if request.runtime.context.role == "guest":
        request = request.override(tools=[
            t for t in request.tools 
            if not getattr(t, "requires_auth", False)
        ])
    return handler(request)

8. 性能监控

8.1 指标收集方案

使用装饰器实现工具指标收集：

python复制from langchain.tools import tool
import time
import statsd

metrics = statsd.StatsClient('localhost', 8125)

def track_metrics(func):
    def wrapper(*args, **kwargs):
        start = time.time()
        try:
            result = func(*args, **kwargs)
            metrics.timing(f"tools.{func.__name__}.success", (time.time()-start)*1000)
            return result
        except Exception as e:
            metrics.incr(f"tools.{func.__name__}.error")
            raise
    return wrapper

@track_metrics
@tool
def monitored_tool():
    """带监控的工具"""
    # 业务逻辑...

8.2 关键监控指标

9. 版本管理策略

9.1 语义化版本控制

工具版本号规范：

• 主版本号：不兼容的API修改
• 次版本号：向下兼容的功能新增
• 修订号：问题修正

python复制@tool(version="1.2.0")
def versioned_tool():
    """支持版本控制的工具"""

9.2 多版本共存方案

通过名称后缀实现多版本共存：

python复制@tool("weather-v1")
def weather_v1(city: str) -> str:
    """天气查询V1"""

@tool("weather-v2") 
def weather_v2(city: str, days: int = 1) -> str:
    """天气查询V2（支持多天预报）"""

10. 文档与注释规范

10.1 文档标准模板

python复制@tool
def well_documented_tool(param1: str, param2: int) -> dict:
    """
    工具功能简述（不超过50字）
    
    详细功能描述：
    - 功能点1说明
    - 功能点2说明
    
    参数说明：
    param1: 参数用途（类型约束）
            可选值说明
    param2: 参数用途（取值范围）
    
    返回结构：
    {
        "field1": "说明",
        "field2": "说明"
    }
    
    示例：
    >>> well_documented_tool("test", 5)
    {"field1": "result", "field2": 10}
    
    错误码：
    400: 参数无效
    500: 服务不可用
    """

10.2 文档生成流程

使用pdoc自动生成文档：

bash复制# 安装文档工具
pip install pdoc3

# 生成HTML文档
pdoc --html your_module --output-dir docs

文档应包括：

• 工具列表及功能概述
• 每个工具的详细API说明
• 调用示例和返回示例
• 错误处理指南
• 性能特征说明