LangChain：Tool访问上下文（记忆）

当Tool（工具）能够访问运行时信息（如对话历史、用户数据和持久内存）时，它们的功能最为强大。

Tool访问上下文

当工具能够访问运行时信息（如对话历史、用户数据和持久内存）时，它们的功能最为强大。

Tool（工具）可以通过 ToolRuntime 参数访问运行时信息，该参数提供：

组件	描述	用例
State	短期内存 - 当前对话中存在的可变数据 （消息、计数器、自定义字段）	访问对话历史， 跟踪工具调用计数
Context	不可变配置，在调用时传递 （用户 ID、会话信息）	根据用户身份个性化响应
Store	长期内存 - 跨对话持久化的数据	保存用户偏好，维护知识库
Stream Writer	在工具执行期间发出实时更新	显示长时间运行操作的进度
Execution Info	当前执行的标识和重试信息 （线程 ID、运行 ID、尝试次数）	访问线程/运行 ID， 根据重试状态调整行为
Server Info	在 LangGraph Server 上运行时的服务器 特定元数据（助手 ID、图形 ID、已认证用户）	访问助手 ID、图形 ID 或已认证用户信息
Config	执行的 RunnableConfig	访问回调、标签和元数据
Tool Call ID	当前工具调用的唯一标识符	关联日志和模型调用的工具调用

补充说明：

• ToolRuntime 是一个特殊的参数，它不会出现在工具的输入模式中，因此大模型不会也无法提供该参数的值。LangChain 在执行工具时自动注入。
• 短期内存（State）与长期内存（Store）的区别：State 只在单次对话的当前生命周期中存在，对话结束或重置后消失；Store 可以将数据持久化到数据库或磁盘，跨会话复用。

短期内存 (State)

State 代表对话生命周期内的短期内存。它包括消息历史以及在图形状态中定义的自定义字段。

将 runtime: ToolRuntime 添加到的工具签名中以访问状态。

这个参数会被自动注入并对 LLM 隐藏——它不会出现在工具的模式中。

访问状态

工具可以使用 runtime.state 访问当前的对话状态：

from langchain.tools import tool, ToolRuntime
from langchain.messages import HumanMessage

@tool
def get_last_user_message(runtime: ToolRuntime) -> str:
    """Get the most recent message from the user."""
    # runtime.state 是一个字典，通常包含 "messages" 键，存储对话历史
    messages = runtime.state["messages"]
    # 从消息列表中反向查找最后一条用户消息
    for message in reversed(messages):
        if isinstance(message, HumanMessage):
            return message.content
    return "No user messages found"

# 访问自定义状态字段的示例
@tool
def get_user_preference(pref_name: str, runtime: ToolRuntime) -> str:
    """Get a user preference value."""
    # 假设 state 中有一个 "user_preferences" 字段
    preferences = runtime.state.get("user_preferences", {})
    return preferences.get(pref_name, "Not set")

runtime 参数对模型是隐藏的。对于上面的示例，模型在工具模式中只看到 pref_name。

更新状态

使用 Command 来更新智能体的状态。这对于需要更新自定义状态字段的工具很有用。

在更新中包含一个 ToolMessage，以便模型可以看到工具调用的结果：

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."""
    # Command 用来同时更新状态和返回消息
    return Command(
        update={
            "user_name": new_name,   # 更新状态中的 user_name 字段
            "messages": [
                ToolMessage(
                    content=f"User name set to {new_name}.",
                    tool_call_id=runtime.tool_call_id,   # 关联本次工具调用的 ID
                )
            ],
        }
    )

当工具更新状态变量时，请考虑为这些字段定义 reducer。由于 LLM 可以并行调用多个工具，reducer 决定了当同一状态字段被并发工具调用更新时如何解决冲突。

补充说明：

• Command 是 LangGraph 中用于更新图状态的核心机制。工具返回 Command 后，LangGraph 执行器会应用其中的 update 字段，将新值合并到当前状态。
• Reducer：当多个工具并行运行并试图更新同一状态字段时，需要定义合并策略（reducer），例如取最大值、拼接列表等。默认情况下，直接赋值可能导致覆盖。

上下文 (Context)

Context 提供在调用时传递的不可变配置数据。将其用于用户 ID、会话详细信息或不应在对话期间更改的应用程序特定设置。

通过 runtime.context 访问上下文：

from dataclasses import dataclass
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime

# 模拟的用户数据库
USER_DATABASE = {
    "user123": {
        "name": "Alice Johnson",
        "account_type": "Premium",
        "balance": 5000,
        "email": "alice@example.com"
    },
    "user456": {
        "name": "Bob Smith",
        "account_type": "Standard",
        "balance": 1200,
        "email": "bob@example.com"
    }
}

@dataclass
class UserContext:
    user_id: str   # 上下文只包含用户ID，不可变

@tool
def get_account_info(runtime: ToolRuntime[UserContext]) -> str:
    """Get the current user's account information."""
    # 从注入的上下文中获取用户ID
    user_id = runtime.context.user_id
    if user_id in USER_DATABASE:
        user = USER_DATABASE[user_id]
        return f"Account holder: {user['name']}\nType: {user['account_type']}\nBalance: ${user['balance']}"
    return "User not found"

