Day 15 / 共 20 天 · 阶段4 工具与 Agent

Agent:create_agent 把 Day14 的手写循环,编译成一张 LangGraph 状态图

Day14 结尾你手写了 while 循环:"有工单就执行、回填、再问模型"。今天读 LangChain v1 的 Agent 真身——libs/langchain_v1/langchain/agents/factory.pycreate_agent。三个发现:①它不是一个类,是一个"造图工厂":产出 LangGraph 的 CompiledStateGraph;②图的骨架就两个节点(model ⇄ tools),ReAct 循环全靠两条条件边来回跳;③循环的退出条件,源码注释亲口说了——len(tool_calls) == 0,"这是 Agent 循环的经典退出条件"。读完你会发现:Agent 没有魔法,就是 Day14 + 一张图。

📍 你在 20 天里的位置(阶段4:工具与 Agent · D13-16)
S1 LCEL S2 模型·消息·提示 S3 数据与 RAG D13 Tool 抽象 D14 工具调用 D15 Agent D16 记忆 S5 进阶收官
💡 先用两个类比兜住今天 类比一:Agent 像"带工具箱的实习生",create_agent 是入职培训流程。Day14 里你是"手把手师傅":模型每开一张工单你都亲手执行、亲手回填。Agent 化 = 你把规矩写成流程图贴墙上——"收到任务先思考(model 节点)→ 要用工具就去工具台(tools 节点)→ 拿到结果回来继续思考 → 直到不再需要工具,交差"——实习生照图自己跑,你只看最终结果。类比二:状态图像"棋盘 + 走子规则"。节点是格子(干活的地方),边是走法(去哪儿的规则),state(消息列表)是棋局——每走一步棋局变厚一层,走子规则只看当前棋局最后几步(最后那条 AIMessage 有没有工单)决定下一格。
L01

痛点:手写循环,越写越像一个框架

🤔 痛点Day14 的 while 循环看似三行,产品化时会滚雪球:任务可能要连环多轮工具调用(查天气→根据结果查穿衣建议→再总结);要防死循环(模型来回调同一个工具怎么办);要并行执行多张工单;要在"执行敏感工具前"插一步人工确认;中途崩了要能断点续跑;还想流式看到每一步。把这些全塞进手写 while——恭喜,你在重新发明一个残缺的工作流引擎。
💡 本质:把"循环"从代码升格为"图"LangChain v1 的答案:不写循环,画图。把"思考"和"干活"做成图上两个节点,把"要不要继续"做成条件边,把消息列表做成图的 state。图由 LangGraph 执行——于是持久化(checkpointer)、人工介入(interrupt)、并行分发(Send)、流式(stream)这些能力全是引擎自带的,Agent 逻辑本身保持极简。这就是 ReAct(Reason 思考 + Act 行动)模式的工程化落地:推理和行动交替,图负责调度
L02

create_agent:签名即功能清单

入口在 libs/langchain_v1/langchain/agents/factory.py:808(前面 :740/:762/:786 是类型重载):

# libs/langchain_v1/langchain/agents/factory.py:808
def create_agent(
    model: str | BaseChatModel,                    # 模型:实例或 "openai:gpt-4o" 字符串
    tools: Sequence[BaseTool | Callable | dict[str, Any]] | None = None,  # Day13 的货架
    *,
    system_prompt: str | SystemMessage | None = None,   # 人设
    middleware: Sequence[AgentMiddleware[...]] = (),    # 中间件:循环各切面的钩子
    response_format: ResponseFormat | ... | None = None, # 结构化输出
    state_schema: type[AgentState[ResponseT]] | None = None,
    context_schema: type[ContextT] | None = None,
    checkpointer: Checkpointer | None = None,      # ★持久化:断点续跑/多轮会话
    store: BaseStore | None = None,
    interrupt_before: list[str] | None = None,     # ★在某节点前暂停(人工确认)
    interrupt_after: list[str] | None = None,
    debug: bool = False, name: str | None = None, ...
) -> CompiledStateGraph[AgentState[ResponseT], ...]:   # ★返回值:一张编译好的 LangGraph 图!
    """Creates an agent graph that calls tools in a loop
    until a stopping condition is met."""
