1. 为什么需要 ToolRuntime

在 LangChain 里写 Agent 时，工具（Tool）最初往往只是「输入参数 → 返回字符串」的纯函数。但真实业务里，工具几乎不可能这么干净：

• 它需要知道当前是谁在调用（用户 ID、租户、权限、语言偏好）——这类信息在调用链里是只读的，属于 context。
• 它需要读写本次会话的中间结果（已检索到的文档、已确认的订单号、多轮累积的待办）——这类信息在单次运行内可变，属于 state。
• 它需要跨会话持久化数据（用户画像、长期记忆、缓存、文件）——这类信息要活过本次运行，属于 store。

如果把这些东西都塞进工具参数，模型就得自己「记住」并「传回」这些值，既浪费 token 又极易出错。ToolRuntime 就是 LangChain 给出的解法：把运行时上下文注入到工具内部，让模型只关心业务参数，其余由框架自动传递。

一句话概括三者的分工：

概念	生命周期	是否可变	典型内容
context	单次调用链	只读	用户身份、租户、配置、API 客户端
state	单次 Agent 运行	可读写	中间结果、待办、计数器、草稿
store	跨运行持久化	可读写	长期记忆、用户画像、缓存、文件

本文用一个「智能客服工单助手」贯穿全篇，把三者都跑通。

2. 环境准备与版本说明

本文基于 LangChain 1.x 的 langgraph 运行时体系（ToolRuntime 位于 langgraph.prebuilt / langchain.tools 相关命名空间，具体导入路径随小版本略有差异，下面给出兼容写法）。

pip install -U langchain langgraph langchain-openai

import os

os.environ["OPENAI_API_KEY"] = "sk-xxx"
# 若使用兼容接口，可设置 base_url
# os.environ["OPENAI_BASE_URL"] = "https://your-proxy/v1"

先确认版本，避免 API 对不上：

import langchain, langgraph

print("langchain:", langchain.__version__)
print("langgraph:", langgraph.__version__)

建议 langchain >= 1.0、langgraph >= 1.0。若你的版本里 ToolRuntime 导入失败，先升级到最新小版本，再对照官方文档确认命名空间。

3. 三个概念的最小可运行示例

先不接大模型，直接看 ToolRuntime 在工具函数里长什么样。核心就一件事：在工具签名里声明一个 runtime: ToolRuntime 参数，框架会自动注入，模型看不到它。

from typing import Annotated
from langchain_core.tools import tool
from langgraph.prebuilt import ToolRuntime

@tool
def who_am_i(runtime: ToolRuntime) -> str:
    """返回当前调用者的身份信息。"""
    ctx = runtime.context
    return f"user_id={ctx.user_id}, tenant={ctx.tenant}"

注意两点：

1. runtime 参数不要写进 docstring 的参数说明里，模型不需要知道它。
2. 工具对模型暴露的参数只有业务参数，runtime 由框架填充。

下面分别把 context、state、store 拆开讲。

4. context：只读的调用上下文

4.1 定义 context 结构

context 通常用一个 dataclass 或 TypedDict 描述，在创建 Agent 时传入。它适合放「一次请求内不变、但每个请求不同」的东西。

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class UserContext:
    user_id: str
    tenant: str
    language: str = "zh"
    permissions: list[str] = field(default_factory=list)
    # 也可以放已经初始化好的客户端，避免工具里重复建连
    db_dsn: Optional[str] = None

4.2 在工具里读取 context

from langchain_core.tools import tool
from langgraph.prebuilt import ToolRuntime

@tool
def query_my_orders(runtime: ToolRuntime[UserContext], limit: int = 5) -> str:
    """查询当前用户的最近订单。

    Args:
        limit: 返回的订单条数上限。
    """
    ctx = runtime.context
    if "order:read" not in ctx.permissions:
        return "无权限查询订单，请联系管理员开通 order:read。"

    # 这里用假数据模拟数据库查询
    orders = [
        {"id": "A1001", "status": "已发货", "amount": 199.0},
        {"id": "A1002", "status": "待付款", "amount": 59.9},
    ][:limit]

    lines = [f"- {o['id']} | {o['status']} | ¥{o['amount']}" for o in orders]
    return f"用户 {ctx.user_id}（租户 {ctx.tenant}）的订单：\n" + "\n".join(lines)