model = ChatOpenAI(model="gpt-4o")
agent = create_agent(
    model,
    tools=[get_account_info],
    context_schema=UserContext,          # 告诉智能体上下文的类型
    system_prompt="You are a financial assistant."
)

# 调用时通过 context 参数注入真实用户数据
result = agent.invoke(
    {"messages": [{"role": "user", "content": "What's my current balance?"}]},
    context=UserContext(user_id="user123")
)

补充说明：

• Context 是不可变的，在整个对话过程中不会改变，适合存放请求级别的元数据（如用户 ID、租户 ID、会话令牌）。

• 与 State 不同，Context 不会随对话迭代而更新，也不需要在图中定义 reducer。

长期内存 (Store)

BaseStore 提供了跨对话持久化的存储。与 state（短期内存）不同，保存到存储中的数据在未来的会话中仍然可用。通过 runtime.store 访问存储。存储使用命名空间/键模式来组织数据：对于生产部署，请使用持久存储实现（如 PostgresStore），而不是 InMemoryStore。

from typing import Any
from langgraph.store.memory import InMemoryStore
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from langchain_openai import ChatOpenAI

# 读取长期存储中的用户信息
@tool
def get_user_info(user_id: str, runtime: ToolRuntime) -> str:
    """Look up user info."""
    store = runtime.store
    # get(命名空间, 键)，命名空间用于组织数据，例如按类型分组
    user_info = store.get(("users",), user_id)
    # .value 存储实际数据，如果不存在则返回 None
    return str(user_info.value) if user_info else "Unknown user"

# 保存信息到长期存储
@tool
def save_user_info(user_id: str, user_info: dict[str, Any], runtime: ToolRuntime) -> str:
    """Save user info."""
    store = runtime.store
    store.put(("users",), user_id, user_info)   # 存储到 "users" 命名空间下，键为 user_id
    return "Successfully saved user info."

model = ChatOpenAI(model="gpt-4o")
store = InMemoryStore()      # 内存存储（仅用于测试，重启后数据丢失）
agent = create_agent(
    model,
    tools=[get_user_info, save_user_info],
    store=store               # 将存储传入智能体，工具可通过 runtime 访问
)

# 第一次会话：保存用户信息
agent.invoke({
    "messages": [{"role": "user", "content": "Save the following user: userid: abc123, name: Foo, age: 25, email: foo@langchain.dev"}]
})

# 第二次会话（不同的对话线程）：获取用户信息
agent.invoke({
    "messages": [{"role": "user", "content": "Get user info for user with id 'abc123'"}]
})
# 输出：Here is the user info for user with ID "abc123": Name: Foo, Age: 25, Email: foo@langchain.dev

补充说明：

• Store API 设计为通用的键值存储，命名空间可以是元组，支持层级。例如 ("users", "premium") 与 ("users", "trial") 是不同的命名空间。
• 生产环境推荐使用 PostgresStore、RedisStore 等实现了 BaseStore 的持久化后端。

流式写入器 (Stream writer)

在工具执行期间流式传输实时更新。这对于在长时间运行的操作期间向用户提供进度反馈非常有用。使用 runtime.stream_writer 发出自定义更新：

from langchain.tools import tool, ToolRuntime

@tool
def get_weather(city: str, runtime: ToolRuntime) -> str:
    """Get weather for a given city."""
    writer = runtime.stream_writer
    # 流式发送中间状态，用户界面可以实时显示
    writer(f"Looking up data for city: {city}")
    writer(f"Acquired data for city: {city}")
    return f"It's always sunny in {city}!"

如果在工具中使用 runtime.stream_writer，则必须在 LangGraph 执行上下文中调用该工具。

补充说明：

• writer 是一个可调用对象，每次调用都会触发一个 streaming 事件，可在前端或日志中捕获。
• 前提条件：使用 runtime.stream_writer 的工具必须在支持流的 LangGraph 执行上下文中运行，否则写入操作可能无效。

执行信息 (Execution info)

通过 runtime.execution_info 从工具内部访问线程 ID、运行 ID 和重试状态：