返回 CompiledStateGraph★最重要的一行是返回类型:Agent 不是 LangChain 自己发明的运行时,就是一张 LangGraph 图。所以它天生有 invoke/stream/astream,能画 mermaid 图,能挂 checkpointer。
docstring 第一句官方一句话定义 Agent(:828):"calls tools in a loop until a stopping condition is met"——循环调用工具直到满足停止条件。停止条件是什么?L05 见分晓。
middlewarev1 的大特色:把"模型调用前后/工具调用前后/Agent 前后"做成可插拔切面。看 agents/middleware/ 目录:human_in_the_loop.py(人工确认)、model_call_limit.py(防失控限次数)、model_fallback.py(模型降级)……都是现成中间件。
state_schema图的"棋局"默认是 AgentState——核心就是 messages 列表(Day06 的消息体系在这里当了状态)。你可以扩展它塞自己的字段。
L03

搭图:两个节点撑起整个 Agent

往下读工厂内部——先建图(factory.py:1159),再加节点(factory.py:1502):

# libs/langchain_v1/langchain/agents/factory.py:1159(裁剪)
graph: StateGraph[AgentState[ResponseT], ...] = StateGraph(
    state_schema=resolved_state_schema,        # 棋局的形状(messages 等)
    input_schema=input_schema,
    output_schema=output_schema,
    context_schema=context_schema,
)
# ...
# factory.py:1502
graph.add_node("model", RunnableCallable(model_node, amodel_node, trace=False))  # ① 思考节点
if tool_node is not None:
    graph.add_node("tools", tool_node)         # ② 干活节点(ToolNode,来自 LangGraph)
# ...
# factory.py:1620
graph.add_edge(START, entry_node)              # ③ 起点 → 入口(无中间件时就是 "model")
"model" 节点思考中枢:把当前 messages 发给模型,产出新的 AIMessage(可能带工单)。RunnableCallable(model_node, amodel_node) 同时注册同步/异步两个实现——老规矩。
"tools" 节点干活台:ToolNode 是 LangGraph 预制件,收到工单批量执行工具、产出 ToolMessage 列表——Day14 L05 的三行闭环代码,就住在这个节点里
中间件节点是可选花边:1527 之后一大段是把各中间件的 before_agent/before_model/after_model… 钩子也注册成节点、串进主干。没配 middleware 时,图就是最纯的 model ⇄ tools 两节点。
create_agent 产出的图:ReAct 循环 = model ⇄ tools START model 节点 bind_tools + invoke 有工单 → Send("tools", 每单一发) tools 节点 执行工具 → ToolMessage 回执入列 → 回 model 继续想 len(tool_calls)==0 → END 交差 END
图注:橙色边=模型开了工单去干活;绿色边=干完回来继续想;向下的边=没工单了,交差。整个 Agent 就这一圈。
L04

model 节点内部:Day14 的 bind_tools 在这儿被自动调用

model 节点每次执行前,先用 _get_bound_modelfactory.py:1272)现场绑定工具,然后 _execute_model_syncfactory.py:1406)真正调用:

# libs/langchain_v1/langchain/agents/factory.py:1396(裁剪:无结构化输出的标准路径)
if final_tools:
    return (
        request.model.bind_tools(                     # ★Day14 的 bind_tools!
            final_tools, tool_choice=request.tool_choice, **request.model_settings
        ),
        None,
    )
return request.model.bind(**request.model_settings), None   # 没工具就裸调

# factory.py:1406
def _execute_model_sync(request: ModelRequest[ContextT]) -> ModelResponse:
    model_, effective_response_format = _get_bound_model(request)  # ① 绑工具
    messages = request.messages
    if request.system_message:
        messages = [request.system_message, *messages]             # ② system 人设插最前
    output = model_.invoke(messages)                               # ③ ★一次 Day05 的标准调用
    ...
    return ModelResponse(result=messages_list, structured_response=structured_response)
