Day 54 / 共 60 天 · 阶段 9 预制件与流式

逐行①：agent 节点里，模型是怎么被"喂"的

昨天看清了骨架：agent 节点调模型、tools 节点执行工具。今天把 agent 节点掰开揉碎——从"四种 prompt 怎么统一"到"要不要帮你 bind_tools"，再到 call_model 逐行、以及 remaining_steps 如何拦住死循环。这一节点是 Agent 的"大脑入口"，每一行都在为"让模型正确地看到该看的东西"服务。

📍 阶段 9 · 预制件与流式（6 天）你在这里

D53 总览→ D54 逐行①→ D55 ToolNode→ D56 校验→ D57 流式5模式→ D58 流式底层

💡 用一个类比先兜住今天 agent 节点像给秘书交代活儿。交代前你要：把"总要求"（系统 prompt）钉在最前面（L01）；确认秘书手里有没有工具说明书——没有就发一份（bind_tools，L02）；把"总要求 + 工具说明书 + 模型"打包成一个固定流程（static_model，L03）；每次派活先核对历史交接单是否齐全（校验，L04）；最后才让秘书干活、拿回结果（call_model，L05）；还得盯着别让秘书陷入"反复请示"的无限循环（remaining_steps，L06）。

L01

_get_prompt_runnable：四种 prompt 统一成一个 Runnable

🤔 痛点用户可能传字符串 "你是助手"、传 SystemMessage、传一个函数 lambda state: [...]、甚至传一个 Runnable。下游 call_model 不可能为每种写四套逻辑。怎么"抹平差异"？

答案在 chat_agent_executor.py:137：把四种都包成同一个 Runnable：

# chat_agent_executor.py:137
def _get_prompt_runnable(prompt) -> Runnable:
    if prompt is None:                                     # ① 不给：直接取 messages
        prompt_runnable = RunnableCallable(
            lambda state: _get_state_value(state, "messages"), name=PROMPT_RUNNABLE_NAME)
    elif isinstance(prompt, str):                          # ② 字符串：拼成 SystemMessage 放最前
        _system_message = SystemMessage(content=prompt)
        prompt_runnable = RunnableCallable(
            lambda state: [_system_message] + _get_state_value(state, "messages"), ...)
    elif isinstance(prompt, SystemMessage):                # ③ 已是 SystemMessage：同上
        prompt_runnable = RunnableCallable(
            lambda state: [prompt] + _get_state_value(state, "messages"), ...)
    elif inspect.iscoroutinefunction(prompt):              # ④ 异步函数
        prompt_runnable = RunnableCallable(None, prompt, ...)
    elif callable(prompt):                                 # ⑤ 同步函数
        prompt_runnable = RunnableCallable(prompt, ...)
    elif isinstance(prompt, Runnable):                     # ⑥ 已是 Runnable：原样
        prompt_runnable = prompt
    else:
        raise ValueError(f"Got unexpected type for `prompt`: {type(prompt)}")
    return prompt_runnable

prompt is None不给 prompt，就返回一个"把 state 里 messages 原样取出"的 Runnable——啥都不加。

isinstance(prompt, str)字符串：包成 SystemMessage，用 [系统消息] + messages 放最前面。这是最常见用法。

iscoroutinefunction / callable函数：直接当 Runnable 的主体（异步走第二参数）。你能在函数里任意定制发给模型的消息。

isinstance(prompt, Runnable)已经是 Runnable 就原样返回——最灵活的口子。

💡 设计取舍①：为什么费劲统一成 Runnable，而不是 if/else 直接判断？因为下游要做 prompt_runnable | model（管道拼接，L03）。只有 Runnable 才能用 | 组合。把差异消化在"入口处一次性归一化"，之后所有代码都只面对一个统一接口——这是 LangChain 生态"万物皆 Runnable"哲学的体现。代价是多包了一层 RunnableCallable，但换来下游零分支。

👶 一句话不管你怎么写 prompt，最后都变成一个"输入 state、输出一串消息"的小函数。

L02

_should_bind_tools：帮你判断"工具绑了没"

模型要"知道"有哪些工具可调，得先 bind_tools。但用户可能已经自己绑过了。重复绑会出错，于是有 chat_agent_executor.py:173 做智能判断：

