5.工具开发实战

工具是 LangGraph 代理系统与外部世界交互的桥梁。一个设计良好的工具能让模型准确理解其用途、正确构造参数，并在出错时优雅处理。本章将深入探讨如何从零开始构建可靠的工具系统，涵盖自定义函数封装、执行节点集成、动态选择策略以及错误处理机制。

自定义工具函数开发

在 LangGraph 中定义工具最直观的方式是使用装饰器模式。这种方式能将普通 Python 函数无缝转换为模型可调用的工具，关键在于让模型理解函数的意图、参数格式和返回值结构。

从普通函数到工具

将一个普通函数转换为工具，核心是为其添加元数据描述。模型依赖这些描述来决定是否调用工具以及如何构造参数。下面是一个基础乘法工具的实现：

from langchain_core.tools import tool

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

这段代码看起来简单，但背后完成了几件重要的事。装饰器自动提取函数名作为工具标识，将 docstring 作为工具描述，并通过类型注解生成参数模式。当模型看到用户询问"三乘以七是多少"时，能识别出这是数学运算，理解 multiply 工具的作用，并正确构造 {"a": 3, "b": 7} 这样的参数。

对于 JavaScript 环境，工具定义稍显显式，需要手动指定模式：

import { tool } from "@langchain/core/tools";
import { z } from "zod";

const multiply = tool(
  (input) => {
    return input.a * input.b;
  },
  {
    name: "multiply",
    description: "Multiply two numbers.",
    schema: z.object({
      a: z.number().describe("First operand"),
      b: z.number().describe("Second operand"),
    }),
  }
);

这种定义方式的优势在于完全控制工具的行为。函数体专注于业务逻辑，而元数据部分清晰分离，便于维护。工具一旦定义，就可以通过 invoke 方法直接调用，也能绑定到语言模型上。

工具的描述与参数定义

模型对工具的理解完全依赖描述文本。模糊的描述会导致模型误用工具或构造错误参数。来看一个更完整的例子：

from langchain_core.tools import tool

@tool("weather_query", parse_docstring=True)
def get_weather(location: str, unit: str = "celsius") -> dict:
    """Get current weather conditions for a specified location.
    
    Args:
        location: City name or coordinates (e.g., "San Francisco" or "37.7749,-122.4194")
        unit: Temperature unit, either "celsius" or "fahrenheit"
    """
    # 实际实现会调用天气API
    return {
        "temperature": 22,
        "condition": "sunny",
        "humidity": 45
    }

这里使用了 parse_docstring=True 参数，装饰器会自动解析 docstring 中的 Args 部分，为每个参数生成描述。模型因此知道 location 可以是城市名或坐标，unit 只有两种合法值。这种细粒度的描述大幅提升了调用准确率。

对于复杂参数，可以使用 Pydantic 模型显式定义模式：

from pydantic import BaseModel, Field
from langchain_core.tools import tool

class WeatherInput(BaseModel):
    """Input schema for weather query"""
    location: str = Field(description="City name or lat,lon coordinates")
    unit: str = Field(default="celsius", description="Temperature unit")

@tool("advanced_weather", args_schema=WeatherInput)
def get_weather_detailed(location: str, unit: str = "celsius") -> dict:
    return get_weather(location, unit)

这种方式适合参数结构复杂或需要严格验证的场景。Pydantic 提供了类型检查、默认值处理、描述文本等完整功能，工具定义因此更加健壮。

工具返回值处理

工具执行后返回的数据会成为模型继续推理的上下文。返回值格式直接影响模型的理解。简单场景下，返回字符串或数字即可。但对于复杂数据，建议返回结构化信息：

@tool
def search_products(category: str, max_price: float) -> list:
    """Search for products in a category under a price limit."""
    # 模拟数据库查询
    results = [
        {"name": "Wireless Mouse", "price": 29.99},
        {"name": "USB-C Hub", "price": 49.99}
    ]
    return results

模型收到列表后，能自然理解每个元素的属性。如果返回的是纯文本，模型需要额外解析，容易出错。工具返回的数据最终会被包装成 ToolMessage，插入到消息流中。这个 Message 包含工具名称、调用 ID 和返回值，构成了完整的调用轨迹。

ToolNode 工具执行器

定义工具只是第一步，如何在图结构中执行工具调用是另一个关键问题。ToolNode 是 LangGraph 提供的预构建节点，专门处理工具执行逻辑，支持并发调用、错误处理和异步操作。

ToolNode的核心职责

ToolNode 的设计哲学是"只做一件事，但做到极致"。它接收包含 tool_calls 的 AIMessage，提取所有调用请求，并行执行，然后返回 ToolMessage 列表。基本用法如下：

from langgraph.prebuilt import ToolNode

def get_weather(location: str):
    """Call to get the current weather."""
    if location.lower() in ["sf", "san francisco"]:
        return "It's 60 degrees and foggy."
    else:
        return "It's 90 degrees and sunny."