关键点：权限校验放在工具内部，而不是让模型判断。 模型可能被提示词诱导，但 runtime.context 是服务端注入的，模型改不了。

4.3 创建 Agent 并传入 context

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

agent = create_react_agent(
    model=llm,
    tools=[query_my_orders],
    context_schema=UserContext,   # 声明 context 的类型
)

ctx = UserContext(
    user_id="u_1024",
    tenant="acme",
    permissions=["order:read"],
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "帮我看看最近的订单"}]},
    context=ctx,
)

print(result["messages"][-1].content)

context_schema 让框架知道该注入什么类型；invoke(..., context=ctx) 把实例传进去。工具里 runtime.context 拿到的就是这个 ctx。

4.4 context 的常见坑

• 不要试图在工具里修改 context。 它是只读语义，改了也不会回传，容易造成「以为生效了」的错觉。
• 不要把大对象塞进 context。 它会被序列化/传递，放数据库连接池这类不可序列化对象时要确认运行时是否支持（多数场景建议放工厂函数或全局单例）。
• context 不是记忆。 它只活在一次调用链里，跨请求要重新构造。

5. state：单次运行内可读写的中间状态

5.1 state 与 context 的区别

context 是「外部给的、只读的」；state 是「运行过程中自己攒的、可读写的」。典型场景：多步检索时累积文档、记录已确认的工单号、统计工具调用次数。

5.2 定义 state 结构

from typing import TypedDict, Annotated
from operator import add

class AgentState(TypedDict):
    messages: Annotated[list, add]
    # 自定义字段
    ticket_id: str | None
    collected_docs: Annotated[list[str], add]
    tool_calls_count: int

Annotated[list, add] 表示这个字段用「追加」方式合并，适合累积型数据。

5.3 在工具里读写 state

from langchain_core.tools import tool
from langgraph.prebuilt import ToolRuntime

@tool
def create_ticket(runtime: ToolRuntime[UserContext, AgentState], summary: str) -> str:
    """根据问题摘要创建工单。

    Args:
        summary: 问题的一句话摘要。
    """
    ctx = runtime.context
    state = runtime.state

    # 读 state：如果已经建过工单，直接复用
    if state.get("ticket_id"):
        return f"本次会话已有工单 {state['ticket_id']}，无需重复创建。"

    # 模拟创建
    new_id = f"T-{ctx.tenant}-{abs(hash(summary)) % 10000:04d}"

    # 写 state：通过 runtime 提供的更新接口
    runtime.update_state({"ticket_id": new_id})

    return f"已创建工单 {new_id}，摘要：{summary}"

不同小版本里写 state 的接口名可能是 runtime.update_state(...) 或返回一个 Command(update=...)。若你的版本报 AttributeError，优先查官方文档的「writing to state from tools」章节，两种写法语义一致：把增量合并回图状态。

5.4 完整跑一遍

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

agent = create_react_agent(
    model=llm,
    tools=[query_my_orders, create_ticket],
    context_schema=UserContext,
    state_schema=AgentState,
)

ctx = UserContext(user_id="u_1024", tenant="acme", permissions=["order:read"])

result = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "我的订单 A1002 一直没付款成功，帮我建个工单"}
        ],
        "ticket_id": None,
        "collected_docs": [],
        "tool_calls_count": 0,
    },
    context=ctx,
)

print(result["messages"][-1].content)
print("最终 state.ticket_id =", result.get("ticket_id"))

第二次让模型再建工单时，工具会读到 state["ticket_id"] 已存在，直接复用——这就是 state 的价值：让工具之间通过共享状态协作，而不是靠模型在对话里复述。

5.5 state 的注意事项

• state 字段要在初始输入里给默认值，否则工具里 state.get(...) 可能拿到 None 甚至 KeyError。
• 并发工具调用要小心：多个工具同时写同一字段时，合并策略（add / 覆盖）决定最终结果，设计时想清楚。
• state 不持久化：运行结束就没了，需要留存请落到 store。

