工具在 LangChain 中的作用

在 LangChain 中，工具的设计遵循特定的输入/输出契约，使用 Pydantic v2 模型来定义参数和返回结构，确保了数据的一致性和可靠性。这种设计思路强调 RORO（接收一个对象，返回一个对象）原则。

工具的执行体可以是纯函数或者更复杂的逻辑处理，内部可能涉及 API 调用、数据库操作、MCP 等实际业务逻辑的实现。

async def

工具元数据包含了名称、描述、类型、超时时间和并发限制等关键信息。这些元数据帮助 LLM 根据描述和 schema 来决定何时以及如何调用工具。

通过特定的组件，LangChain 能够将上述各个部分组合成可复用的动作单元，无论是集成到 Agent、Router 还是 LCEL Flow 中，都能保持接口的一致性。

Tool

BaseTool

核心属性与类型

属性	含义	实践要点
工具唯一标识	使用动词开头的标识符	确保每个工具都有一个清晰且唯一的标识
功能说明	提供给 LLM 的参考信息	详细说明输入约束、单位和可能的失败模式
Pydantic BaseModel 输入	用于类型检查和默认值设置	确保所有输入都经过统一的类型验证和默认值处理
直接返回给用户的输出	确定工具输出是否直接展示给用户	在没有 Agent 的场景中，此选项通常设为 True
异步/同步执行入口	定义工具的执行方式	对于 I/O 密集型任务，建议使用异步执行
自定义异常处理	处理工具执行过程中可能出现的异常	将异常转换为用户友好的错误消息
过滤和监控	用于路由器或审计目的	实现工具的过滤和监控功能，支持更精细的管理和分析

name

query_weather

description

args_schema

return_direct

coroutine

func

handle_tool_error

tags

metadata

常见的工具类型包括：

• 基础工具：最简单的封装形式，适用于同步函数。
• 异步工具：具有原生异步入口，适合处理异步任务。
• 带参数校验的工具：强制使用 args_schema，方便处理复杂的参数需求。
• 自定义工具：允许完全自定义行为，例如实现缓存、批处理和跟踪等功能。

Tool

AsyncTool

StructuredTool

BaseTool

三种创建工具的方式

3.1 装饰器方式（最轻量）

from langchain.agents import tool
@tool("get_exchange_rate", return_direct=True)
def get_exchange_rate(pair: str) -> str:
    """输入类似 'USD/CNY'，返回六位小数的即期汇率。"""
    if "/" not in pair:
        raise ValueError("货币对格式应为 XXX/YYY")
    base, quote = pair.upper().split("/")
    rate = fetch_rate(base=base, quote=quote)  # 同步 HTTP
    return f"{base}/{quote}={rate:.6f}"

特点：自动推断单个参数的类型，非常适合快速开发演示。

@tool

3.2 结构化工具（推荐默认）

from langchain.tools import StructuredTool
from pydantic import BaseModel, Field

class WeatherInput(BaseModel):
    city: str = Field(..., description="城市英文名，如 beijing")
    is_metric: bool = Field(default=True, description="是否使用摄氏度")

async def fetch_weather(city: str, is_metric: bool) -> dict:
    units = "metric" if is_metric else "imperial"
    return await weather_client.query(city=city, units=units)

weather_tool = StructuredTool.from_function(
    name="query_weather",
    description="查询城市实时天气，失败返回错误信息",
    func=fetch_weather,
    coroutine=fetch_weather,  # 支持异步
    args_schema=WeatherInput,
)

特点：显式的 schema 定义，支持同时注册同步和异步方法，适应多种应用场景。

StructuredTool.from_function

func

coroutine

3.3 手动子类化（高级可控）

from langchain.tools import BaseTool
from pydantic import BaseModel

class StockInput(BaseModel):
    symbol: str

class StockTool(BaseTool):
    name = "query_stock"
    description = "获取股票实时价，示例：symbol=TSLA"
    args_schema = StockInput

    def _run(self, symbol: str) -> str:
        price = sync_client.get_price(symbol)
        return f"{symbol} 当前价格 {price}"

BaseTool

异步方法 _arun 定义如下：

async def _arun(self, symbol: str) -> str:
    price = await async_client.get_price(symbol)
    return f"{symbol} 当前价格 {price}"

创建了一个 StockTool 实例。

特点包括但不限于：支持扩展属性、在功能模块内加入缓存、熔断机制及链路追踪等自定义逻辑。

_run/_arun

同步与异步调用路径

当 Agent/LCEL 在同步环境中运行时，它会调用相应的同步函数，最终触发一系列操作，这些操作可能包括但不限于以下几种情况：

Tool.__call__()

最终可能触发的操作有：

_run

或者

func

此流程适用于脚本、后台任务或 FastAPI 的同步端点。

而在异步事件循环中（例如 FastAPI

async def

、LangServe、Jupyter asyncio），Agent 将运行异步代码，使用特定的方法来执行任务：

await tool.arun(...)

这将导致执行

_arun

或

coroutine

。

在混合场景下，用户可以通过同步方式

tool.invoke()

和异步方式

await tool.ainvoke()

手动控制调用过程，例如：

result_sync = weather_tool.invoke({"city": "beijing"})
    result_async = await weather_tool.ainvoke({"city": "shanghai"})

最佳实践

建议对外部公开

func

与

coroutine

，同时在内部实现中保持逻辑的一致性，以便于维护和测试。

工具异常处理

为了确保输入的有效性，可以利用 Pydantic 进行提前验证，若验证失败，则直接抛出异常，Agent 将接收到结构化的错误提示。

field_validator

ValidationError

对于预期中的异常情况，可以通过自定义错误信息来提高用户体验：

from langchain.tools import ToolException
    async def safe_fetch_weather(city: str, is_metric: bool) -> dict:
        if not city:
            raise ToolException("city 不能为空")
        try:
            return await weather_client.query(city=city, units="metric")
        except httpx.HTTPStatusError as exc:
            raise ToolException(f"上游接口 {exc.response.status_code} 错误")
    def render_tool_error(error: Exception) -> str:
        if isinstance(error, ToolException):
            return str(error)
        return "内部错误：请稍后重试"

通过以上配置，可以创建一个结构化的工具实例，用于查询城市的实时天气：

weather_tool = StructuredTool.from_function(
        name="query_weather",
        description="查询城市实时天气",
        func=fetch_weather,
        coroutine=fetch_weather,
        args_schema=WeatherInput,
        handle_tool_error=render_tool_error,
    )

日志与可观察性

在工具内部记录关键信息，如日志、执行时间以及重试次数，有助于故障排查。

trace_id

幂等与重试

对于可以重放的调用，建议在工具内部实现幂等性和自动重试机制，这样外部 Agent 无需关心重试逻辑。

async_retry

全局异常处理函数

可以构建一个全局的工具错误处理函数，该函数可以根据服务名称格式化错误信息，提供更具体的错误反馈：

from collections.abc import Callable
    def build_global_tool_error_handler(service_name: str) -> Callable[[Exception], str]:
        def handle(error: Exception) -> str:
            logger.error("tool=%s error=%s", service_name, error)
            if isinstance(error, ToolException):
                return str(error)
            return f"{service_name} 暂时不可用，请稍后重试"
        return handle
    global_tool_error_handler = build_global_tool_error_handler("weather")
    weather_tool = StructuredTool.from_function(
        ...,
        handle_tool_error=global_tool_error_handler,
    )

通过这种方式，不仅可以根据工具名称和 trace_id 提供统一格式化的错误信息，还能通过工厂函数为不同的工具生成专门的错误处理器，既符合类型注解的要求，又便于集中管理和维护。

handle_tool_error