def get_coolest_cities():
    """Get a list of coolest cities"""
    return "nyc, sf"

tool_node = ToolNode([get_weather, get_coolest_cities])

这个节点会自动识别消息中的工具调用，匹配函数名，执行对应函数。如果模型在一次响应中请求了多个工具（例如同时查询天气和城市列表），ToolNode 会并行执行这些调用，显著降低总耗时。

执行时，只需将包含工具调用的消息传递给节点：

from langchain_core.messages import AIMessage

tool_calls = [
    {"name": "get_weather", "args": {"location": "sf"}, "id": "1"},
    {"name": "get_coolest_cities", "args": {}, "id": "2"}
]
ai_message = AIMessage(content="", tool_calls=tool_calls)

result = tool_node.invoke({"messages": [ai_message]})

返回结果包含两个 ToolMessage，顺序与调用顺序一致。每个 ToolMessage 记录了工具名称、调用 ID 和返回值，模型能准确识别哪个结果对应哪个请求。

在图结构中集成ToolNode

在完整的代理工作流中，ToolNode 通常与条件边配合使用。模型生成响应后，系统检查是否包含工具调用，如果有则路由到 ToolNode，执行完再回到模型进行下一步推理：

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode, tools_condition

def model_node(state: MessagesState):
    response = model_with_tools.invoke(state["messages"])
    return {"messages": [response]}

tool_node = ToolNode([get_weather, get_coolest_cities])

builder = StateGraph(MessagesState)
builder.add_node("model", model_node)
builder.add_node("tools", tool_node)

builder.add_edge(START, "model")
builder.add_conditional_edges(
    "model",
    tools_condition,  # 检查最后一条消息是否有tool_calls
    ["tools", END]      # 有则去tools，没有则结束
)
builder.add_edge("tools", "model")  # 工具执行完返回模型

graph = builder.compile()

tools_condition 是预置的条件函数，自动检查最后一条消息。这种设计让工作流极其简洁，开发者无需手动解析消息内容。图的循环结构自然形成了 ReAct 模式：模型思考、调用工具、观察结果、再思考，直到问题解决。

并行工具执行机制

现代 LLM 支持在一次响应中调用多个工具。ToolNode 的并行执行能力在此场景下价值凸显。假设用户问"比较纽约和旧金山的天气，并告诉我哪个城市更凉爽"，模型可能生成两个天气查询调用：

tool_calls = [
    {"name": "get_weather", "args": {"location": "nyc"}, "id": "1"},
    {"name": "get_weather", "args": {"location": "sf"}, "id": "2"}
]

ToolNode 会同时发起两个 API 请求（如果是真实网络调用），总耗时接近单次调用，而非顺序执行的总和。执行完成后，返回两个 ToolMessage，模型基于两者结果进行综合判断。

这种并行能力对工具数量多的场景尤为重要。代理可能需要同时查询数据库、调用计算器、搜索网络，ToolNode 确保这些操作不会相互阻塞。需要注意的是，并行执行意味着工具函数应该是无状态的，或者能安全处理并发调用。如果工具共享全局资源，需要自行实现同步机制。

运行时动态工具选择

静态工具绑定适合简单场景，但生产环境中，工具可用性往往取决于运行时上下文。用户权限、数据范围、业务规则都可能影响工具选择。LangGraph 支持在运行时动态决定绑定哪些工具。

静态工具绑定的局限性

常规做法是在创建模型时绑定工具：

model_with_tools = model.bind_tools([weather_tool, calculator, search_tool])

这种方式的问题在于工具列表固定。假设有敏感的数据删除工具，只有管理员能用，普通用户不应该访问。静态绑定无法区分用户身份。或者工具数量极多（上百个），全部绑定会消耗大量 token，且干扰模型选择。

基于上下文的工具筛选

解决方案是在运行时根据上下文筛选工具。create_react_agent 支持传入一个配置函数，该函数在每次调用时执行，返回绑定好工具的模型：

from dataclasses import dataclass
from typing import Literal
from langgraph.prebuilt import create_react_agent
from langgraph.runtime import Runtime

@dataclass
class CustomContext:
    user_role: Literal["admin", "user"]
    available_tools: list[Literal["weather", "calculator", "data_delete"]]

def configure_model(state, runtime: Runtime[CustomContext]):
    """根据运行时上下文动态绑定工具"""
    all_tools = {
        "weather": weather_tool,
        "calculator": calculator,
        "data_delete": data_delete_tool
    }
    
    # 根据用户角色筛选工具
    allowed_tools = []
    for tool_name in runtime.context.available_tools:
        if tool_name == "data_delete" and runtime.context.user_role != "admin":
            continue  # 非管理员不能使用删除工具
        allowed_tools.append(all_tools[tool_name])
    
    return model.bind_tools(allowed_tools)

