Agent:create_agent 把 Day14 的手写循环,编译成一张 LangGraph 状态图
Day14 结尾你手写了 while 循环:"有工单就执行、回填、再问模型"。今天读 LangChain v1 的 Agent 真身——libs/langchain_v1/langchain/agents/factory.py 的 create_agent。三个发现:①它不是一个类,是一个"造图工厂":产出 LangGraph 的 CompiledStateGraph;②图的骨架就两个节点(model ⇄ tools),ReAct 循环全靠两条条件边来回跳;③循环的退出条件,源码注释亲口说了——len(tool_calls) == 0,"这是 Agent 循环的经典退出条件"。读完你会发现:Agent 没有魔法,就是 Day14 + 一张图。
痛点:手写循环,越写越像一个框架
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 的消息体系在这里当了状态)。你可以扩展它塞自己的字段。搭图:两个节点撑起整个 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 两节点。model 节点内部:Day14 的 bind_tools 在这儿被自动调用
model 节点每次执行前,先用 _get_bound_model(factory.py:1272)现场绑定工具,然后 _execute_model_sync(factory.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_node(factory.py:1433)构造 ModelRequest 后,会穿过所有 wrap_model_call 中间件再到 _execute_model_sync——重试、降级、限流都在这层包夹里发生,核心执行逻辑保持干净。循环边:ReAct 的"继续还是收工"写在两个函数里
先看 model 走完之后去哪——_make_model_to_tools_edge(factory.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_edge(factory.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 步)兜底超限报错。给实习生发工牌,也要给他定下班时间。与 LangGraph 的关系 + 今日小结
今天 import 痕迹已经说明一切(factory.py:26:from langgraph.graph.state import StateGraph)。三兄弟分工:
| 项目 | 角色 | 在 Agent 里的贡献 |
|---|---|---|
langchain_core | 标准件库 | BaseChatModel/消息/Tool/Runnable——图上每个节点里跑的"零件" |
langgraph | 执行引擎 | StateGraph/条件边/Send/checkpointer/interrupt——"循环"本身 |
langchain(v1) | 成品组装厂 | create_agent + middleware:把上面两层拼成一句话可用的 Agent |
agent = create_agent("openai:gpt-4o", tools=[get_weather], system_prompt="你是天气助手");agent.invoke({"messages": [HumanMessage("上海和北京哪边更适合明天户外跑步?")]}) → START→model:模型开 2 张工单 get_weather(city="上海") id=call_a1、get_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/
runnables/history.py 的 RunnableWithMessageHistory 和 chat_history.py:对话历史怎么按 session_id 存取、怎么自动注入链的输入,让你的 Agent 记住"我刚才说过我在上海"。