bind_tools 每轮都发生工具清单不是"训练"进模型的,而是每轮请求都随身携带(Day14 讲过 bind 只是挂参数)。也因此中间件可以在运行中动态增删工具——下一轮 final_tools 变了,绑定就变了。
结构化输出的小魔术:1388 附近:如果配了 response_format(ToolStrategy),最终答案也被做成一个"特殊工具",并 tool_choice="any" 强制模型必须调工具——"填表交卷"也走工单通道,输出格式就有了硬保证。
middleware 包夹model_nodefactory.py:1433)构造 ModelRequest 后,会穿过所有 wrap_model_call 中间件再到 _execute_model_sync——重试、降级、限流都在这层包夹里发生,核心执行逻辑保持干净。
L05

循环边:ReAct 的"继续还是收工"写在两个函数里

先看 model 走完之后去哪——_make_model_to_tools_edgefactory.py:1840):

# libs/langchain_v1/langchain/agents/factory.py:1846(裁剪)
def model_to_tools(state: dict[str, Any]) -> str | list[Send] | None:
    ...
    last_ai_message, tool_messages = _fetch_last_ai_and_tool_messages(state["messages"])
    ...
    # 3. If the model hasn't called any tools, exit the loop
    # this is the classic exit condition for an agent loop      ← 源码原话!
    if len(last_ai_message.tool_calls) == 0:
        return end_destination                        # ★没工单 → 收工

    pending_tool_calls = [c for c in last_ai_message.tool_calls
                          if c["id"] not in tool_message_ids     # 还没回执的工单
                          and c["name"] not in structured_output_tools]
    if pending_tool_calls:
        return [Send("tools", [tool_call]) for tool_call in pending_tool_calls]  # ★每单一发,并行执行

再看 tools 干完之后去哪——_make_tools_to_model_edgefactory.py:1921):

# libs/langchain_v1/langchain/agents/factory.py:1928(裁剪)
def tools_to_model(state: dict[str, Any]) -> str | None:
    last_ai_message, tool_messages = _fetch_last_ai_and_tool_messages(state["messages"])
    ...
    # 2. Exit condition: All executed tools have return_direct=True
    if client_side_tool_calls and all(
        tool_node.tools_by_name[c["name"]].return_direct for c in client_side_tool_calls):
        return end_destination                        # 工具声明"结果直出"→ 不再回模型

    # 3. Exit condition: A structured output tool was executed
    if any(t.name in structured_output_tools for t in tool_messages):
        return end_destination                        # 已经"填表交卷"→ 收工

    # 4. Default: Continue the loop
    return model_destination                          # ★默认:带着回执回去继续想
len(tool_calls) == 0★整个 Agent 的灵魂判断,注释亲口盖章"classic exit condition"。循环停不停,由模型自己决定——它觉得信息够了就不开工单,图自然走向 END。Day14 你手写的 while 条件,在这里变成了一条条件边。
Send("tools", [tool_call])模型一口气开 3 张工单?Send 是 LangGraph 的"分发信封"——每张工单单独发一份给 tools 节点,并行执行,回执各自写回 state。这是手写 for 循环串行执行拿不到的免费加速。
pending 去重只派发"还没有对应回执"的工单(对比 tool_message_ids)——恢复执行/重放时不会重复干活。工单号在这里第三次立功。
return_directDay13 埋的字段兑现:工具声明 return_direct=True,结果不再交回模型润色,直接作为最终输出——适合"查询类"工具省一轮模型调用。
⚠️ 坑:模型不收工怎么办?退出条件握在模型手里,就存在"它一直开工单"的风险(工具结果不理想 → 换参数再试 → 无限循环烧钱)。生产上必须加护栏:中间件 model_call_limit.py 限制模型调用次数、LangGraph 自带 recursion_limit(默认 25 步)兜底超限报错。给实习生发工牌,也要给他定下班时间。
L06

与 LangGraph 的关系 + 今日小结