6. store：跨会话的持久化存储

6.1 store 是什么

store 是 LangGraph 提供的键值存储抽象，支持命名空间（namespace）分层，适合放长期记忆、用户画像、缓存。它和 state 最大的区别是：store 活过本次运行，下次 invoke 还能读到。

6.2 初始化 store

from langgraph.store.memory import InMemoryStore

# 开发阶段用内存版，生产可换 PostgresStore 等
store = InMemoryStore()

6.3 在工具里读写 store

from langchain_core.tools import tool
from langgraph.prebuilt import ToolRuntime

@tool
def remember_preference(runtime: ToolRuntime[UserContext], key: str, value: str) -> str:
    """记住用户的某个偏好设置。

    Args:
        key: 偏好名称，如 language、channel。
        value: 偏好值。
    """
    ctx = runtime.context
    store = runtime.store

    namespace = ("users", ctx.user_id, "preferences")
    store.put(namespace, key, {"value": value})
    return f"已记住偏好 {key}={value}"

@tool
def recall_preference(runtime: ToolRuntime[UserContext], key: str) -> str:
    """读取用户之前记住的偏好。

    Args:
        key: 偏好名称。
    """
    ctx = runtime.context
    store = runtime.store

    namespace = ("users", ctx.user_id, "preferences")
    item = store.get(namespace, key)
    if item is None:
        return f"没有找到偏好 {key}"
    return f"偏好 {key}={item.value['value']}"

namespace 用元组分层，("users", user_id, "preferences") 天然实现了按用户隔离，不需要自己拼字符串 key。

6.4 把 store 挂到 Agent 上

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

agent = create_react_agent(
    model=llm,
    tools=[remember_preference, recall_preference],
    context_schema=UserContext,
    store=store,          # 关键：把 store 注入运行时
)

ctx = UserContext(user_id="u_1024", tenant="acme")

# 第一次：记住偏好
agent.invoke(
    {"role": "user", "content": "以后都用英文回复我，帮我记住"},
    context=ctx,
)

# 第二次：新会话，仍能读到
result = agent.invoke(
    {"role": "user", "content": "我之前设置的回复语言是什么？"},
    context=ctx,
)
print(result["messages"][-1].content)

第二次 invoke 是全新的运行，state 已经清空，但 store 里的偏好还在——这就是持久化的意义。

6.5 store 的进阶用法

按前缀搜索，适合做「回忆最近 N 条记忆」：

@tool
def list_preferences(runtime: ToolRuntime[UserContext]) -> str:
    """列出当前用户的所有偏好。"""
    ctx = runtime.context
    namespace = ("users", ctx.user_id, "preferences")
    items = runtime.store.search(namespace)
    if not items:
        return "暂无偏好记录"
    return "\n".join(f"- {it.key}: {it.value['value']}" for it in items)

语义检索（需要配置 embedding）：InMemoryStore(index={"embed": embeddings, "dims": 1536})，之后用 store.search(namespace, query="...") 做相似度召回，适合长期记忆场景。

7. 三者协同：一个完整的客服工单助手

把 context、state、store 放进同一个 Agent，看它们如何各司其职。

from dataclasses import dataclass, field
from typing import TypedDict, Annotated
from operator import add

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import ToolRuntime, create_react_agent
from langgraph.store.memory import InMemoryStore


# ---------- context ----------
@dataclass
class UserContext:
    user_id: str
    tenant: str
    permissions: list[str] = field(default_factory=list)


# ---------- state ----------
class AgentState(TypedDict):
    messages: Annotated[list, add]
    ticket_id: str | None
    collected_docs: Annotated[list[str], add]


# ---------- tools ----------
@tool
def search_kb(runtime: ToolRuntime[UserContext, AgentState], query: str) -> str:
    """检索知识库。

    Args:
        query: 检索关键词。
    """
    # 假数据
    docs = {
        "付款": "付款失败常见原因：银行卡限额、网络超时、风控拦截。",
        "发货": "默认 48 小时内发货，偏远地区顺延。",
    }
    hit = [v for k, v in docs.items() if k in query]
    if not hit:
        return "知识库未命中，建议转人工。"

    # 写入 state，供后续工具使用
    runtime.update_state({"collected_docs": hit})
    return "\n".join(hit)


