LangGraph工具系统：构建智能代理工作流的核心技术

1. LangGraph工具系统概述

在LangGraph框架中，"工具"(Tool)是构建智能代理(Agent)工作流的核心组件。作为LangChain生态的重要扩展，LangGraph的工具系统提供了比传统LangChain更强大的状态管理和上下文访问能力。工具本质上是一个可执行函数，封装了特定功能逻辑，允许大语言模型(LLM)通过结构化方式与外部系统交互。

我最近在开发一个金融客服机器人时，深刻体会到合理设计工具的重要性。当用户询问"我的账户余额是多少"时，系统需要调用get_account_info工具，而这个工具必须能安全地访问用户上下文和数据库。LangGraph通过ToolRuntime机制，优雅地解决了工具执行时的状态管理难题。

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. @tool装饰器标记这是一个LangGraph工具
2. 清晰的文档字符串(docstring)说明工具功能
3. 明确的返回类型注解

提示：文档字符串的质量直接影响工具被调用的准确性。建议采用"动词+宾语"的句式，如"获取X"、"查询Y"、"设置Z"。

2.2 带参数的工具

实际业务场景中，工具通常需要参数输入：

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

class WeatherInput:
    location: str = Field(description="城市名称或坐标")
    units: str = Field(default="celsius", 
                      description="温度单位(celsius/fahrenheit)")
    
@tool(args_schema=WeatherInput)
def get_weather(location: str, units: str = "celsius") -> dict:
    """获取指定地点的天气信息"""
    # 实际项目中这里会调用天气API
    return {
        "location": location,
        "temperature": 22 if units == "celsius" else 72,
        "units": units,
        "conditions": "sunny"
    }

这个天气查询工具展示了：

• 使用Pydantic模型定义参数结构(WeatherInput)
• 每个字段都有详细的description
• 参数有默认值(units="celsius")
• 返回结构化数据而非纯文本

我在电商项目中就吃过亏：一个商品查询工具最初只返回文本，后来发现LLM很难从中提取具体属性，改为返回JSON结构后准确率提升了40%。

3. 高级工具特性

3.1 状态管理

LangGraph最强大的特性之一是工具可以访问工作流状态。这是通过ToolRuntime参数实现的：

python复制from langchain.tools import tool, ToolRuntime

@tool
def get_unread_count(runtime: ToolRuntime) -> int:
    """获取当前对话中的未读消息数"""
    messages = runtime.state["messages"]
    return sum(1 for msg in messages if not msg.get("read", False))

关键点：

• runtime参数由系统自动注入，不会暴露给LLM
• 通过runtime.state访问工作流状态
• 状态是短期的，仅存在于当前对话

在客服系统中，我用这个特性实现了"显示未读消息"功能。相比传统会话管理，LangGraph的状态机制更加直观。

3.2 上下文访问

工具可以获取调用时的上下文信息：

python复制from dataclasses import dataclass
from langchain.tools import tool, ToolRuntime

@dataclass
class UserContext:
    user_id: str
    privilege: str = "basic"

@tool
def get_user_profile(runtime: ToolRuntime[UserContext]) -> str:
    """获取当前用户资料"""
    if runtime.context.privilege == "admin":
        return "Full profile data"
    else:
        return "Basic profile data"

上下文与状态的区别：

• 上下文(context): 调用时传入的不可变数据
• 状态(state): 工作流运行期间可变的共享数据

3.3 持久化存储

对于需要长期保存的数据，可以使用Store：

python复制from langchain.tools import tool, ToolRuntime

@tool
def save_preference(key: str, value: str, runtime: ToolRuntime) -> str:
    """保存用户偏好设置"""
    runtime.store.put(("preferences",), key, value)
    return "Preference saved"

@tool 
def load_preference(key: str, runtime: ToolRuntime) -> str:
    """读取用户偏好设置"""
    item = runtime.store.get(("preferences",), key)
    return item.value if item else "Not found"

Store使用命名空间+键的存储模式，适合保存用户配置、历史记录等持久化数据。

4. 工具执行控制

4.1 返回值处理

LangGraph支持多种返回值形式：

python复制from langchain.messages import ToolMessage
from langgraph.types import Command

# 返回纯文本
@tool
def get_time() -> str:
    """返回当前时间"""
    from datetime import datetime
    return datetime.now().strftime("%H:%M")

# 返回结构化数据  
@tool
def get_user() -> dict:
    """返回用户信息"""
    return {"name": "Alice", "age": 30}

# 更新状态并返回消息
@tool
def set_language(lang: str, runtime: ToolRuntime) -> Command:
    """设置界面语言"""
    return Command(
        update={"language": lang},
        messages=[ToolMessage(content=f"Language set to {lang}")]
    )

