langchain工具结果透传

在 LangChain / LangGraph 里工具调用结果存起来供下一个工具使用这件事本质上是把结果写回图状态State——Runtime 本身只是一个运行期上下文载体并不是持久存储桶。下面把正确的做法讲清楚。先理清三个概念LangChain v1 基于 LangGraph运行期上下文分三层别搞混 层次性质生命周期用途context配置不可变单次运行user_id、API Key、DB 连接等state状态可变单次运行或单轮对话工具结果、中间变量都存在这里​store长期记忆可变跨对话用户画像、历史偏好工具调用结果要供下一个工具使用正确做法是写回 state——要么作为ToolMessage进messages列表要么写进你自定义的 state 字段 。做法一让结果自然进入 messages最常用LangGraph 的ToolNode执行完工具后会把返回值包装成ToolMessage追加到state[messages]里 。下一轮 LLM 推理时自然能看到这个结果并决定调用下一个工具。关键是 state 里的messages字段必须用累积 reduceradd_messages否则会覆盖掉之前的消息下一轮 LLM 会因为tool 消息找不到配对的 tool_calls而报错 from typing import Annotated, List from langchain_core.messages import BaseMessage from langgraph.graph import StateGraph from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[List[BaseMessage], add_messages]工具本身只需要正常返回结果即可ToolNode会自动处理ToolRuntime注入、ToolMessage构造等工作 from langchain_core.tools import tool tool def search_user(user_id: str) - str: 根据用户 ID 查用户信息 return fnameAlice, age30, cityBeijing tool def get_weather(city: str) - str: 根据城市查天气 return f{city} 今天晴25°CLLM 看到search_user返回的ToolMessage后会从中提取cityBeijing再决定调用get_weather(Beijing)——这就是上一个工具的结果供下一个工具使用的自然流转。做法二把结果写进自定义 state 字段结构化传递如果工具结果需要以结构化字段的形式显式存下来而不仅仅是躺在 messages 里让 LLM 自己提取工具需要返回一个Command对象来更新 state from langchain.tools import tool, ToolRuntime from langchain.messages import ToolMessage from langgraph.types import Command from typing import TypedDict class MyState(TypedDict): messages: list user_profile: dict # 自定义字段 current_step: str # 流程控制字段 tool def fetch_user_profile( user_id: str, runtime: ToolRuntime[None, MyState] ) - Command: 获取用户画像并写入 state profile {name: Alice, city: Beijing, vip: True} return Command(update{ # 必须回写 ToolMessage保证消息配对 messages: [ToolMessage( contentstr(profile), tool_call_idruntime.tool_call_id )], # 写入自定义字段下一个工具可从 runtime.state 读取 user_profile: profile, current_step: next_stage })下一个工具通过runtime.state读取上一个工具写入的字段 tool def recommend_plan(runtime: ToolRuntime[None, MyState]) - str: 基于上一个工具写入的用户画像做推荐 profile runtime.state.get(user_profile, {}) if not profile: return 未找到用户画像 city profile.get(city) return f为 {city} 的用户推荐本地服务 调用runtime.state.get()时务必提供默认值——首次调用时该字段可能还没初始化直接[]取值会抛 KeyError 导致 Agent 循环崩溃 。做法三跨工具共享运行期依赖用 context如果下一个工具需要的是user_id、API Key、数据库连接​ 这类运行期依赖不是上一个工具的计算结果应该走context而不是 statefrom dataclasses import dataclass from langchain.tools import ToolRuntime, tool dataclass class AppContext: user_id: str db_conn: object api_key: str tool def query_orders(runtime: ToolRuntime[AppContext]) - str: 工具通过 runtime.context 拿到运行期依赖 uid runtime.context.user_id conn runtime.context.db_conn # ... 查询逻辑 return f用户 {uid} 有 3 笔订单调用时传入 context agent.invoke( {messages: 查询我的订单}, contextAppContext(user_idu_123, db_connconn, api_keyxxx) )完整可运行的最小示例from typing import Annotated, TypedDict, List from langchain_core.messages import BaseMessage from langchain_core.tools import tool from langchain.tools import ToolNode, ToolRuntime from langchain.messages import ToolMessage from langgraph.types import Command from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages # 1. 定义带自定义字段的 state class AgentState(TypedDict): messages: Annotated[List[BaseMessage], add_messages] last_result: dict # 存上一个工具的结构化结果 step: str # 2. 工具 A写入 state tool def tool_a(query: str, runtime: ToolRuntime[None, AgentState]) - Command: result {query: query, answer: f答案:{query}的结果} return Command(update{ messages: [ToolMessage( contentstr(result), tool_call_idruntime.tool_call_id )], last_result: result, step: after_a }) # 3. 工具 B读取上一个工具写入的 last_result tool def tool_b(runtime: ToolRuntime[None, AgentState]) - str: prev runtime.state.get(last_result, {}) if not prev: return 没有前置结果 return f基于 {prev.get(answer)} 做进一步处理 # 4. 构建图用 ToolNode 自动处理 ToolRuntime 注入 tools [tool_a, tool_b] tool_node ToolNode(tools) def call_model(state: AgentState): # 这里绑定 tools 给 LLM response model.bind_tools(tools).invoke(state[messages]) return {messages: [response]} graph (StateGraph(AgentState) .add_node(agent, call_model) .add_node(tools, tool_node) .add_edge(START, agent) .add_conditional_edges(agent, lambda s: tools if s[messages][-1].tool_calls else END, [tools, END]) .add_edge(tools, agent) .compile() )⚠️关键提醒如果你想自己写工具执行节点不用预置ToolNode直接tool.ainvoke()时ToolRuntime不会被自动注入runtime参数会变成必填导致校验失败。这种情况下要么改用ToolNode要么手动构造ToolRuntime注入 ——生产环境强烈建议直接用ToolNode。几种场景的选择建议工具结果是给 LLM 看的最常见→ 直接返回靠messagesToolMessage自然流转state 用add_messagesreducer工具结果需要结构化存下来供后续节点/工具读取​ → 返回Command(update{...})写自定义 state 字段跨会话保留用户偏好、历史记忆→ 用runtime.store长期记忆不是 state运行期依赖user_id、连接、密钥→ 走contextinvoke 时传入多轮对话持久化​ → 编译图时加checkpointerMemorySaver()用thread_id隔离 一句话记住Runtime 是管道State 才是仓库。工具结果要供下一个工具用本质是写回 StateRuntime 只是在这一次调用里帮你把 state、context、store 端到工具面前 。如果你的场景是多 Agent 交接handoff还需要把上一个 Agent 的AIMessageToolMessage配对传给下一个 Agent这部分逻辑稍微特殊需要的话告诉我你的具体架构我给一份针对多 Agent 的写法。