@tool
def create_ticket(runtime: ToolRuntime[UserContext, AgentState], summary: str) -> str:
    """创建工单。

    Args:
        summary: 问题摘要。
    """
    ctx, state = runtime.context, runtime.state
    if state.get("ticket_id"):
        return f"已存在工单 {state['ticket_id']}，复用。"

    new_id = f"T-{ctx.tenant}-{abs(hash(summary)) % 10000:04d}"
    runtime.update_state({"ticket_id": new_id})
    return f"已创建工单 {new_id}"


@tool
def save_profile(runtime: ToolRuntime[UserContext], key: str, value: str) -> str:
    """保存用户长期画像。

    Args:
        key: 画像字段。
        value: 字段值。
    """
    ctx = runtime.context
    runtime.store.put(("users", ctx.user_id), key, {"value": value})
    return f"已保存 {key}={value}"


# ---------- agent ----------
store = InMemoryStore()
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

agent = create_react_agent(
    model=llm,
    tools=[search_kb, create_ticket, save_profile],
    context_schema=UserContext,
    state_schema=AgentState,
    store=store,
)

ctx = UserContext(user_id="u_1024", tenant="acme", permissions=["kb:read"])

result = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "我付款一直失败，帮我查一下并建个工单"}
        ],
        "ticket_id": None,
        "collected_docs": [],
    },
    context=ctx,
)

print(result["messages"][-1].content)
print("ticket_id:", result.get("ticket_id"))
print("collected_docs:", result.get("collected_docs"))

执行链路大致是：

8. 选型对照表与最佳实践

需求	用哪个	理由
当前用户是谁、什么权限	context	服务端注入，模型不可篡改
本次会话已查到哪些文档	state	运行内共享，工具间协作
用户上次说的偏好	store	跨会话持久化
调用次数统计	state	运行内计数
长期记忆 / 向量召回	store	支持 namespace + 语义检索

几条实战经验：

1. 权限永远放 context，别放 state。 state 可被工具改写，context 不行。
2. state 字段给默认值。 初始输入里把 ticket_id、collected_docs 都显式初始化，避免 None 引发的分支错误。
3. store 的 namespace 要设计好。 用元组分层（("users", uid, "preferences")），比拼接字符串更安全、更好检索。
4. 不要把大对象塞进 state。 state 会随图状态传递，放几百 MB 的文档会拖垮性能，存引用或 ID 即可。
5. 工具 docstring 只写业务参数。 runtime 是注入参数，写进 docstring 会误导模型。

9. 常见问题排查

Q1：工具里 runtime.context 是 None？
检查 create_react_agent 是否传了 context_schema，以及 invoke 是否传了 context=。

Q2：runtime.update_state 报 AttributeError？
版本差异。旧版可能要求工具返回 Command(update={...})，新版支持 runtime.update_state。查一下当前版本的迁移说明。

Q3：store 里读不到上次写的数据？
确认 create_react_agent(store=store) 传的是同一个 store 实例；InMemoryStore 是进程内存储，重启进程数据就没了，生产环境换 PostgresStore。

Q4：模型总是自己编造 user_id？
因为工具参数里暴露了 user_id。把它从工具签名里删掉，改从 runtime.context 取，模型就无从编造。

10. 小结

ToolRuntime 的本质是把「模型不该操心的事」从工具参数里剥离出来：

• context：外部注入的只读身份与配置，做权限校验、多租户隔离。
• state：单次运行内的共享草稿纸，让工具之间协作而不靠模型复述。
• store：跨会话的持久化仓库，承载长期记忆与用户画像。

掌握这三者的边界，你的 Agent 工具就能从「玩具函数」升级为「能上生产的业务组件」。下一步可以尝试把 InMemoryStore 换成 PostgresStore，并给 store 配上 embedding 做语义记忆召回。