今天 import 痕迹已经说明一切(factory.py:26from langgraph.graph.state import StateGraph)。三兄弟分工:

项目角色在 Agent 里的贡献
langchain_core标准件库BaseChatModel/消息/Tool/Runnable——图上每个节点里跑的"零件"
langgraph执行引擎StateGraph/条件边/Send/checkpointer/interrupt——"循环"本身
langchain(v1)成品组装厂create_agent + middleware:把上面两层拼成一句话可用的 Agent
📝 真实值:一次 Agent 运行的完整轨迹 agent = create_agent("openai:gpt-4o", tools=[get_weather], system_prompt="你是天气助手")agent.invoke({"messages": [HumanMessage("上海和北京哪边更适合明天户外跑步?")]}) → START→model:模型开 2 张工单 get_weather(city="上海") id=call_a1get_weather(city="北京") id=call_b2 → 条件边发出 2 个 Send("tools", ...) 并行执行 → 2 条 ToolMessage("上海 26 度晴 / 北京 33 度高温")入列 → tools_to_model 返回 "model" → 模型看回执:tool_calls=[],直接回答"推荐上海:26 度晴天更适合跑步,北京明天 33 度高温请避免户外运动" → len(tool_calls)==0 → END。最终 state["messages"] 共 5 条:Human + AI(2 工单) + 2×Tool + AI(结论)。

👶 小白:那我到底该学 LangChain 的 create_agent,还是直接学 LangGraph 自己画图?

👨‍🏫 老师:一个是"整车",一个是"底盘+发动机"。create_agent 是预装好的整车——标准 ReAct 循环 + 常用中间件,一行起步,覆盖 80% 的"工具调用型"场景。当你的流程不再是"模型⇄工具"这个形状(多 Agent 协作、复杂审批流、自定义状态机),就该下探到 LangGraph 手工画图——今天读的 factory.py 本身就是最好的"官方画图示范":你未来手写图时,add_node/条件边/Send 的用法照着它抄就行。先开整车上路,再学改装。

🧠 今天你应该能回答

  • create_agent 返回什么类型?意味着什么?(CompiledStateGraph——Agent 就是一张 LangGraph 图,天生有 invoke/stream/持久化)
  • 图的最小骨架?(model 节点思考 + tools 节点干活 + 两条条件边来回跳)
  • ReAct 循环的经典退出条件?(最后一条 AIMessage 的 len(tool_calls) == 0,源码注释原话)
  • 多张工单怎么执行?(条件边返回多个 Send("tools", [tool_call]),并行)
  • 除"没工单"外还有哪些退出口?(return_direct 工具直出 / 结构化输出工具已交卷)
  • LangChain 与 LangGraph 的关系?(core 出零件,langgraph 出引擎,langchain v1 组装成品)

✋ 10 分钟动手

cd /Users/bitmart/work/codes/github/AI_WORK/langchain

# 1. 工厂签名与"造图"证据
sed -n '808,830p'  libs/langchain_v1/langchain/agents/factory.py   # create_agent 签名
sed -n '1159,1170p' libs/langchain_v1/langchain/agents/factory.py  # StateGraph(...)
grep -n 'add_node("model"\|add_node("tools"' libs/langchain_v1/langchain/agents/factory.py

# 2. 循环的两条边(重点读注释)
sed -n '1846,1892p' libs/langchain_v1/langchain/agents/factory.py  # model_to_tools
sed -n '1928,1954p' libs/langchain_v1/langchain/agents/factory.py  # tools_to_model

# 3. 现成中间件目录:每个文件名都是一个生产级痛点
ls libs/langchain_v1/langchain/agents/middleware/
明日预告 · Day 16:Agent 会干活了,但每次 invoke 都是"金鱼记忆"——上一轮聊的它全忘。明天读 runnables/history.pyRunnableWithMessageHistorychat_history.py:对话历史怎么按 session_id 存取、怎么自动注入链的输入,让你的 Agent 记住"我刚才说过我在上海"。
← Day 14 工具调用闭环 Day 16 · 记忆与对话历史 →