# chat_agent_executor.py:173
def _should_bind_tools(model, tools, num_builtin=0) -> bool:
    if isinstance(model, RunnableSequence):            # 从管道里挑出真正的模型步
        model = next((step for step in model.steps
            if isinstance(step, (RunnableBinding, BaseChatModel))), model)
    if not isinstance(model, RunnableBinding):
        return True                                    # 没绑过任何东西 → 需要绑
    if "tools" not in model.kwargs:
        return True                                    # 绑了但没绑 tools → 需要绑
    bound_tools = model.kwargs["tools"]
    if len(tools) != len(bound_tools) - num_builtin:   # 数量对不上 → 直接报错
        raise ValueError("Number of tools in the model.bind_tools() and tools passed"
            " to create_react_agent must match ...")
    tool_names = set(tool.name for tool in tools)
    ...                                                # 逐个核对名字（OpenAI/Anthropic 两种格式）
    if missing_tools := tool_names - bound_tool_names:
        raise ValueError(f"Missing tools '{missing_tools}' in the model.bind_tools()")
    return False                                       # 已正确绑齐 → 不用再绑

not RunnableBinding → True模型是"裸模型"（没 .bind(...) 过），肯定要帮它绑工具。

"tools" not in kwargs → True绑过别的（比如温度参数）但没绑 tools，也要补绑。

len(tools) != len(bound)-num_builtin数量校验：你自己绑的工具数和传进来的对不上，立刻报错——防止"以为绑了其实漏了"。

missing_tools → raise名字校验：逐个核对工具名（兼容 OpenAI 的 function.name 和 Anthropic 的 name 两种结构），有缺失就报错。

return False全对上了 → 你已经正确绑好，函数不再重复绑，尊重你的配置（比如你可能绑了额外参数）。

⚠️ 边界：为什么不无脑再 bind 一次省事？因为你可能通过 model.bind_tools(tools, tool_choice="required") 传了额外的绑定参数（强制调工具、并行开关等）。如果框架无脑再 bind_tools(tools) 一次，会覆盖掉你的定制。所以源码宁可写这一大段校验，也要在"你已正确绑好"时 return False 放手不管。这是"框架不越俎代庖"的克制。

L03

组装 static_model：prompt | model 一条管道

前两步的成果在这里合体（chat_agent_executor.py:567）：

# chat_agent_executor.py:567
if not is_dynamic_model:
    if isinstance(model, str):                          # "openai:gpt-4" 这种字符串
        from langchain.chat_models import init_chat_model
        model = cast(BaseChatModel, init_chat_model(model))   # 转成真模型实例
    if (_should_bind_tools(model, tool_classes, num_builtin=len(llm_builtin_tools))
            and len(tool_classes + llm_builtin_tools) > 0):
        model = model.bind_tools(tool_classes + llm_builtin_tools)   # 需要就绑
    static_model = _get_prompt_runnable(prompt) | model   # ★ prompt 管道接模型
else:
    static_model = None                                  # 动态模型：运行时再造（L07）

isinstance(model, str)你传 "openai:gpt-4o" 这种字符串，用 init_chat_model 转成真实模型对象（需装 langchain）。

if _should_bind_tools(...) and 有工具L02 判断"该绑"且确实有工具，才 bind_tools。内建工具（dict）也一起绑给模型。

_get_prompt_runnable(prompt) | model今天的高潮：把 L01 的 prompt Runnable 用 | 接上模型，得到 static_model。喂它一个 state，它自动"加系统提示 → 调模型 → 出 AIMessage"。

is_dynamic_model → None如果 model 是个"根据 state 选模型"的函数，此刻无法定死，留到运行时（L07 的 _resolve_model）。

💡 本质：一个 | 省掉一个节点朴素实现可能会为"加 prompt"单独建一个节点。这里用 Runnable 的管道 prompt | model 把它压进 agent 节点内部——图上只有一个 agent 节点，prompt 处理是它的"前置微操"。图越简单，越好画、好调试、好流式。

L04

_get_model_input_state：取消息 + 校验历史完整

调模型前，得先把"该给模型看的消息"准备好并校验。chat_agent_executor.py:636：

# chat_agent_executor.py:636
def _get_model_input_state(state):
    if pre_model_hook is not None:
        messages = (_get_state_value(state, "llm_input_messages")
                    ) or _get_state_value(state, "messages")   # 钩子可能改写了输入
        ...
    else:
        messages = _get_state_value(state, "messages")
    if messages is None:
        raise ValueError(error_msg)
    _validate_chat_history(messages)                    # ★ 校验：每个 tool_call 都有对应结果
    state["messages"] = messages
    return state

