LangChain ToolRuntime 入门：context、state 与 store 的实际用法

1. ToolRuntime 是什么 ToolRuntime 是 LangChain 执行工具时自动传入的运行对象。 工具参数分为两类： 普通参数：模型根据用户问题填写，例如 keyword ； runtime ：框架自动注入，模型不能填写。 from langchain.tools import ToolRuntime, tool @tool def search ( keyword: str , runtime: ToolRuntime ) -> str :     # keyword 由模型填写。     # runtime 由 LangChain 自动传入。     return f"搜索： {keyword} " 2. 三个核心对象 对象 作用 典型内容 context 本次运行的固定条件 用户 ID、权限、数据库连接 state 当前流程的临时状态 消息、步骤、临时结果 store 跨请求保存的数据 用户偏好、长期记忆 记忆方法： context 管运行条件， state 管当前过程， store 管长期数据。 3. context：传入用户和运行依赖 context 由程序创建，再传给 Agent。需要用户身份时，应使用登录系统已经验证过的信息，不能从用户文字中读取用户 ID。 from dataclasses import dataclass from langchain.agents import create_agent from langchain.chat_models import init_chat_model from langchain.tools import ToolRuntime, tool @dataclass class RequestContext :     # 这些值由程序传入，模型不能自行修改。    user_id: int    role: str @tool def show_user ( runtime: ToolRuntime[RequestContext] ) -> str :     # 工具从 context 读取用户身份，不接受模型传入的 user_id。     return f"用户 {runtime.context.user_id} ，角色 {runtime.context.role} " # 根据实际环境配置模型。 model = init_chat_model( "openai:gpt-4o-mini" ) # context_schema 只声明上下文类型。 agent = create_agent(    model=model,    tools=[show_user],    context_schema=RequestContext, ) # 真实用户信息应来自登录系统。 context = RequestContext(user_id= 1001 , role= "user" ) # context 参数才是真正把用户信息传给 Agent。 result = agent.invoke(   { "messages" : [{ "role" : "user" , "content" : "查询当前用户" }]},    context=context, ) print (result[ "messages" ][- 1 ].content) 适用场景：权限判断、按用户查询数据、传入数据库会话、传入租户信息。 注意： context_schema 不会自动创建用户信息；必须先创建 RequestContext(...) ，再使用 context=... 传入。 4. state：当前流程的临时状态 state 是 Agent 当前正在处理的内容。最简单的例子是读取当前消息数量。 from langchain.agents import create_agent from langchain.chat_models import init_chat_model from langchain.tools import ToolRuntime, tool @tool def count_messages ( runtime: ToolRuntime ) -> str :     # messages 是 Agent state 中自带的消息列表。     # get 的第二个参数 [] 是默认值：没有 messages 时返回空列表。    messages = runtime.state.get( "messages" , [])     return f"当前共有 { len (messages)} 条消息" model = init_chat_model( "openai:gpt-4o-mini" ) agent = create_agent(    model=model,    tools=[count_messages], ) # 模型调用工具时，ToolRuntime 会自动带上当前 state。 result = agent.invoke(   { "messages" : [{ "role" : "user" , "content" : "数一下当前消息" }]} ) print (result[ "messages" ][- 1 ].content) 适用场景：读取消息、记录当前步骤、保存重试次数和临时查询结果。 实际用法：统计模型调用次数 当前代码把 model_call_count 放进 state，由中间件在模型调用完成后加 1： from typing import Any , NotRequired from langchain.agents import AgentState from langchain.agents.middleware import AgentMiddleware from langgraph.runtime import Runtime class CustomAgentState ( AgentState ):     # AgentState 已经提供 messages，这里只增加调用次数。    model_call_count: NotRequired[ int ] class ModelCallLimitMiddleware (AgentMiddleware[CustomAgentState]):     # 注册扩展后的 state 类型。    state_schema = CustomAgentState     def after_model (        self, state: CustomAgentState, runtime: Runtime   ) -> dict [ str , Any ]:         # 没有记录时从 0 开始，每完成一次模型调用就加 1。        count = state.get( "model_call_count" , 0 )         return { "model_call_count" : count + 1 } 流式调用时， updates 事件可以读到中间件更新后的 state： from langchain.agents import create_agent from langchain.chat_models import init_chat_model model = init_chat_model( "openai:gpt-4o-mini" ) agent = create_agent(    model=model,    middleware=[ModelCallLimitMiddleware()], ) async def read_model_call_count () -> None :    input_state = {         "messages" : [{ "role" : "user" , "content" : "你好" }],         "model_call_count" : 0 ,   }    model_call_count = 0     async for mode, data in agent.astream(        input_state,        stream_mode=[ "messages" , "updates" ],   ):         if mode != "updates" :             continue        update = data.get( "ModelCallLimitMiddleware.after_model" )         if update:            model_call_count = update[ "model_call_count" ]     print ( f"本轮模型调用次数： {model_call_count} " ) 这里的次数只属于当前 Agent 流程，适合做日志、限次和调试信息。 5. store：跨请求保存数据 store 适合保存下一次请求仍然需要的数据，例如： 用户语言和主题偏好； 用户主动要求记住的信息； 跨会话的设置； 需要复用的业务缓存。 不适合把所有聊天内容无条件放进 store 。长期保存前，应先判断内容是否有价值，并按用户或租户隔离命名空间。 实际项目通常把 store.put() 和 store.get() 封装到工具中。工具从 context 获取当前登录用户 ID，外部调用时不需要重复传入用户 ID： from dataclasses import dataclass from langchain.agents import create_agent from langchain.chat_models import init_chat_model from langchain.tools import ToolRuntime, tool from langgraph.store.memory import InMemoryStore @dataclass class UserContext :     # 这个 ID 应由登录系统传入，不能由模型填写。    user_id: str @tool( description= "保存用户喜欢的颜色。" ) def save_color ( color: str , runtime: ToolRuntime[UserContext] ) -> str :     # 工具通过 runtime.store 保存数据。    namespace = ( "user" , runtime.context.user_id)    runtime.store.put(        namespace,         "favorite_color" ,       { "value" : color},   )     return "颜色偏好已保存" @tool( description= "读取用户之前保存的喜欢颜色。" ) def read_color ( runtime: ToolRuntime[UserContext] ) -> str :     # 读取时使用相同的 namespace 和 key。    namespace = ( "user" , runtime.context.user_id)    item = runtime.store.get(namespace, "favorite_color" )     if item is None :         return "还没有保存颜色偏好"     return f"喜欢的颜色是： {item.value[ 'value' ]} " # store 应在应用启动时创建，多个请求才能共用它。 store = InMemoryStore() model = init_chat_model( "openai:gpt-4o-mini" ) # 把 store 和 context 类型传给 Agent，工具才能使用 runtime.store 和 runtime.context。 agent = create_agent(    model=model,    tools=[save_color, read_color],    context_schema=UserContext,    store=store, ) # 每次请求传入当前登录用户的上下文。 context = UserContext(user_id= "1001" ) agent.invoke(   { "messages" : [{ "role" : "user" , "content" : "记住我喜欢蓝色" }]},    context=context, ) # 下一次请求仍使用同一个 store，工具可以读到之前保存的内容。 next_context = UserContext(user_id= "1001" ) result = agent.invoke(   { "messages" : [{ "role" : "user" , "content" : "我喜欢什么颜色？" }]},    context=next_context, ) print (result[ "messages" ][- 1 ].content) InMemoryStore 只保存在内存中，程序重启后会丢失；需要永久保存时，应换成数据库等持久化实现。 7. 如何选择 本次运行的用户、权限和依赖：使用 context 。 当前流程的消息和临时结果：使用 state 。 下次请求仍要使用的数据：使用 store 。 总结 ToolRuntime 是工具获取运行信息的入口： context 管本次运行条件； state 管当前工作过程； store 管跨请求数据。 先判断数据的生命周期，再选择对象，工具代码就不会混乱。