第 4 篇收尾留了张欠条：手动执行循环是 agent 的骨架，但每个应用重抄一遍并不体面，create_agent 会把它收编成一行；第 5 篇讲完链式编排，又把话头递到了 agent 门前。本篇是《LangChain 从入门到实战》第 6 篇，主角就是 create_agent：先把第 4 篇的 while 循环折算成一行调用，再用 model、tools、system_prompt 三个参数搭起最小代理，重点拆 checkpointer 配 thread_id 的多轮记忆，然后说清 result 怎么读、stream 怎么调试，最后以适用边界收束——它不够用时，路通向哪里。
1、从手动循环到一行成品
先定口径：create_agent 由主包提供（当前 1.4.x，见 PyPI 主页），从 langchain.agents 导入；返回值不是函数，而是一张编译好的 LangGraph 图，安装主包时 langgraph 会作为依赖一并装上，行为细节以 官方 Agents 文档为准。
老教程里的
create_react_agent住在langgraph.prebuilt，1.0 起推荐入口收拢到主包的create_agent——见到旧导入，直接按新入口改写。
第 4 篇用 14 行 while 循环拼出“提议、执行、回填”的闭环；get_weather 工具与 model 原样搬来，同一件事的成品写法如下：
from langchain.agents import create_agent
agent = create_agent(model, [get_weather]) # model 沿用第 4 篇
result = agent.invoke(
{"messages": [{"role": "user", "content": "杭州今天多少度？"}]})
print(result["messages"][-1].content)
解读：循环没有消失，只是搬进了库函数——create_agent 内部用 LangGraph 搭了一张“模型节点加工具节点”的双节点循环图：模型节点的输出带 tool_calls 就路由去工具节点，执行完回到模型节点，直到不再发起调用。手动循环里的每件活在成品里的新去处如下：
| 手动循环里的活 | 成品里由谁干 |
|---|---|
| tool_map 建索引与分发 | 工具节点按注册名分发 |
| tool_calls 判断终止 | 条件边路由到结束 |
| messages 手动追加 | 状态字段自动累积 |
| 最大轮数自己补 | recursion_limit 步数上限 |
第 4 篇练习 2 让你给 while 循环手补轮数上限，成品的兜底是 recursion_limit——超限抛 GraphRecursionError，1.0.6 起默认 1000，工具循环不会真的无限转。输入形态顺带看清：invoke 吃的是初始 State，messages 键装消息列表，{"role": "user", ...} 形式的字典会被规范成 HumanMessage。
2、最小代理与系统提示
三个参数撑起九成场景。model 接受 init_chat_model 初始化的实例，也接受 "openai:gpt-6.1-terra" 这样的字符串简写；tools 是 @tool 工具列表，第 4 篇的装饰器语法原样适用；system_prompt 给整段会话定人设与规矩，接受字符串、SystemMessage 或接收状态的函数。完整示例如下：
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
@tool
def get_weather(city: str) -> str:
"""查询指定城市实时天气，返回天气现象与摄氏温度。"""
data = {"杭州": "晴，22 度", "北京": "多云，18 度"}
return data.get(city, "暂无该城市数据")
model = init_chat_model("openai:gpt-6.1-terra", temperature=0)
agent = create_agent(
model,
[get_weather],
system_prompt="你是天气播报员：查到数据用一句话播报，查不到就明说，不要编造。",
)
result = agent.invoke(
{"messages": [{"role": "user", "content": "杭州今天天气如何？"}]})
print(result["messages"][-1].content)
一次典型运行，print 出的是“杭州今天晴，气温 22 摄氏度”。解读三点。模型第一步发起 get_weather 调用、工具节点执行回填、模型第二步产出播报，两步全部发生在 invoke 内部，调用方只看到最终回答。系统提示里“查不到就明说”是给循环上的第一道护栏：工具回填“暂无该城市数据”后，模型照规矩承认，而不是硬编一条。第 4 篇说工具描述决定调用质量，系统提示则决定拿到结果后怎么用——两层提示词一个管路由、一个管收尾，要配合着写。
3、给 agent 接上记忆
手动循环还有个没解决的缺口：每次调用 messages 都从零开始，第二轮对话要自己把历史搬进列表。create_agent 默认同样如此——没有 checkpointer 的 agent 每次调用都失忆。失忆现场如下：
agent = create_agent(model, [get_weather],
system_prompt="你是天气播报员。")
r1 = agent.invoke({"messages": [
{"role": "user", "content": "我叫阿策，请记住。"}]})
r2 = agent.invoke({"messages": [
{"role": "user", "content": "我叫什么名字？"}]})
print(r2["messages"][-1].content) # 抱歉，我不知道你的名字
两次 invoke 互不相干：名字躺在 r1 的历史里，r2 的 messages 只有一条新问题。接上记忆只需两处改动：
from langgraph.checkpoint.memory import InMemorySaver
agent = create_agent(model, [get_weather],
system_prompt="你是天气播报员。",
checkpointer=InMemorySaver()) # 改动一：存档器
config = {"configurable": {"thread_id": "user-001"}} # 改动二：线程号
r1 = agent.invoke({"messages": [
{"role": "user", "content": "我叫阿策，请记住。"}]}, config)
r2 = agent.invoke({"messages": [
{"role": "user", "content": "我叫什么名字？"}]}, config)
print(r2["messages"][-1].content) # 你叫阿策
解读四点。checkpointer 是检查点存档器，agent 每步跑完把状态落档，下次调用先取回再继续。thread_id 标识一段会话，官方口径是用 thread_id 持久化会话历史，前提是配好 checkpointer——两个参数配套出现，缺一个记忆都不成立。同一 thread_id 复用同一份历史，换个值就是全新对话，多用户隔离靠的就是这个键。InMemorySaver 只存进程内存，重启即失，生产要换 SqliteSaver、PostgresSaver 这类落盘实现，见 持久化文档。记忆的流转如下图：
图后一句解读：存档的不是摘要而是完整消息列表，第 4 节读 result 时能亲眼看到两个轮次的全部消息；至于长对话的压缩与跨线程的长期记忆，是下一篇记忆专篇的课题。
4、result 的正确读法
result 是跑完一轮的最终 State，核心键就是 messages——带 checkpointer 时装的是整个线程的累计历史，不带时是本次调用的历史。一次“查天气”的典型轨迹如下：
| 顺序 | 消息类型 | 承载内容 |
|---|---|---|
| 1 | HumanMessage | 用户问题 |
| 2 | AIMessage | tool_calls 调用请求 |
| 3 | ToolMessage | 工具执行结果回填 |
| 4 | AIMessage | 最终回答 |
两条取值要领：最终回答取最后一条消息的 content，即 result["messages"][-1].content，取 [0] 拿到的是用户自己的问题；排查问题要看中间轨迹，遍历写法如下：
for msg in result["messages"]:
name = type(msg).__name__
calls = getattr(msg, "tool_calls", None)
if calls: # AIMessage 发起调用
for tc in calls:
print(name, "调用", tc["name"], tc["args"])
elif name == "ToolMessage": # 工具结果回填
print(name, "结果", msg.content)
else: # 问题与最终回答
print(name, msg.content)
一次典型运行的输出如下：
HumanMessage 杭州今天天气如何？
AIMessage 调用 get_weather {'city': '杭州'}
ToolMessage 结果 杭州：晴，22 度
AIMessage 杭州今天晴，气温 22 摄氏度。
解读：这段遍历把第 4 篇的 tool_calls 字段与 ToolMessage 回填串成了完整时间线——agent 没有引入新消息类型，用的全是前几篇见过的消息对象，抽象换的是流水线、不换零件。带 checkpointer 后列表会越积越长，第二轮 invoke 的 messages 里第一轮的问题与回答都在场：记忆不是魔法，就是把这份列表原样再喂给模型，上下文窗口与 token 成本的账留到记忆专篇细算。
5、用 stream 看执行轨迹
invoke 只给终态，过程是黑盒；调试时想知道模型到底调了几次工具、参数是什么，用 stream 的 updates 模式——它在每个节点跑完吐一次状态增量：
config = {"configurable": {"thread_id": "debug-1"}}
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "杭州今天多少度？"}]},
config,
stream_mode="updates",
):
for node, update in chunk.items(): # 节点名到状态增量
msg = update["messages"][-1]
calls = getattr(msg, "tool_calls", None)
print(f"[{node}] {type(msg).__name__} {calls or msg.content}")
一次典型运行的输出如下：
[model] AIMessage [{'name': 'get_weather', 'args': {'city': '杭州'}, ...}]
[tools] ToolMessage 杭州：晴，22 度
[model] AIMessage 杭州今天晴，气温 22 摄氏度。
解读四点。chunk 的形状是“节点名到该节点状态增量”的字典，节点名固定为 model 与 tools——它们是内部图的注册名。从段数能直接读出工具链长度：两段 model 中间夹一段 tools 是一次“调用、回填、作答”，第 4 篇的两步工具链在这里会打出五段。updates 看的是步进，要做逐 token 的打字机效果换 stream_mode="messages"，两种模式的差异见 流式文档。调试轮次同样会写进记忆，干净的做法是像示例那样单开一个 thread_id。
6、create_agent 的边界
成品代理的形状是固定的：一个模型、一组工具、一张双节点循环图。官方留了 state_schema 参数（传入 AgentState 的子类）往状态里加自定义字段，但图的拓扑改不了。三类需求的对比如下：
| 需求形态 | 顺手工具 | 说明 |
|---|---|---|
| 固定管道跑一遍 | LCEL 链 | 第 5 篇，无循环三行成型 |
| 单 agent 自主工具循环 | create_agent | 本篇主角 |
| 定制流程与多 agent 协作 | LangGraph | 手搓图，控制权全在手里 |
展开第三行：要在工具循环外加固定工序（先检索再汇总的强制顺序）、要多个 agent 分工协作、要在关键动作前插入人工审批的中断、要节点级重试与并行——都得自己搭图。而搭图的语法正是 create_agent 内部在用的那套：模型节点、工具节点、条件边、回边。本站后续的《LangGraph 从入门到实战》（Agent 专题收官系列）专讲这件事，create_agent 内部那张图会在其中一篇被逐行拆开——学完那一篇，这个一行成品对你不再有黑盒。回收第 5 篇的口诀再补半句：链是声明出来的，agent 是循环出来的，循环用成品，图自己搭。
7、实战带记忆的调研助手
把三件能力拼起来：两个工具加 checkpointer 记忆加多轮追问。场景是内部文档调研——一个工具检索文档、一个工具记笔记，助手跨轮次记住调研上下文。工具定义如下：
from langchain_core.tools import tool
notes = [] # 笔记本，进程内共享
@tool
def search_docs(keyword: str) -> str:
"""按关键词检索内部技术文档，返回标题与摘要。"""
docs = {
"记忆": "《会话记忆设计》：用 checkpointer 按线程存取完整历史。",
"工具": "《工具描述指南》：描述要写清做什么、输入输出与边界。",
}
return docs.get(keyword, "未检索到相关文档")
@tool
def save_note(text: str) -> str:
"""把一条调研要点写入笔记，返回确认信息。"""
notes.append(text)
return f"已记录（第 {len(notes)} 条）"
组装 agent 并连问两轮：
researcher = create_agent(
model,
[search_docs, save_note],
system_prompt="你是调研助手：先检索文档，再把要点整理成一句话"
"存入笔记，最后答复用户。",
checkpointer=InMemorySaver(),
)
config = {"configurable": {"thread_id": "research-1"}}
for q in ("调研一下 agent 的记忆怎么做，检索并记笔记。",
"刚才记的第 1 条笔记原文是什么？"):
r = researcher.invoke(
{"messages": [{"role": "user", "content": q}]}, config)
print("问：", q)
print("答：", r["messages"][-1].content, "\n")
一次典型运行的输出如下：
问：调研一下 agent 的记忆怎么做，检索并记笔记。
答：已检索到《会话记忆设计》：用 checkpointer 按线程存取完整历史；要点已记为第 1 条笔记。
问：刚才记的第 1 条笔记原文是什么？
答：第 1 条笔记是："用 checkpointer 按线程存取完整历史。"
解读三点。第一轮是标准工具循环：模型调 search_docs、再调 save_note、最后作答，全程一次 invoke。第二轮没有触发任何工具调用——“刚才记的笔记”在线程历史里，模型直接从第一轮的回答取材作答，这正是 checkpointer 在工作；把 thread_id 换个值再问第二句，它立刻失忆。notes 列表在进程内跨轮共享，但那是 Python 变量的作用域，与 agent 记忆是两回事——进程重启后 notes 与 InMemorySaver 一起清零，生产落地要各换各的持久化方案。
8、总结与练习
回顾全文：create_agent 一行收编第 4 篇的手动循环，内部是 LangGraph 的双节点循环图；model、tools、system_prompt 三参数撑起九成场景；checkpointer 配 thread_id 让 agent 跨调用记住历史，同一线程号复用同一份历史；result["messages"] 是全轨迹，最终回答取最后一条的 content；stream 的 updates 模式按节点吐增量，是数工具链长度的调试利器；固定管道归 LCEL，定制图归 LangGraph。对于手头的应用，建议把第 4 篇的手动循环助手换成 create_agent 版：先裸跑通工具循环，再补 checkpointer 记忆，最后用 stream 把轨迹打一遍——三步走完，成品代理对你就是透明的。下一篇是记忆专篇：checkpointer 存的是完整历史，长对话的压缩与跨线程的长期记忆在那里展开。
练习两道：
- 把第 7 节的第二问换一个全新
thread_id再发，观察失忆；切回原线程复问，确认历史还在 - 把第 4 篇的天气加计算器交给
create_agent，用 updates 模式数一数两步工具链共打出几段更新