校验逻辑 chat_agent_executor.py:243：

# chat_agent_executor.py:243
def _validate_chat_history(messages) -> None:
    all_tool_calls = [tc for m in messages if isinstance(m, AIMessage) for tc in m.tool_calls]
    tool_call_ids_with_results = {m.tool_call_id for m in messages if isinstance(m, ToolMessage)}
    tool_calls_without_results = [tc for tc in all_tool_calls
        if tc["id"] not in tool_call_ids_with_results]
    if not tool_calls_without_results:
        return                                          # 全都有结果 → 通过
    raise ValueError(create_error_message(              # 有 tool_call 没结果 → 报错
        message="Found AIMessages with tool_calls that do not have a corresponding ToolMessage..."
        , error_code=ErrorCode.INVALID_CHAT_HISTORY))

llm_input_messages or messages配了 pre_model_hook 时，钩子可能把"给模型看的消息"写进 llm_input_messages（比如压缩后的）——优先用它，否则用完整 messages。

all_tool_calls收集历史里所有 AIMessage 发起过的 tool_call。

tool_call_ids_with_results收集所有已有结果的 tool_call_id（来自 ToolMessage）。

without_results → raise核心校验：如果有 tool_call 找不到对应 ToolMessage，直接报错。因为绝大多数模型提供商强制要求"每个工具调用必须有结果配对"。

⚠️ 边界：为什么这个校验必须有？想象你手动 update_state（D44 时间旅行）删掉了一条 ToolMessage，但留着发起它的 AIMessage.tool_calls。下一轮调模型时，OpenAI/Anthropic 会直接报 400 错误："tool_call 没有对应的 tool 结果"。框架提前在本地校验并抛出可读的错误信息（含前几个孤儿 tool_call），比让你去看云端 API 的晦涩 400 友好得多。这是"把远端错误左移到本地、把晦涩错误翻译成人话"的防御式编程。

L05

call_model：agent 节点的主函数逐行

万事俱备，看主函数 chat_agent_executor.py:661：

# chat_agent_executor.py:661
def call_model(state, runtime, config):
    if is_async_dynamic_model:                          # ① 异步动态模型却被同步调用 → 报错
        raise RuntimeError("Async model callable provided but agent invoked synchronously...")
    model_input = _get_model_input_state(state)         # ② 取消息+校验（L04）
    if is_dynamic_model:
        dynamic_model = _resolve_model(state, runtime)  # 动态：运行时造模型
        response = cast(AIMessage, dynamic_model.invoke(model_input, config))
    else:
        response = cast(AIMessage, static_model.invoke(model_input, config))  # ③ 调模型
    response.name = name                                # ④ 给消息盖上 agent 名字
    if _are_more_steps_needed(state, response):         # ⑤ 步数不够 → 返回"抱歉"占位
        return {"messages": [AIMessage(id=response.id,
            content="Sorry, need more steps to process this request.")]}
    return {"messages": [response]}                     # ⑥ 正常：追加这条 AIMessage

is_async_dynamic_model → RuntimeError你给的模型是异步函数，却用 agent.invoke()（同步）跑——提前拦住，提示改用 ainvoke。防止运行到一半才崩。

model_input = _get_model_input_stateL04 的准备+校验。此时 state["messages"] 已就绪。

static_model.invoke(model_input, config)真正调模型：static_model 是 prompt|model，一步到位加提示、调模型、出 AIMessage。

response.name = name给这条 AIMessage 打上 agent 的名字——多 Agent 系统里用来区分"谁说的话"。

return {"messages": [response]}返回一个列表，因为 add_messages reducer 会把它追加到现有 messages 尾部（D08）。

👶 一句话call_model = 校验历史 → 让模型说话 → 盖章 → 追加。就这么朴素。

L06

remaining_steps：拦住"无限调工具"的死循环

🤔 痛点模型可能抽风，一直要求调工具、永远不给最终答案——ReAct 的环就转不停了。除了 D17 的 recursion_limit 硬上限，还有更"聪明"的软拦截吗？

chat_agent_executor.py:620：