from langchain.tools import tool, ToolRuntime

@tool
def log_execution_context(runtime: ToolRuntime) -> str:
    """Log execution identity information."""
    info = runtime.execution_info
    print(f"Thread: {info.thread_id}, Run: {info.run_id}")
    print(f"Attempt: {info.node_attempt}")
    return "done"

• thread_id 和 run_id 用于追踪和日志关联，尤其在分布式部署或长时间运行的工作流中非常有用。
• node_attempt 在智能体自动重试失败步骤时递增，可用于实现退避策略或记录重试次数。

服务器信息 (Server info)

当你的工具在 LangGraph Server 上运行时，通过 runtime.server_info 访问助手 ID、图形 ID 和已认证用户：

from langchain.tools import tool, ToolRuntime

@tool
def get_assistant_scoped_data(runtime: ToolRuntime) -> str:
    """Fetch data scoped to the current assistant."""
    server = runtime.server_info
    if server is not None:
        # 仅在 LangGraph Server 环境中这些字段才不为 None
        print(f"Assistant: {server.assistant_id}, Graph: {server.graph_id}")
        if server.user is not None:
            print(f"User: {server.user.identity}")
    return "done"

当工具不在 LangGraph Server 上运行时（例如，在本地开发或测试期间），server_info 为 None。

• 该功能高度依赖部署环境。本地开发或简单脚本中 server_info 为 None，因此代码需要做空值检查。
• assistant_id 和 graph_id 可用来实现多租户数据隔离或动态加载不同的配置。

ToolNode

ToolNode 是一个预构建的节点，用于在 LangGraph 工作流中执行工具。它会自动处理并行工具执行、错误处理和状态注入。对于需要对工具执行模式进行精细控制的自定义工作流，请使用 ToolNode，而不是 create_agent。它是支持智能体工具执行的基础构建块。

• ToolNode 是 LangGraph 内置的图节点，它接收来自 LLM 的工具调用请求（通常存储在消息列表的 tool_calls 字段中），并发执行这些工具，然后将结果作为 ToolMessage 添加回消息列表。
• 与 create_agent 高层封装相比，ToolNode 提供了更底层的控制，允许开发者自定义图结构、条件分支和循环。

基本用法

from langchain.tools import tool
from langgraph.prebuilt import ToolNode
from langgraph.graph import StateGraph, MessagesState, START, END

@tool
def search(query: str) -> str:
    """Search for information."""
    return f"Results for: {query}"

@tool
def calculator(expression: str) -> str:
    """Evaluate a math expression."""
    return str(eval(expression))

# 创建一个 ToolNode，管理搜索和计算器两个工具
tool_node = ToolNode([search, calculator])

# 在图中使用该节点
builder = StateGraph(MessagesState)
builder.add_node("tools", tool_node)   # 将 "tools" 命名为该节点
# ... 添加 LLM 节点和其他边

工具返回值

你可以为工具选择不同的返回值：

• 返回一个字符串，用于人类可读的结果。
• 返回一个对象，用于模型应解析的结构化结果。
• 当需要写入状态时，返回一个带有可选消息的 Command。

返回字符串

当工具应为模型提供纯文本以供其阅读并在下一次响应中使用时，返回一个字符串。

@tool
def get_weather(city: str) -> str:
    """Get weather for a city."""
    return f"It is currently sunny in {city}."

行为：返回值将转换为 ToolMessage。模型看到该文本并决定下一步做什么。除非模型或其他工具稍后更改，否则不会更改智能体的状态字段。

返回对象

当你的工具生成模型应检查的结构化数据时，返回一个对象（例如，一个字典）。

@tool
def get_weather_data(city: str) -> dict:
    """Get structured weather data for a city."""
    return {
        "city": city,
        "temperature_c": 22,
        "conditions": "sunny",
    }

行为：对象被序列化并作为工具输出发送回去。模型可以读取特定字段并对其进行推理。与返回字符串一样，这不会直接更新图形状态。

返回 Command

当工具需要更新图形状态（例如，设置用户偏好或应用程序状态）时，返回一个 Command。你可以返回带有或不带 ToolMessage 的 Command。如果模型需要看到工具成功（例如，确认偏好更改），请在更新中包含一个 ToolMessage，并使用 runtime.tool_call_id 作为 tool_call_id 参数。

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