4.2 直接返回模式

设置return_direct=True可以让工具结果直接返回给用户，跳过LLM处理：

python复制@tool(return_direct=True)
def get_balance(account: str) -> str:
    """获取账户余额"""
    return f"账户 {account} 余额：$1000"

适用场景：

• 结果已经是用户友好的格式
• 不需要LLM进一步处理
• 需要确保输出不被修改

4.3 错误处理

健壮的工具应该包含错误处理：

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

@tool
def divide_numbers(a: float, b: float) -> float:
    """两数相除"""
    try:
        return a / b
    except ZeroDivisionError:
        return ToolMessage(content="错误：除数不能为零")

更复杂的错误处理可以通过中间件实现：

python复制from langchain.agents.middleware import wrap_tool_call

@wrap_tool_call
def error_handler(request, next):
    try:
        return next(request)
    except Exception as e:
        return ToolMessage(
            content=f"工具执行失败：{str(e)}",
            tool_call_id=request.tool_call["id"]
        )

5. 动态工具管理

5.1 基于状态的工具过滤

可以根据工作流状态动态显示/隐藏工具：

python复制from langchain.agents.middleware import wrap_model_call

@wrap_model_call
def filter_tools(request, next):
    if not request.state.get("authenticated"):
        # 未认证用户只能使用基础工具
        request.tools = [t for t in request.tools if t.name.startswith("public_")]
    return next(request)

5.2 运行时工具注册

更灵活的方式是动态注册工具：

python复制class DynamicToolMiddleware:
    def wrap_model_call(self, request, next):
        if "需要计算" in request.messages[-1].content:
            request.tools.append(create_calculator_tool())
        return next(request)
    
    def wrap_tool_call(self, request, next):
        if request.tool_call["name"] == "dynamic_calc":
            return handle_dynamic_calc(request)
        return next(request)

6. 实战经验分享

在开发客服机器人过程中，我总结了以下最佳实践：

1. 命名规范：工具名使用"动词_名词"格式，如query_balance、update_profile

2. 参数设计：

   • 必填参数放在前面
   • 每个参数必须有详细描述
   • 使用枚举类型限制选项

3. 权限控制：

python复制@tool
def delete_account(runtime: ToolRuntime) -> str:
    """删除当前账户(需要管理员权限)"""
    if runtime.context.role != "admin":
        return "权限不足"
    # 执行删除逻辑

4. 性能优化：

   • 耗时工具实现进度反馈
   • 使用缓存减少重复计算
   • 批量处理替代频繁调用

5. 调试技巧：

   • 记录工具调用日志
   • 添加@tool(verbose=True)查看详细执行信息
   • 使用Mock工具进行测试

一个综合性的电商工具示例：

python复制from typing import Literal
from pydantic import Field
from langchain.tools import tool, ToolRuntime

class ProductQuery:
    category: str = Field(description="商品类别")
    sort_by: Literal["price", "sales", "rating"] = Field(
        default="price",
        description="排序方式"
    )
    limit: int = Field(
        default=5,
        description="返回结果数量",
        gt=0,
        le=20
    )

@tool(args_schema=ProductQuery)
def search_products(
    category: str,
    sort_by: str = "price",
    limit: int = 5,
    runtime: ToolRuntime = None
) -> list[dict]:
    """搜索指定类别的商品"""
    # 实际项目会查询数据库
    products = [
        {"name": f"{category}商品{i}", "price": 100-i*10}
        for i in range(1, limit+1)
    ]
    
    # 记录搜索历史
    if runtime:
        runtime.store.append(
            ("history", "searches"),
            {"category": category, "time": datetime.now()}
        )
    
    return sorted(products, key=lambda x: x[sort_by])

7. 常见问题排查

问题1：工具未被调用

• 检查工具名称是否明确
• 验证文档字符串是否清晰
• 确认参数schema定义正确

问题2：状态更新不生效

• 确保使用了Command返回值
• 检查state字段是否在graph定义中声明
• 验证reducer函数是否正确处理并发更新

问题3：上下文获取为None

• 检查调用时是否传入了context
• 确认context_schema与实际类型匹配
• 验证用户认证流程是否正确

问题4：Store数据不持久

• 确认使用的是持久化Store实现(如PostgresStore)
• 检查命名空间和键的组合是否正确
• 验证是否有足够的写入权限

性能优化案例：
在实现一个天气查询工具时，最初每次调用都直接访问外部API，导致响应慢且容易超时。后来引入本地缓存，将结果缓存10分钟，并添加重试机制，使成功率从75%提升到99%，平均响应时间从1.2秒降到0.3秒。