# chat_agent_executor.py:620
def _are_more_steps_needed(state, response) -> bool:
    has_tool_calls = isinstance(response, AIMessage) and response.tool_calls
    all_tools_return_direct = (
        all(call["name"] in should_return_direct for call in response.tool_calls)
        if isinstance(response, AIMessage) else False)
    remaining_steps = _get_state_value(state, "remaining_steps", None)
    if remaining_steps is not None:
        if remaining_steps < 1 and all_tools_return_direct:
            return True                                 # 只剩不到1步、且全是直接返回工具
        elif remaining_steps < 2 and has_tool_calls:
            return True                                 # 只剩不到2步、还想调工具
    return False

remaining_steps运行时自动注入的"还能走几步"配额（AgentState 里那个字段，D53）。每过一个超步递减。

< 2 and has_tool_calls → True关键：模型还想调工具，但剩余步数不够跑完"工具→再回模型"这一来一回（要 2 步），提前判定"步数不够"。

< 1 and all_return_direct更细的边界：如果工具全是 return_direct（结果即答案，不用再回模型），那只需 1 步，门槛放宽到 <1。

返回 True 的后果（见 L05）call_model 会丢弃模型这条"还要调工具"的响应，改返回一句 "Sorry, need more steps..." 正常结束——优雅收尾，而非崩溃。

💡 设计取舍②：为什么不直接让它撞 recursion_limit 报错？recursion_limit（D17）是硬刹车：超了直接抛 GraphRecursionError，用户看到的是一个异常。而 remaining_steps 是软着陆：在"再走一步就会撞墙"时提前一步优雅停下，返回一句人能看懂的话。对终端用户体验来说，"抱歉，这个问题需要更多步骤"远好于"程序崩溃了"。两者并存：软拦截兜日常，硬上限兜极端。这是"给失败一个体面的出口"的设计。

L07

动态模型 _resolve_model + 今日小结

最后补上"动态选模型"这条支线（chat_agent_executor.py:599）：

# chat_agent_executor.py:599
def _resolve_model(state, runtime) -> LanguageModelLike:
    if is_dynamic_model:
        return _get_prompt_runnable(prompt) | model(state, runtime)  # 运行时调 model(state,runtime)
    else:
        return static_model

💡 本质：动态模型 = 把"选哪个模型"推迟到运行时静态模型在建图时就 prompt|model 定死（L03）。但如果你想"简单问题用便宜模型、复杂问题用贵模型"，就传一个 model(state, runtime)->BaseChatModel 函数。此时 static_model=None，每次 call_model 现场调 _resolve_model 拼 prompt | model(state,runtime)。同样的管道套路，只是把 model 从"值"换成"运行时求值的函数"。

图注：prompt 归一化 + model 绑工具，用 | 合成一个 static_model。

图注：call_model 的五道关卡，最后二选一（优雅停 / 正常追加）。

                🧠 今天你应该能回答
                四种 prompt 为什么都要包成 Runnable？（为了能用 | 和 model 拼管道）
_should_bind_tools 何时返回 False？（你已正确绑齐工具，框架不重复绑以免覆盖你的定制）
static_model 是什么？（prompt_runnable | model，一个"state→AIMessage"的管道）
_validate_chat_history 拦什么？（有 tool_call 却没配对 ToolMessage，本地提前报错）
remaining_steps 和 recursion_limit 区别？（软着陆优雅停 vs 硬上限抛异常）
动态模型和静态模型的差别？（选模型的时机：建图时定死 vs 运行时求值）

            

✋ 10 分钟动手

# 1. 三段核心逐行读
sed -n '137,170p' libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py   # prompt 四合一
sed -n '620,660p' libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py   # 步数门槛
sed -n '661,700p' libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py   # call_model

# 2. 亲手触发 remaining_steps 软着陆
python - <<'PY'
from langgraph.prebuilt import create_react_agent
from langchain_core.tools import tool
@tool
def loop_tool()->str:
    "总让模型想继续调用"
    return "keep going"
# 传一个总是要求继续调工具的假模型时，会看到 "Sorry, need more steps..."
print("看 chat_agent_executor.py:620 的门槛：<2 步且还要调工具就停")
PY

明天预告 · Day 55：进 tools 节点内部——ToolNode 源码。看它如何用线程池并行执行多个工具、如何把各种异常转成 ToolMessage 喂回模型（而不是崩溃）、以及 handle_tool_errors 的四种配置形态。

← Day 53 总览 Day 55 · ToolNode 源码 →