J'Blog

4 课 · 核心

语义检索

基于 Chroma 向量库,与 Workspace 知识库分离。用自然语言描述你想查的内容。 非管理员每人每天可检索 5 次(消耗站点 AI 配额)。

工具调用

原文:Tools 工具调用

一句话

工具(tools)扩展了 agents 的能力,使其能够获取实时数据、执行代码、查询外部数据库并在世界中执行操作。

什么时候翻这页

当你需要为 agent 创建自定义工具,或者需要理解工具如何与模型交互时。 当你需要访问运行时信息(如对话历史、用户数据和持久化内存)时。 当你需要处理工具错误或进行动态工具选择时。

核心概念

  • 工具(Tool): 可调用函数,具有明确定义的输入和输出,传递给 chat model
  • 工具描述(Tool Description): 函数的文档字符串,帮助模型理解何时使用该工具
  • 工具运行时(ToolRuntime): 提供对状态、上下文、存储等运行时信息的访问
  • 中间件(Middleware): 用于处理工具调用、错误处理和动态工具选择的机制
  • 直接返回(return_direct): 工具输出直接作为最终响应,绕过额外的模型调用

怎么做

创建基本工具

使用 @tool 装饰器创建最简单的工具:

from langchain.tools import tool

@tool
def search_database(query: str, limit: int = 10) -> str:
    """Search the customer database for records matching the query.

    Args:
        query: Search terms to look for
        limit: Maximum number of results to return
    """
    return f"Found {limit} results for '{query}'"

自定义工具属性

自定义工具名称

@tool("web_search")  # 自定义名称
def search(query: str) -> str:
    """Search the web for information."""
    return f"Results for: {query}"

自定义工具描述

@tool("calculator", description="Performs arithmetic calculations. Use this for any math problems.")
def calc(expression: str) -> str:
    """Evaluate mathematical expressions."""
    return str(eval(expression))

高级模式定义

使用 Pydantic 模型定义复杂输入:

from pydantic import BaseModel, Field
from typing import Literal

class WeatherInput(BaseModel):
    """Input for weather queries."""
    location: str = Field(description="City name or coordinates")
    units: Literal["celsius", "fahrenheit"] = Field(
        default="celsius",
        description="Temperature unit preference"
    )
    include_forecast: bool = Field(
        default=False,
        description="Include 5-day forecast"
    )

@tool(args_schema=WeatherInput)
def get_weather(location: str, units: str = "celsius", include_forecast: bool = False) -> str:
    """Get current weather and optional forecast."""
    temp = 22 if units == "celsius" else 72
    result = f"Current weather in {location}: {temp} degrees {units[0].upper()}"
    if include_forecast:
        result += "\nNext 5 days: Sunny"
    return result

访问运行时信息

通过 ToolRuntime 参数访问运行时信息:

from langchain.tools import tool, ToolRuntime

@tool
def get_last_user_message(runtime: ToolRuntime) -> str:
    """Get the most recent message from the user."""
    messages = runtime.state["messages"]
    
    # Find the last human message
    for message in reversed(messages):
        if isinstance(message, HumanMessage):
            return message.content
    
    return "No user messages found"

更新状态

使用 Command 更新 agent 的状态:

from langchain.agents import AgentState
from langchain.messages import ToolMessage
from langchain.tools import ToolRuntime, tool
from langgraph.types import Command

class CustomState(AgentState):
    user_name: str

@tool
def set_user_name(new_name: str, runtime: ToolRuntime[None, CustomState]) -> Command:
    """Set the user's name in the conversation state."""
    return Command(
        update={
            "user_name": new_name,
            "messages": [
                ToolMessage(
                    content=f"User name set to {new_name}.",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
        }
    )

直接返回工具

使用 return_direct 让工具输出直接作为最终响应:

from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI

@tool(return_direct=True)
def fetch_order_status(order_id: str) -> str:
    """Fetch the current status of a customer order."""
    # In production, query your order management system here
    return f"Order {order_id} is shipped and will arrive in 2 days."

agent = create_agent(
    ChatOpenAI(model="google_genai:gemini-3.5-flash"),
    tools=[fetch_order_status],
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "What is the status of order #12345?"}]
})
# The agent returns the tool output directly without another LLM call:
# "Order 12345 is shipped and will arrive in 2 days."

错误处理

使用中间件处理工具错误:

from collections.abc import Callable
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage
from langchain.tools.tool_node import ToolCallRequest

@wrap_tool_call
def handle_tool_errors(
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], ToolMessage],
) -> ToolMessage:
    """Convert tool exceptions into ToolMessages the model can handle."""
    try:
        return handler(request)
    except Exception as e:
        return ToolMessage(
            content=f"Tool error: Please check your input and try again. ({e})",
            tool_call_id=request.tool_call["id"],
        )

agent = create_agent(
    model="google_genai:gemini-3.5-flash",
    tools=[],
    middleware=[handle_tool_errors],
)

动态工具选择

基于状态过滤工具:

from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable

@wrap_model_call
def state_based_tools(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
    """Filter tools based on conversation State."""
    # Read from State: check if user has authenticated
    state = request.state
    is_authenticated = state.get("authenticated", False)
    message_count = len(state["messages"])

    # Only enable sensitive tools after authentication
    if not is_authenticated:
        tools = [t for t in request.tools if t.name.startswith("public_")]
        request = request.override(tools=tools)
    elif message_count < 5:
        # Limit tools early in conversation
        tools = [t for t in request.tools if t.name != "advanced_search"]
        request = request.override(tools=tools)

    return handler(request)

agent = create_agent(
    model="gpt-5.4",
    tools=[public_search, private_search, advanced_search],
    middleware=[state_based_tools]
)

命令 / API 速查

  • @tool: 创建基本工具的装饰器
  • ToolRuntime: 提供对运行时信息的访问
  • return_direct: 工具参数,使输出直接作为最终响应
  • wrap_tool_call: 中间件装饰器,用于包装工具调用
  • wrap_model_call: 中间件装饰器,用于包装模型调用
  • args_schema: 指定工具输入模式的参数
  • Command: 用于更新 agent 状态的类型

与 LangGraph / RAG 手册的联系

  • LangGraph 提供了更底层的编排能力,而 LangChain 的工具系统是构建在 LangGraph 之上的高层抽象
  • RAG 系统通常使用 retriever 工具来获取外部知识,工具系统可以与 RAG 集成以增强 agent 的能力
  • 工具系统与 LangGraph 的状态管理紧密集成,允许工具访问和更新对话状态

初学者易错点

  1. 忽略类型提示: 类型提示是必需的,因为它们定义了工具的输入模式
  2. 使用保留参数名: 避免使用 configruntime 作为工具参数名,这些是保留的
  3. 过度使用 return_direct: 仅在工具输出是完整、用户就绪的答案时使用,否则模型无法进一步处理结果
  4. 忘记错误处理: 工具可能会失败,适当的错误处理对于健壮的 agent 至关重要
  5. 工具名称格式: 偏好使用 snake_case 而不是包含空格或特殊字符的名称

相关词条

  • create_agent: 创建 agent 的方法
  • middleware: 处理工具调用和模型调用的中间件
  • models: LangChain 中的模型系统
  • RAG: 检索增强生成系统
  • retriever: 用于检索外部知识的组件

官方原文:https://docs.langchain.com/oss/python/langchain/tools