@tool
def set_language(language: str, runtime: ToolRuntime) -> Command:
    """Set the preferred response language."""
    return Command(
        update={
            "preferred_language": language,
            "messages": [
                ToolMessage(
                    content=f"Language set to {language}.",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
        }
    )

行为：Command 使用 update 更新状态。更新后的状态在同一运行的后续步骤中可用。对于可能被并行工具调用更新的字段，请使用 reducer。

补充说明：

• 当工具返回 Command 且其中包含 ToolMessage 时，模型能立即看到该消息，并据此决定是否继续进行其他操作。
• 如果不包含 ToolMessage，模型仍然可以通过状态变化间接得知结果，但模型不会主动“看到”一条反馈消息。

错误处理

配置如何处理工具错误。参阅 ToolNode API 参考以获取所有选项。

from langgraph.prebuilt import ToolNode

# 默认行为：捕获调用异常，重新抛出执行错误
tool_node = ToolNode(tools)

# 捕获所有错误，并将错误消息返回给 LLM
tool_node = ToolNode(tools, handle_tool_errors=True)

# 自定义错误消息（字符串）
tool_node = ToolNode(tools, handle_tool_errors="Something went wrong, please try again.")

# 自定义错误处理函数
def handle_error(e: ValueError) -> str:
    return f"Invalid input: {e}"

tool_node = ToolNode(tools, handle_tool_errors=handle_error)

# 只捕获特定类型的异常
tool_node = ToolNode(tools, handle_tool_errors=(ValueError, TypeError))

补充说明：

• handle_tool_errors 的默认行为（False）会直接抛出异常，中断图执行。
• 当设置为 True 或自定义消息/函数时，工具节点会捕获异常，并将错误信息包装成 ToolMessage 返回给 LLM，让模型有机会纠正错误或向用户解释问题。
• 这是构建鲁棒智能体的关键机制，避免因个别工具调用失败导致整个对话崩溃。

使用tools_condition进行路由

使用 tools_condition 根据 LLM 是否进行了工具调用来进行条件路由：

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

builder = StateGraph(MessagesState)
builder.add_node("llm", call_llm)          # 假设 call_llm 是调用 LLM 的函数
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "llm")
# 条件边：如果 LLM 回复中包含工具调用，则路由到 "tools" 节点；否则结束
builder.add_conditional_edges("llm", tools_condition)
builder.add_edge("tools", "llm")           # 工具执行后返回 LLM 继续处理
graph = builder.compile()

补充说明：

• tools_condition 检查消息列表中最后一条消息的 tool_calls 属性，若不为空则返回字符串 "tools"，否则返回 END。
• 这种模式实现了经典的 ReAct 循环（Reason – Act – Observe），即 LLM 思考 -> 调用工具 -> 观察结果 -> 继续思考，直到任务完成。

状态注入

工具可以通过 ToolRuntime 访问当前的图形状态。有关从工具访问状态、上下文和长期内存的更多详细信息，请参阅访问上下文。

from langchain.tools import tool, ToolRuntime
from langgraph.prebuilt import ToolNode

@tool
def get_message_count(runtime: ToolRuntime) -> str:
    """Get the number of messages in the conversation."""
    messages = runtime.state["messages"]
    return f"There are {len(messages)} messages."

tool_node = ToolNode([get_message_count])

补充说明：

• 这种模式允许工具读取（但不能直接修改）当前图的状态。如需修改，应返回 Command。
• ToolNode 会自动识别带有 runtime 参数的函数，并为它们注入正确的运行时对象。

预构建工具

LangChain 为常见任务（如网页搜索、代码解释、数据库访问等）提供了大量的预构建工具和工具包集合。这些即用型工具可以直接集成到你的智能体中，无需编写自定义代码。有关按类别组织的可用工具的完整列表，请参阅工具和工具包集成页面。

• 预构建工具包括：DuckDuckGo 搜索、Wikipedia 查询、ArXiv 论文检索、Python REPL、SQL 数据库查询、文件系统操作等。
• 工具包（Toolkits）是对一组相关工具的封装，例如 GitHubToolkit、SlackToolkit、VectorStoreToolkit 等。

服务器端工具使用

一些聊天模型具有内置工具，这些工具由模型提供者在服务器端执行。这些包括诸如网页搜索和代码解释器之类的功能，你无需定义或托管工具逻辑。有关启用和使用这些内置工具的详细信息，请参阅各个聊天模型集成页面和工具调用文档。

• 例如 OpenAI 的 code_interpreter、web_search 工具，以及 Anthropic Claude 的 bash、text_editor 工具。
• 这些工具在调用时不需要你编写任何实现代码，只需在请求中声明要使用的工具名称，模型提供方会在其安全环境中执行并返回结果。
• 这种方式的优点是简单、安全（运行在提供方沙箱中），缺点是可定制性差且可能产生额外费用。

资料引用

1. LangChain Docs > Tools