agent = create_react_agent(
    configure_model,  # 传入配置函数而非固定模型
    tools=[weather_tool, calculator, data_delete_tool]  # 提供所有工具用于初始化
)

调用时代入上下文：

agent.invoke(
    {"messages": [{"role": "user", "content": "delete all data"}]},
    context=CustomContext(user_role="user", available_tools=["data_delete"])
)

虽然工具列表包含 data_delete，但配置函数会过滤掉它，模型无法调用。这种方式将权限逻辑与业务逻辑解耦，代理本身无需关心权限，只需专注于任务执行。

动态绑定工具到模型

另一种模式是根据用户查询动态选择工具。当工具数量庞大时，全部绑定不现实。可以先通过语义搜索筛选相关工具，再绑定到模型：

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS

# 假设有100+工具，预先建立向量索引
all_tools = [tool1, tool2, ..., tool100]
tool_descriptions = [t.description for t in all_tools]
vectorstore = FAISS.from_texts(tool_descriptions, embeddings)

def dynamic_tool_binding(state):
    user_query = state["messages"][-1].content
    
    # 检索最相关的5个工具
    relevant_tools = vectorstore.similarity_search(user_query, k=5)
    selected_tools = [all_tools[i] for i in relevant_tools.indices]
    
    return model.bind_tools(selected_tools)

agent = create_react_agent(dynamic_tool_binding, tools=all_tools)

这种方法显著减少 token 消耗，提升模型选择准确率。LangGraph 社区提供了 langgraph-bigtool 库，封装了这类模式。动态选择也适用于多租户场景，每个租户有自己的工具集，运行时根据租户 ID 加载对应工具。

工具调用错误处理

工具执行难免出错：网络超时、参数无效、权限不足。健壮的错误处理机制能确保代理在遇到问题时优雅降级，而非直接崩溃。

默认错误处理行为

ToolNode 默认启用错误处理。工具抛出异常时，不会中断整个工作流，而是将错误信息包装成 ToolMessage，状态标记为 error：

@tool
def risky_operation(value: int) -> str:
    """一个可能失败的工具"""
    if value < 0:
        raise ValueError("Negative values are not allowed")
    return f"Processed {value}"

tool_node = ToolNode([risky_operation])

# 模拟模型调用时传入负数
tool_calls = [{"name": "risky_operation", "args": {"value": -5}, "id": "1"}]
ai_message = AIMessage(content="", tool_calls=tool_calls)

result = tool_node.invoke({"messages": [ai_message]})

返回的 ToolMessage 内容为 "Error: Negative values are not allowed\n Please fix your mistakes."，状态字段为 "error"。模型收到错误信息后，能理解问题所在，尝试修正参数再次调用。这种自愈能力是代理系统的重要特性。

自定义错误消息

默认错误信息包含堆栈跟踪，可能过于技术化。可以通过 handle_tool_errors 参数自定义：

tool_node = ToolNode(
    [risky_operation],
    handle_tool_errors="Value must be positive. Please check your input."
)

现在返回的错误信息更友好，模型能更好理解并指导用户。也可以传入函数动态生成错误消息：

def custom_error_handler(error: Exception, tool_name: str) -> str:
    if isinstance(error, ValueError):
        return f"Invalid input for {tool_name}: {str(error)}"
    return f"Unexpected error in {tool_name}: {str(error)}"

tool_node = ToolNode([risky_operation], handle_tool_errors=custom_error_handler)

这种方式让错误处理更灵活，可以根据异常类型提供针对性建议。

错误处理在代理中的应用

在 create_react_agent 中，错误处理通过 ToolNode 实现。要自定义错误行为，可以传入配置好的 ToolNode 实例：

from langgraph.prebuilt import create_react_agent, ToolNode

# 创建带自定义错误处理的工具节点
safe_tool_node = ToolNode(
    [risky_operation],
    handle_tool_errors="Please provide a positive number."
)

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=safe_tool_node  # 传入节点实例而非工具列表
)

# 测试错误场景
agent.invoke({"messages": [{"role": "user", "content": "process -10"}]})

代理会捕获错误，生成友好的提示，可能引导用户重新输入。对于严重错误，可以禁用自动处理，让异常向上传播：

strict_tool_node = ToolNode([risky_operation], handle_tool_errors=False)

此时异常会抛出到调用层，适合需要事务回滚或特殊清理逻辑的场景。错误处理策略应根据工具性质选择：对于非关键工具，宽容的错误处理更合适；对于金融交易等关键操作，可能需要严格失败。

工具开发是构建可靠代理的基石。从清晰的函数定义到健壮的执行节点，从灵活的动态选择到周到的错误处理，每个环节都影响着代理的最终表现。掌握这些技术后，我们就能构建出既强大又可靠的代理系统。

下一章将探讨记忆系统，这是让代理具备持续学习能力的关键。我们将深入研究短期记忆与长期记忆的实现机制，以及如何在工具中读写记忆，构建真正智能的对话系统。