✍️ 前言

说来有点不好意思。我确实把LangChain、LangGraph、Deep Agents的官方文档从头到尾翻了个遍，前前后后加起来，光是整理的笔记就快200万字了。那段时间我逢人就说“LangChain我熟了”，甚至一度觉得自己已经站在了LLM应用开发的浪尖上，只差一个项目来证明自己。

然后项目来了。

需求很简单：做一个能读写项目文件、能执行脚本的编码助手。听起来不就是官方文档里那几个示例代码拼一下的事吗？

结果，现实给了我三记响亮的耳光：

• 第一记耳光：create_deep_agent 一跑起来，Token像被开了水龙头，几万几万地消耗，而我的Agent正沉迷于一项很有仪式感的工作——对着TodoList反复记账，就是不写代码。
• 第二记耳光：我想让它执行 script/run.py，它反手给我一个 D:/my_project/script/run.py 的绝对路径错误。我明明设置了 virtual_mode=True，可它根本不买账。
• 第三记耳光：它在死循环里转了10分钟，我对着终端发呆，脑子里只有一个问题：我看了200万字的文档，到底看了个啥？

后来我才意识到问题出在哪——官方文档教你的，全是在“理想环境”下跑“玩具Demo”。它告诉你每个工具怎么用，却从没告诉你在一个真实项目里，这些工具凑在一起会怎么打架。

create_deep_agent 看起来是“开箱即用”的万金油，实际上它是一辆出厂就装满了配件的SUV——配件都是好东西，但如果你不知道每个配件为什么装在这里、怎么调校，它跑起来可能比自行车还慢，还容易翻沟里。

所以这篇文章，就是我的“翻沟实录”加“修车指南”。

我会把那个让我头疼的 deep_agent 拆回最基础的 create_agent，再一个配件一个配件地装回去。在这个过程中，你会看到：

• 那个让人又爱又恨的 TodoListMiddleware 到底是干什么的；
• Backend 的虚拟路径和真实路径之间，隔着一个怎样的大坑；
• 以及，如何让一个通用框架，为你特定的工程场景服务，而不是反过来。

如果你也经历过“读完文档觉得自己无所不能，一写项目发现自己什么都不会”的落差，这篇文章应该是为你写的。

毕竟，200万字的笔记都写了，总不能被一个斜杠卡住吧。

环境配置：Windows 上三分钟搭好开发环境

在动手之前，先花三分钟把环境搭好。

我选 uv 作为包管理工具，理由很简单：快。用 Rust 写的，解析依赖、同步环境的速度比传统工具快 10 到 100 倍。而且它和 pip 的体验几乎无缝衔接，上手成本极低。

注：LangChain 和 LangGraph 要求 Python 3.10 以上，我本机装的是 Python 3.14.2，完全满足。

我本机已经装了 Python 3.14.2，所以直接用 pip 全局安装 uv 就行：

pip install uv

装完后验证一下：

uv --version

输出类似 uv 0.5.x 就说明装好了。

在项目根目录下，执行：

uv venv

uv venv 会自动识别当前 Python 版本（3.14.2），并在当前目录创建 .venv 文件夹。整个过程两三秒，比传统 python -m venv 快不少。

如果你想指定 Python 版本，可以用 uv venv --python 3.14.2，但因为我本机只有一个 Python 版本，直接 uv venv 就够了。

Windows 下激活：

.venv\Scripts\activate

激活成功后，命令行前缀会出现 (.venv)，说明环境已就绪。

如果之后想退出环境，执行 deactivate 即可。

在项目根目录下，执行初始化命令：

uv init

uv init 会生成一个 pyproject.toml 文件，这是 Python 项目的标准配置文件，里面声明了项目的元信息和依赖范围。

生成的 pyproject.toml 大概长这样：

[project]
name = "my-coding-agent"
version = "0.1.0"
requires-python = ">=3.14"
dependencies = []

它不会自动创建虚拟环境，这一步在后面由 uv sync 完成。

用 uv add 添加项目需要的核心包：

uv add langchain langgraph deepagents langchain-openai

这条命令会做三件事：

1. 更新 pyproject.toml：在 [project.dependencies] 下写入依赖包及版本范围。
2. 创建 .venv 虚拟环境（如果尚未创建）。
3. 生成 uv.lock 文件：锁定所有依赖的精确版本，确保团队环境一致。

如果想加开发依赖（如 pytest、ruff 等），加 --dev 参数即可：

uv add --dev pytest ruff

依赖添加完成后，执行同步命令，将所有依赖安装到虚拟环境中：

uv sync

uv sync 会读取 uv.lock（如果存在）或 pyproject.toml，精确安装所有包。整个过程几秒到几十秒，比传统 pip install 快不少。

以后新增或删除依赖，只需要 uv add / uv remove，然后再次 uv sync 即可。

执行完上述步骤后，项目目录结构是这样的：

my-coding-agent/
├── .venv/                 # 虚拟环境（自动创建）
├── pyproject.toml         # 项目配置 + 依赖声明
├── uv.lock                # 精确版本锁定文件（自动生成）
└── main.py                # 你的代码（手动创建）

如果你用 Anthropic 的模型，把 langchain-openai 换成 langchain-anthropic 就行。

uv add 比 pip install 快不少，背后用的是 Rust 写的解析器和缓存机制。但用法和 pip 完全一样，零学习成本。

跑个简单的检查脚本，确认一切正常：

# check_env.py
import sys
print(f"Python version: {sys.version}")

import langchain
print(f"LangChain version: {langchain.__version__}")

import langgraph
print("LangGraph imported successfully")

import deepagents
print("DeepAgents imported successfully")

uv run python check_env.py

或者直接激活环境后 python check_env.py 也行。

如果输出了 Python 3.14.2 和各包的版本号，说明环境已经准备好了。

如果你安装了pycharm 2026的话，他是支持uv快捷指令的，只需要在界面上点点点即可：

先搞清楚这四个东西到底是什么关系

在开始写代码之前，我觉得有必要先把这堆名词捋清楚——langchain-core、langgraph、langchain、deepagents。因为如果连它们之间的关系都搞不明白，后面写代码就是在黑盒里瞎试。

坦白说，我当初看文档的时候，就是把这几个概念混在一起理解的。看完感觉“我都懂了”，结果一写项目才发现——我连谁依赖谁都没搞清楚。这章就当是帮我、也帮读者把这笔糊涂账算清楚。

先上结论。LangChain 官方把这四个东西分成了三个层次：

层次	包/库	官方定位	一句话解释
基础设施层	langchain-core	基础抽象	定义接口规范，不干活
运行时层	langgraph	Agent Runtime	管“怎么跑”——状态、持久化、中断恢复
框架层	langchain	Agent Framework	管“怎么用”——模型、工具、中间件
工具链层	deepagents	Agent Harness	管“拿来用”——开箱即用的全套能力

它们的关系是层层叠加上去的：

langchain-core ← langgraph ← langchain ← deepagents

下面逐个拆解。

langchain-core：一切的基础，但不做任何事

langchain-core 是整个 LangChain 生态的地基。它不提供任何具体功能，而是定义了所有组件该怎么说话——也就是接口和抽象类。

具体来说，langchain-core 里定义了：

• Chat models 的接口（怎么调用模型）
• Tools 的接口（工具长什么样）
• Vector stores、Retrievers 的接口
• Runnables 协议（LCEL 的核心，让不同组件能串起来）

你可以把 langchain-core 想象成 USB 接口的标准——它规定了插头长什么样、怎么传数据，但它本身不能充电也不能传文件。

任何 Provider（OpenAI、Anthropic、Google 等）只要实现了 langchain-core 里定义的接口，就能被整个 LangChain 生态使用。这就是 LangChain 能够“一套代码换模型”的秘密。

关键点：你不会直接和 langchain-core 打交道，但它是一切能跑起来的前提。

langgraph：Agent 的“发动机”与“刹车系统”

如果说 langchain-core 是接口标准，那 langgraph 就是真正干活的运行时。

langgraph 解决的核心问题是：Agent 怎么可靠地运行？

它把 Agent 的执行建模成一张图（Graph） ——节点是步骤，边是转移条件，每个步骤都能读写共享状态。基于这个图模型，langgraph 提供了生产级 Agent 必备的能力：

• 持久化执行（Durable Execution） ：Agent 可以在失败后从中断点恢复，而不是从头开始。
• 检查点（Checkpointing） ：每一步的状态都会被保存。
• 流式传输（Streaming） ：可以实时观察 Agent 的每一步思考。
• 人机回环（Human-in-the-loop） ：在执行关键操作前暂停，等待人工确认。

你可以把 langgraph 想象成汽车的发动机和刹车系统——它决定车怎么跑、怎么停、怎么在熄火后重新启动。但你不会直接坐在发动机上开车。

关键点：langgraph 是底层运行时，提供了“可靠性”和“可控性”，但使用门槛较高——你需要自己定义图的结构、节点、边。

langchain（create_agent）：给你一个“方向盘”

langchain（特指 1.0 版本中的 create_agent）是在 langgraph 之上构建的高层抽象。

它解决的核心问题是：开发者怎么更方便地创建 Agent？

你不需要懂图、不需要管状态、不需要写节点和边——只需要告诉 create_agent 三样东西：

1. 用什么模型（model）
2. 有什么工具（tools）
3. 怎么扩展行为（middleware）

然后 create_agent 在内部帮你编译成一个 langgraph 图，跑起来就是一个标准的 ReAct 循环（思考 → 选工具 → 执行 → 观察 → 再思考，直到给出答案）。

如果说 langgraph 是发动机，那 create_agent 就是方向盘+油门+刹车——你不用懂发动机怎么工作的，踩油门就能走。

create_agent 最重要的特性是中间件（Middleware） 。中间件让你能在 Agent 的执行流程中插入自定义逻辑——比如在调用模型前修改 Prompt、在工具执行后记录日志、在上下文太长时自动摘要。我们后面会详细用到它。

关键点：langchain（create_agent）让你用最少的代码跑起一个 Agent，同时通过中间件保留了足够的扩展性。

deepagents：给你一辆“顶配整车”

deepagents 是在 create_agent 之上又包了一层。它解决的核心问题是：怎么让 Agent 开箱即用地处理复杂任务？

deepagents 没引入新的运行时——它底层跑的依然是 langgraph。它做的事情是：在 create_agent 的基础上，预置了一整套中间件和默认配置。

默认自带的能力包括：

• 规划（Planning） ：write_todos 工具，让 Agent 能列出待办清单、跟踪进度
• 文件系统（Filesystem） ：ls、read_file、write_file 等工具，让 Agent 能读写文件
• 子 Agent（Subagents） ：主 Agent 可以把复杂任务委派给专门的子 Agent 执行
• 上下文管理：自动摘要历史对话、大工具结果自动落盘
• Backend 抽象：可插拔的存储后端，支持内存、本地文件系统、跨线程持久化等

如果说 create_agent 是一辆基础款轿车，那 deepagents 就是顶配版SUV——导航、全景天窗、座椅加热、自适应巡航全都给你装好了。但你如果不了解这些配置是怎么工作的，开起来可能反而觉得笨重。

关键点：deepagents 追求的是“开箱即用”，代价是“黑盒感”更强——你不知道它默认装了多少东西、这些东西之间怎么互动。

小结：从下到上，越来越“傻瓜”

把这四个东西串起来看，其实是一条清晰的抽象递进链：

langchain-core（接口规范）
       ↓
langgraph（运行时：怎么可靠地跑）
       ↓
langchain / create_agent（框架：怎么方便地创建）
       ↓
deepagents（工具链：怎么开箱即用地解决复杂问题）

越往下，控制力越强，但上手越难；越往上，开发越快，但黑盒感越重。

一个值得记住的原则：从最高的抽象起步，只有当上层确实满足不了需求时，再下沉到下层。

这也就解释了为什么我在第一章里选择了 deepagents——因为它最快。但也解释了为什么我后来那么痛苦——因为我没搞清楚它默认装的那些“配置”到底是怎么工作的，一旦出了 bug，我连从哪里开始排查都不知道。

所以接下来的章节，我们会反过来走：从最基础的 create_agent 开始，一个中间件一个中间件地往上加，直到拼出一个和 create_deep_agent 功能相当的编码助手。这样你就能看清楚——那个让你头疼的 deep_agent，里面到底装了些什么。

基础篇：从模型调用开始，让 Agent 先“活”起来

从这一章开始，我们正式动手写代码。

整个第三章的目标是搭建一个能跑起来的 Coding Agent 基础版本。它可能还不够聪明，可能偶尔会犯错，但至少它能听懂人话、能调用工具、不会一上来就死循环烧 Token。

而模型调用，是所有 Agent 的起点。

模型调用：Agent 的“心脏”

一个 Agent 的完整循环本质上是：

用户输入 → 模型推理 → 判断是否调用工具 → 执行工具 → 将结果喂给模型 → 继续推理 → ... → 最终输出

无论中间件怎么装、Backend 怎么配、路由怎么走，最终都绕不开这一行代码：

response = model.invoke(messages)

所以，把模型调用这一步调通、调稳，后面的所有东西才有意义。

在开始写代码之前，有一件事必须想清楚：我们到底用什么模型？

我本地是一台 RTX 5060 的笔记本，显存只有 8GB。这个硬件条件决定了我不可能跑 30B、70B 的大模型，只能在“小参数”模型里找最优解。Ollama 上能用的代码模型其实不少，但各自的情况差别很大：

CodeLlama 系列（Meta 出品）

这是 Meta 基于 Llama 2 开发的早期代码模型系列，算是代码模型的“老前辈”了。Ollama 上提供了 7B、13B、34B、70B 四个版本。

它们的硬件门槛是这样的：

• CodeLlama 7B：全精度 13.4GB，4-bit 量化后约 4.5-5.0GB。8GB 显存勉强能跑，但留给上下文（KV Cache）的空间很小，处理长代码时会非常吃力。
• CodeLlama 13B：全精度 26GB，量化后约 8.0-9.5GB。已经接近甚至超过 8GB 的上限，基本跑不动。
• CodeLlama 34B / 70B：量产后分别需要 18GB 和 40GB 以上的显存，直接不用考虑。

更关键的是，CodeLlama 基于的是 Llama 2 架构，技术相对较老。在代码生成、多语言支持、指令遵循能力上，和 newer 的模型有明显差距。

CodeGemma 系列（Google 出品）

这是 Google 推出的轻量级代码模型，主打“小而快”。Ollama 上提供 2B 和 7B 两个版本。

• CodeGemma 2B：非常轻量，对显存极度友好。但 2B 参数量的模型在代码理解和生成上会显得比较“笨”，而且这个版本主要面向代码补全（fill-in-the-middle）场景，不太适合对话式的 Coding Agent。
• CodeGemma 7B：4-bit 量化后约 5.0-5.3GB，显存占用很宽裕。Google 官方说它在 Python 和 TypeScript 上表现不错。但 CodeGemma 的训练目标更偏向“代码补全”而非“执行复杂编程任务”，作为 Agent 的大脑可能会力不从心。

Qwen2.5-Coder 系列（阿里出品）

这是目前公认的“最佳小参数代码模型”，专门为代码生成、代码推理、多语言支持而优化。

• Qwen2.5-Coder 7B：4-bit 量化后约 4.7GB，8GB 显存跑起来很从容，上下文窗口可以开到 32768。在 HumanEval（代码生成评测基准）等测试中，它的得分远超同量级的 CodeLlama 和 CodeGemma。
• Qwen2.5-Coder 14B：量化后约 9.0GB，稍超出了 8GB 的上限。但可以通过更低精度的量化（如 Q3_K_M）把体积压到 6-7GB 左右，不过这会在一定程度上牺牲模型的表现。

综合下来，我的选择是 Qwen2.5-Coder 7B。 它在性能、显存占用、智能水平之间取得了最佳平衡。CodeLlama 太老，CodeGemma 定位不太对，而 Qwen 就是为“写代码”这件事设计的。

拉取命令很简单：

ollama pull qwen2.5-coder:7b

拉取时建议把上下文窗口设大一点——Coding Agent 经常要处理几百上千行的代码文件，默认的 2048 远远不够。可以在拉取时指定：

ollama pull qwen2.5-coder:7b --num-ctx 32768

或者等我们后面在代码里通过参数配置，效果是一样的。

如果你在跑完全部的代码后发现 7B 不够用，想试试更大参数的模型，可以用低精度版本：ollama pull qwen2.5-coder:14b-q3_K_M。但我的建议是先把 7B 的流程跑通，真遇到能力瓶颈再考虑升级——毕竟 Agent 的表现很大程度上取决于 prompt 设计和工具封装，不一定全是模型的问题。

确认本地模型已就绪：

ollama list

能看到 qwen2.5-coder:7b 就说明准备好了。测试一下能否正常推理：

ollama run qwen2.5-coder:7b "写一个Python函数，计算斐波那契数列"

如果能正常输出代码，说明模型本身没问题。

接下来，用 LangChain 的 ChatOllama 把模型接入我们的代码。创建一个 model_demo.py：

# model_demo.py
from langchain_ollama import ChatOllama

model = ChatOllama(
    model="qwen2.5-coder:7b",
    temperature=0.1,
    num_predict=2048,
    num_ctx=32768,
    top_p=0.9,
)

或者

from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="qwen2.5-coder:7b",          # 指定你在Ollama中使用的模型名
    base_url="http://localhost:11434/v1", # Ollama的OpenAI兼容API地址[reference:4][reference:5]
    api_key="ollama",                  # API key是必需的，但Ollama会忽略它[reference:6][reference:7][reference:8]
    temperature=0.1,
    max_tokens=2048,
)

这里有几个参数值得解释一下：

• temperature：控制输出的随机性。0 表示每次都选概率最高的 token，输出最确定；1 表示更随机。代码生成这种场景需要确定性高一些，0.1 是个好起点。
• num_predict：最大输出 token 数，相当于“模型最多说多少字”。2048 对于生成一个函数或一个类足够了，如果要生成整个文件再调大。
• num_ctx：上下文窗口大小，决定了模型能“记住”多少历史对话和代码内容。32768 是 7B 模型能支持的较大值，足够处理中等规模的文件。
• top_p：核采样参数，和 temperature 一起控制输出的多样性。0.9 是常见取值，让模型在保留质量的同时有一点灵活性。

注意：num_ctx 在代码里设置，和在 ollama pull 时设置的效果是一样的。在代码里设置的好处是可以针对不同任务灵活调整——处理长文件时调大，简单问答时调小省显存。

最简单的调用方式是用 invoke() 直接传入字符串：

response = model.invoke("用Python写一个快速排序函数")
print(response.content)

运行：

uv run python model_demo.py

如果能正确输出快速排序的 Python 代码，你的模型调用就已经打通了。

invoke() 会一次性等待全部结果返回。对于代码生成这种可能需要几秒钟的任务，等起来有点煎熬。

LangChain 提供了流式接口 stream()，可以逐块（chunk）输出结果：

# model_stream_demo.py
from langchain_ollama import ChatOllama

model = ChatOllama(
    model="qwen2.5-coder:7b",
    temperature=0.1,
)

for chunk in model.stream("用Python写一个快速排序函数"):
    print(chunk.content, end="", flush=True)

运行后，你会看到代码一个字一个字地往外蹦。流式输出在处理长代码时尤其重要——用户可以提前看到部分输出，不用干等着。后面我们构建 Agent 时，默认就会使用 stream()。

刚才我们用 invoke() 传入了一个字符串，这其实是 LangChain 帮你做了一层简化。它内部会自动把字符串转换成 HumanMessage。

LangChain 的标准消息格式是这样的：

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

messages = [
    SystemMessage(content="你是一个Python专家，只输出代码，不要解释"),
    HumanMessage(content="写一个快速排序函数"),
]

多轮对话时，把历史消息都放进列表：

messages = [
    SystemMessage(content="你是一个Python专家"),
    HumanMessage(content="什么是装饰器？"),
    AIMessage(content="装饰器是Python中一种高阶函数..."),
    HumanMessage(content="那如何带参数？"),
]

response = model.invoke(messages)

这个格式很重要，因为 Agent 的核心就是不断地把用户消息、模型回复、工具执行结果拼成一个消息列表，喂给模型。后续我们在构建 Agent 时，会频繁地和这些消息类型打交道。

一个小技巧：SystemMessage 在 Ollama 中对应的是 system 角色，它帮助设定模型的身份和规则，比如“你是一个 Python 专家”。HumanMessage 是用户输入，AIMessage 是模型的回复。LangChain 会自动把它们映射成 Ollama 所需的 messages 数组格式。

几个容易踩的坑

在你开始跑模型之前，有几点值得提前知道：

坑一：Ollama 的默认配置可能导致模型不输出任何内容。

有些模型对 temperature 和 num_predict 的默认值比较敏感。如果发现模型不回答或者回答到一半就停了，检查是否在 ChatOllama 里明确设置了这两个参数。我的建议是直接用上面给的配置模板。

坑二：num_ctx 设大了会爆显存。

32768 是 7B 模型在这个显卡上的安全值。如果你设到 65536 甚至 131072，模型加载时可能直接 OOM（内存溢出）。如果你确实需要更长的上下文，建议调低 num_predict（输出长度）来腾出空间。

坑三：模型输出完代码后可能继续啰嗦。

代码模型有时候会在输出完代码块后继续解释、总结，甚至问“还有什么需要帮助的吗”。如果你希望 Agent 只输出代码，可以在系统提示词里强调“只输出代码，不要解释”，或者在代码里加 stop 参数：

model = ChatOllama(
    model="qwen2.5-coder:7b",
    temperature=0.1,
    stop=["```", "```\n\n"],  # 在一个代码块结束后就停下来
)

stop 参数告诉模型遇到指定的字符串就停止生成。不过这个方式有一定局限性，最稳妥的还是写好系统提示词。

记忆：让 Agent 拥有“短期”和“长期”的脑子

上一节我们让模型能“说话”了，但它有个致命的问题：它没有记忆。

每次调用 model.invoke() 都是独立的，模型不会记得你上一句说了什么。这在单轮问答里没问题，但 Agent 的本质是多轮交互——用户说“帮我写个函数”，Agent 写了，用户又说“再加个参数校验”，这时候 Agent 如果忘了刚才写了什么，整个对话就崩了。

所以这一节我们要给模型装上“记忆”。

短期记忆：用 list 存下所有对话

最朴素的记忆方式，就是把所有对话消息存进一个列表。

LangChain 已经定义好了三种基础消息类型：

• SystemMessage：设定模型的角色和行为规则
• HumanMessage：用户发送的消息
• AIMessage：模型回复的消息

用一个 list 来存它们，就是最轻量的短期记忆。

from langchain.messages import SystemMessage,HumanMessage,AIMessage

# 用类型别名标记消息类型，方便类型检查
BasicMessage = AIMessage | SystemMessage | HumanMessage

# 短期记忆：一个按顺序存放所有消息的列表
memory: list[BasicMessage] = []

每次用户说话，就把 HumanMessage 追加进去；每次模型回复，就把 AIMessage 追加进去。这样下一次调用模型时，把整个 memory 列表传进去，模型就能“看到”历史对话了。

在开始对话之前，我们需要通过 SystemMessage 给模型设定一个身份和规则。这是 Coding Agent 的系统提示词：

system_prompt = """你是一个专业的编程助手，名叫 langcode。

你的核心能力：
1. 编写高质量的代码，遵循最佳实践
2. 解释技术概念和代码逻辑
3. 调试和修复代码中的问题
4. 根据需求推荐合适的技术方案

行为准则：
- 代码输出时使用 Markdown 代码块标注语言
- 对于复杂任务，先给出思路再写代码
- 如果需求不清晰，主动提问澄清
- 保持回答简洁专业，不闲聊
- 不要编造不存在的库或API

你正在为一位开发者提供服务，请认真、准确地完成每一次编程任务。"""

memory.append(SystemMessage(content=system_prompt))

系统提示词会作为第一条消息永远留在 memory 里，它不会被遗忘。

有了记忆，我们就可以把上一节的单次调用改成一个持续的对话循环。

先把模型配置好。这里我用的是上一节提到的 langchain_openai + base_url 方式连接 Ollama，这样代码结构和调用云端 OpenAI 完全一致，可移植性更好。

# 3.2_memory_demo.py
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from pydantic import SecretStr

BasicMessage = AIMessage | SystemMessage | HumanMessage

# 配置模型（通过 Ollama 的 OpenAI 兼容接口）
model = ChatOpenAI(
    model="qwen2.5-coder:7b",
    base_url="http://localhost:11434/v1",
    api_key=SecretStr("ollama"),  # Ollama 会忽略这个值，但参数必须传
    temperature=0.1,
    max_tokens=2048,
)

# 短期记忆
memory: list[BasicMessage] = []

# 注入系统提示词
system_prompt = """你是一个专业的编程助手，名叫 langcode。

你的核心能力：
1. 编写高质量的代码，遵循最佳实践
2. 解释技术概念和代码逻辑
3. 调试和修复代码中的问题
4. 根据需求推荐合适的技术方案

行为准则：
- 代码输出时使用 Markdown 代码块标注语言
- 对于复杂任务，先给出思路再写代码
- 如果需求不清晰，主动提问澄清
- 保持回答简洁专业，不闲聊
- 不要编造不存在的库或API

你正在为一位开发者提供服务，请认真、准确地完成每一次编程任务。"""

memory.append(SystemMessage(content=system_prompt))

然后实现对话循环：

def chat_loop():
    print("langcode已启动。输入 /quit 退出对话。")
    
    while True:
        # 1. 获取用户输入
        user_input = input("\n> ")
        if user_input.strip().lower() == "/quit":
            print("Goodbye!")
            break
        
        # 2. 将用户消息存入记忆
        memory.append(HumanMessage(content=user_input))
        
        # 3. 调用模型，传入完整记忆
        print("\langcode: ", end="", flush=True)
        full_response = ""
        for chunk in model.stream(memory):
            # 流式输出
            print(chunk.content, end="", flush=True)
            full_response += chunk.content
        print()  # 换行
        
        # 4. 将模型回复存入记忆
        memory.append(AIMessage(content=full_response))

if __name__ == "__main__":
    chat_loop()

运行效果：

langcode已启动。输入 /quit 退出对话。

> 用Python写一个快速排序函数

langcode: 
def quick_sort(arr):
    if len(arr) <= 1:
        return arr
    pivot = arr[0]
    left = [x for x in arr[1:] if x <= pivot]
    right = [x for x in arr[1:] if x > pivot]
    return quick_sort(left) + [pivot] + quick_sort(right)

这是一个经典的快速排序实现，时间复杂度 O(n log n)。

> 加个注释，解释一下思路

langcode: python
def quick_sort(arr):
    """
    快速排序：分治思想的经典应用
    - 选择基准元素（这里选第一个）
    - 将数组分为小于等于基准和大于基准两部分
    - 递归排序左右两部分，再合并
    """
    if len(arr) <= 1:
        return arr
    pivot = arr[0]
    left = [x for x in arr[1:] if x <= pivot]
    right = [x for x in arr[1:] if x > pivot]
    return quick_sort(left) + [pivot] + quick_sort(right)

注释已经加上了，解释了分治逻辑。

看到区别了吗？第二次提问时，模型“记得”我刚才写的是快速排序，所以直接基于已有代码加了注释，而不是重写一遍。这就是短期记忆的作用。

我们实现了持续对话循环，memory 列表会随着对话轮次无限增长。但模型有一个硬限制：上下文窗口（Context Window）有限。

我们的 qwen2.5-coder:7b 设置的 num_ctx=32768，大约能容纳 32768 个 token。一旦 memory 中的消息总 token 数超过这个值，模型就会报错或开始“遗忘”最早的内容。

所以，我们需要给短期记忆加一道护栏：当消息数量超过阈值时，自动裁剪。

最直接的实现方式是：只保留最近的 N 条消息，更早的直接丢弃。

# 在 3.2.3 的代码基础上增加上下文管理

MAX_MESSAGES = 20  # 保留最近 20 条消息（不含系统提示词）

def trim_memory(mem: list[BasicMessage], max_messages: int = MAX_MESSAGES) -> list[BasicMessage]:
    """
    裁剪记忆，保留系统提示词 + 最近 max_messages 条消息。
    """
    # 分离系统提示词和其他消息
    system_msgs = [m for m in mem if isinstance(m, SystemMessage)]
    other_msgs = [m for m in mem if not isinstance(m, SystemMessage)]

    # 只保留最近的 max_messages 条非系统消息
    trimmed_other = other_msgs[-max_messages:] if len(other_msgs) > max_messages else other_msgs

    # 返回系统提示词 + 裁剪后的消息
    return system_msgs + trimmed_other

trim_memory 应该在每轮对话结束后调用，防止下一轮开始时 memory 已经超限：

def chat_loop():
    memory = [SystemMessage(content=system_prompt)]
    print("langcode已启动。输入 /quit 退出对话。")
    
    while True:
        user_input = input("\n> ")
        if user_input.strip().lower() == "/quit":
            print("Goodbye!")
            return
        
        memory.append(HumanMessage(content=user_input))
        
        print("\langcode: ", end="", flush=True)
        full_response = ""
        for chunk in model.stream(memory):
            print(chunk.content, end="", flush=True)
            full_response += chunk.content
        print()
        
        memory.append(AIMessage(content=full_response))
        
        # 🔥 每轮结束后裁剪记忆，防止上下文溢出
        memory = trim_memory(memory)

按条数裁剪虽然简单，但不够精确——一条消息可能只有几个字，也可能包含几千行代码。更好的方式是按 Token 数量裁剪。

使用 tiktoken（OpenAI 的 tokenizer）或 transformers 的 tokenizer 来统计 token 数：

import tiktoken

def count_tokens(text: str, model: str = "gpt-4") -> int:
    """统计一段文本的 token 数量"""
    encoding = tiktoken.encoding_for_model(model)
    return len(encoding.encode(text))

def trim_memory_by_tokens(
    memory: list[BasicMessage], 
    max_tokens: int = 28000  # 留一些余量给模型输出
) -> list[BasicMessage]:
    """
    按 token 数量裁剪记忆，保留系统提示词 + 尽可能多的最近消息。
    """
    system_msgs = [m for m in memory if isinstance(m, SystemMessage)]
    other_msgs = [m for m in memory if not isinstance(m, SystemMessage)]
    
    # 从最新的消息开始累加
    kept_msgs = []
    current_tokens = 0
    
    for msg in reversed(other_msgs):
        msg_tokens = count_tokens(msg.content)
        if current_tokens + msg_tokens > max_tokens:
            break
        kept_msgs.insert(0, msg)
        current_tokens += msg_tokens
    
    return system_msgs + kept_msgs

不过 tiktoken 对 qwen2.5-coder 可能不是完全精确，但作为估算已经够用。在“进阶篇”中我们还可以讨论如何用更专业的工具（如 transformers 的 tokenizer）来做精准控制。

你可能会问：除了裁剪，能不能把早期对话压缩成摘要，保留关键信息而不是直接丢掉？

答案是可以，但摘要本身需要消耗一次模型调用（额外的 token 和时间）。对于“基础篇”，先用裁剪策略保持简单；在“进阶篇”中，我们可以引入 SummarizationMiddleware（摘要中间件），让 Agent 在上下文快满时自动生成摘要，实现更智能的上下文管理。

长期记忆：对话不能随程序关闭而丢失

上面的 memory 存在内存里，程序一关就没了。如果想让 Agent 在多次会话之间保留记忆，就需要长期记忆。

长期记忆有几种实现思路：

1. 存成文件：把对话历史序列化成 JSON 存到磁盘，下次启动时加载。
2. 向量数据库：把历史对话的语义向量存下来，按相似度检索相关记忆（RAG 方式）。
3. 数据库存储：用 SQLite 或 PostgreSQL 存结构化对话记录，按时间或主题查询。

对于 Coding Agent 来说，最朴素且实用的长期记忆是存储用户偏好和项目上下文，而不是把所有对话都存下来。比如：

def load_short_term_memory()->list[HumanMessage]:
    """从磁盘加载最近一条对话"""
    if os.path.exists("memory.json"):
        with open("memory.json", "r",encoding="utf-8") as f:
            data = json.load(f)
            # 只加载最近的 20 条消息，避免上下文过长
            return [HumanMessage(content=m["content"])  for m in data[-20:] if m["role"] == "human"]
    return []

def save_long_term_memory(messages:list[HumanMessage]) -> None:
    """保存到磁盘"""
    # 仅保存非系统消息（不包括 system_prompt）
    saved_messages = [{"role":m.type,"content":m.content} for m in messages]
    with open("memory.json", "w",encoding="utf-8") as f:
        json.dump(saved_messages,f)

对于更复杂的长期记忆需求（比如跨项目记忆技术栈偏好、记住用户常用的库），用向量数据库做语义检索会更合适。这个话题我们在“进阶篇”里展开。

然后我们在用户退出的时候将历史会话持久化存储，同时在加载新会话的时候把历史对话拼接进去：

def chat_loop():
    # 注意：系统提示词永远在列表最前面
    memory.extend(load_short_term_memory())
    print("langcode已启动。输入 /quit 退出对话。")

    while True:
        # 1. 获取用户输入
        user_input = input("\n> ")
        if user_input.strip().lower() == "/quit":
            save_long_term_memory(list(filter(lambda m:m.type=="human",memory)))
            print("Goodbye!")
            break

        # 2. 将用户消息存入记忆
        memory.append(HumanMessage(content=user_input))

        # 3. 调用模型，传入完整记忆
        print("\langcode: ", end="", flush=True)
        full_response = ""
        for chunk in model.stream(memory):
            # 流式输出
            print(chunk.content, end="", flush=True)
            full_response += chunk.content
        print()  # 换行

        # 4. 将模型回复存入记忆
        memory.append(AIMessage(content=full_response))

记忆管理的挑战

在 Coding Agent 中，记忆管理有两个核心挑战：

挑战一：上下文窗口有限

我们的模型 num_ctx=32768，也就是大约能容纳 32768 个 token。如果对话积累了几十轮，加上每次的代码输出，很容易撑爆。

一个常用的策略是滑动窗口：只保留最近 N 条消息，更早的用摘要代替。

# 伪代码：自动裁剪记忆
MAX_MESSAGES = 20

def trim_memory():
    if len(memory) > MAX_MESSAGES:
        # 保留系统提示词 + 最近 MAX_MESSAGES-1 条消息
        memory = [memory[0]] + memory[-(MAX_MESSAGES-1):]

挑战二：哪些信息该记住，哪些该忘？

不是所有对话都有价值。一段完整的代码对话可能有几十轮，其中混杂着“试错-修正”的噪音。更好的方式是用一个上层 Agent 定期对历史对话做摘要，把“用户意图”和“最终产出”提炼出来，存入长期记忆。

工具：让 Agent 长出“手”，主动保存有价值的信息

在 3.2 节的最后，我们留下了一个问题：什么时候才是保存用户有价值信息的时机？

现在答案已经很清晰了——当 Agent 通过工具调用（Tool Call）主动判断需要保存时。

换句话说，保存的决策权从“开发者预先写好逻辑”变成了“Agent 在运行时自主判断”。这就是 Agent 和普通脚本的本质区别：

普通脚本：if user_says_save(): save_to_file()
Agent：  用户说"记住这个" → 模型理解意图 → 调用 write_file 工具 → 保存

下面我们来定义 Agent 需要的核心工具。

核心工具一：写入文件 (write_file)

这是 Agent 保存信息的核心工具。它的定义很简单，但背后的设计意图很重要：

from langchain.tools import tool
from pathlib import Path

from typing import Annotated

from langchain.messages import ToolMessage

# 设置一个基础工作目录，所有文件操作都在这个目录下进行
FsBackend = Path.cwd() / "backend"

def _resolve_path(path: str) -> Path:
    """将用户提供的路径解析为绝对路径，并确保它在 BASE_DIR 内"""
    p = Path(path)
    # 如果路径是绝对路径，去掉盘符，转为相对路径（或直接报错）
    if p.is_absolute():
        # 安全措施：不允许绝对路径，防止删除系统文件
        raise ValueError(f"Absolute paths are not allowed: {path}")
    # 拼接基础目录并解析为绝对路径
    resolved = (FsBackend / p).resolve()
    # 检查是否在 BASE_DIR 内（防止通过 ../ 逃逸）
    if not str(resolved).startswith(str(FsBackend.resolve())):
        raise ValueError(f"Path is outside the workspace: {path}")
    return resolved


@tool(name_or_callable="write_file")
def write_file(
        file_path: Annotated[str, "Absolute path where the file should be created. Must be absolute, not relative."],
        content: Annotated[str, "The text content to write to the file. This parameter is required."]
) -> ToolMessage:
    """
    Writes to a new file in the filesystem.

    Usage:
    - The write_file tool will create the a new file.
    - Prefer to edit existing files (with the edit_file tool) over creating new ones when possible.
    """
    try:
        file_path = _resolve_path(file_path)
        # 确保父目录存在
        file_path.write_text(content, encoding="utf-8")

        return ToolMessage(
            content=f"Updated file {file_path}",
            name="write_file",
            tool_call_id=1,
            status="success",
        )
    except Exception as e:
        return ToolMessage(
            content=str(e),
            name="write_file",
            tool_call_id=1,
            status="error",
        )

注意 write_file 的 Docstring —— 它明确告诉模型当前工具的作用是写一个content到一个文件

核心工具二：读取文件 (read_file)

有了写入，自然需要读取：

@tool(name_or_callable="read_file")
def read_file(
    file_path: Annotated[str, "Absolute path to the file to read. Must be absolute, not relative."],
) -> ToolMessage:
    """Reads a file from the filesystem.

Assume this tool is able to read all files. If the User provides a path to a file assume that path is valid. It is okay to read a file that does not exist; an error will be returned.

- You should ALWAYS make sure a file has been read before editing it."""
    file_path = _resolve_path(file_path)
    try:
        content:str = file_path.read_text(encoding="utf-8")
        return ToolMessage(
            content=content,
            name="read_file",
            tool_call_id=2,
            status="success",
        )
    except ValueError as e:
        return ToolMessage(
            content=f"Error: {e} for '{file_path}'",
            name="read_file",
            tool_call_id=2,
            status="error",
        )

核心工具三：列出目录 (ls)

让 Agent 能够“看”到当前工作环境：

TOOL_RESULT_TOKEN_LIMIT = 20000  # Same threshold as eviction
TRUNCATION_GUIDANCE = "... [results truncated, try being more specific with your parameters]"

def truncate_if_too_long(result: list[str] | str) -> list[str] | str:
    """Truncate list or string result if it exceeds token limit (rough estimate: 4 chars/token)."""
    if isinstance(result, list):
        total_chars = sum(len(item) for item in result)
        if total_chars > TOOL_RESULT_TOKEN_LIMIT * 4:
            return result[: len(result) * TOOL_RESULT_TOKEN_LIMIT * 4 // total_chars] + [TRUNCATION_GUIDANCE]  # noqa: RUF005  # Concatenation preferred for clarity
        return result
    # string
    if len(result) > TOOL_RESULT_TOKEN_LIMIT * 4:
        return result[: TOOL_RESULT_TOKEN_LIMIT * 4] + "\n" + TRUNCATION_GUIDANCE
    return result

@tool(name_or_callable="ls")
def ls(
    path: Annotated[str, "Absolute path to the directory to list. Must be absolute, not relative."],
) -> ToolMessage:
    """Lists all files in a directory.

This is useful for exploring the filesystem and finding the right file to read or edit.
You should almost ALWAYS use this tool before using the read_file or edit_file tools."""
    dir_path = _resolve_path(path)

    items = []
    for item in sorted(dir_path.iterdir()):
        prefix = "📁" if item.is_dir() else "📄"
        items.append(f"{prefix} {item.name}")

    return ToolMessage(
        content=str(truncate_if_too_long(items)),
        tool_call_id=3,
        name="ls",
        status="success",
    )

有了工具定义，接下来把它们绑定到模型上：

# 汇总所有工具
tools = [write_file, read_file, ls]

# 全局模型
model = ChatOpenAI(
    model="qwen2.5-coder:7b",
    base_url="http://localhost:11434/v1",
    api_key=SecretStr("ollama"),
    temperature=0.1,
).bind_tools(tools=tools)

同时我们需要修改system_prompt：

# 系统提示词
system_prompt = """你是一个专业的编程助手，名叫 langcode。

## 核心能力
1. 编写高质量、规范的代码，遵循最佳实践和语言惯用法
2. 解释技术概念和代码逻辑，帮助用户理解复杂技术问题
3. 调试和修复代码中的问题，提供可落地的解决方案
4. 根据需求推荐合适的技术方案和架构设计

## 工作方式

### 1. 思考与规划
- 复杂任务（涉及多个文件、多种技术）先给出整体思路和计划，再逐步实现
- 简单任务直接给出结果，不冗余解释
- 如果需求不清晰或信息不足，主动提问澄清，不要猜测

### 2. 代码输出规范
- 使用 Markdown 代码块标注语言类型（如 ```python、```javascript）
- 关键函数和类必须包含 docstring 或注释，说明用途、参数和返回值
- 保持代码简洁、可读、可维护，避免过度工程化

### 3. 文件操作规则（重要！）
你拥有读写文件的能力。当遇到以下情况时，应主动使用工具保存信息：
- 用户明确要求保存或持久化某些内容（"帮我保存到..."、"写个文件..."）
- 你生成了可复用的代码模块、配置文件或文档，且用户可能后续使用
- 用户询问"如何保存"或"记不记得"时，建议通过文件保存而非依赖对话记忆
- **所有文件路径必须使用相对路径**，工作目录为当前项目根目录
- 不要在路径中使用反斜杠 `\`，始终使用正斜杠 `/`（如 `"src/utils/helper.py"`）
- 如果目录不存在，工具会自动创建，你无需提前检查

### 4. 安全边界
- 不要编造不存在的库、API 或技术方案
- 不要执行可能危害系统的操作（如删除关键文件、修改系统配置）
- 遇到疑似错误的用户指令，友善地指出潜在风险并提供替代方案

### 5. 沟通风格
- 保持回答简洁、专业、直接，不闲聊
- 使用中文与用户交流，代码和术语可保留英文
- 对于可能耗时较长的任务，提前告知用户预期时间和步骤

你正在为一位开发者提供编程服务，请认真、准确地完成每一次任务。"""

根据官方文档，当模型收到用户输入时，如果判断需要调用工具，model_with_tools.invoke() 返回的 AIMessage 会包含一个 tool_calls 字段：

response = model.invoke([
    SystemMessage(content=system_prompt),
    HumanMessage(content="帮我写一个快速排序，保存到 quicksort.py")
])

print(response.tool_calls)
# 输出类似：
# [
#     {
#         "name": "write_file",
#         "args": {"path": "quicksort.py", "content": "def quick_sort(arr):\n    ..."},
#         "id": "call_abc123"
#     }
# ]

⚠️ 注意：模型选择与工具调用兼容性

经过实际测试，qwen2.5-coder:7b 虽然代码生成能力出色，但其工具调用输出格式不符合 OpenAI 规范——模型会把工具调用写成 JSON 塞进 content 字段，而不是标准的 tool_calls，导致 bind_tools() 无法正常解析。

因此，本教程改用 qwen3.5:4b，它原生支持工具调用，且轻量快速（约 3GB 显存）。

拿到 tool_calls 后，我们需要执行对应的工具函数，然后把结果封装成 ToolMessage 追加到对话中：

from langchain.messages import ToolMessage

# 构建工具名 → 工具函数的映射
tool_map = {tool.name: tool for tool in tools}

# 执行所有工具调用
for tool_call in response.tool_calls:
    tool_name = tool_call["name"]
    tool_args = tool_call["args"]
    tool_id = tool_call["id"]
    
    # 调用对应的工具
    result = tool_map[tool_name].invoke(tool_args)
    
    # 将结果封装为 ToolMessage
    memory.append(ToolMessage(content=result, tool_call_id=tool_id))

ToolMessage 的 tool_call_id 必须与 tool_calls 中的 id 匹配，这样模型才能把工具执行结果和之前的调用对应起来。

现在把整个过程串起来：

用户："帮我写个快速排序，保存到 quicksort.py"
    ↓
1. model_with_tools.invoke(memory)
    ↓
2. 模型返回 AIMessage，包含 tool_calls:
   [{"name": "write_file", "args": {"path": "quicksort.py", "content": "..."}}]
    ↓
3. 执行 write_file.invoke(args)
    ↓
4. 得到结果： "✅ 文件已保存到：/path/to/quicksort.py"
    ↓
5. 将 ToolMessage 追加到 memory
    ↓
6. model_with_tools.invoke(memory)  # 再次调用，让模型知道工具执行完毕
    ↓
7. 模型返回最终回复："已为你生成快速排序并保存到 quicksort.py"

关键点：工具调用后，必须把 ToolMessage 放回 memory，让模型“看到”工具执行的结果，否则模型无法知道操作是否成功，也无法基于结果做后续决策。

完整代码：

import asyncio
from langchain_openai import ChatOpenAI
from pydantic import SecretStr

from langchain.messages import SystemMessage,HumanMessage,AIMessage,ToolMessage

from tool import write_file,read_file,ls

# 汇总所有工具
tools = [write_file, read_file, ls]
# 工具映射：名称 → 工具函数
tool_map = {tool.name: tool for tool in tools}
MAX_TOOL_ITERATIONS = 2

# 我们使用泛型来标记基础Message类型
BasicMessage = AIMessage | SystemMessage | HumanMessage | ToolMessage

# 全局模型
model = ChatOpenAI(
    model="qwen3.5:4b",
    base_url="http://localhost:11434/v1",
    api_key=SecretStr("ollama"),
    temperature=0.1,
).bind_tools(tools=tools)

MAX_MESSAGES = 20  # 保留最近 20 条消息（不含系统提示词）

def trim_memory(mem: list[BasicMessage], max_messages: int = MAX_MESSAGES) -> list[BasicMessage]:
    """
    裁剪记忆，保留系统提示词 + 最近 max_messages 条消息。
    """
    # 分离系统提示词和其他消息
    system_msgs = [m for m in mem if isinstance(m, SystemMessage)]
    other_msgs = [m for m in mem if not isinstance(m, SystemMessage)]

    # 只保留最近的 max_messages 条非系统消息
    trimmed_other = other_msgs[-max_messages:] if len(other_msgs) > max_messages else other_msgs

    # 返回系统提示词 + 裁剪后的消息
    return system_msgs + trimmed_other

# 系统提示词
system_prompt = """你是一个专业的编程助手，名叫 langcode。

## 核心能力
1. 编写高质量、规范的代码，遵循最佳实践和语言惯用法
2. 解释技术概念和代码逻辑，帮助用户理解复杂技术问题
3. 调试和修复代码中的问题，提供可落地的解决方案
4. 根据需求推荐合适的技术方案和架构设计

文件操作规则（重要！）
你拥有读写文件的能力。当遇到以下情况时，应主动使用工具保存信息：
- 用户明确要求保存或持久化某些内容（"帮我保存到..."、"写个文件..."）
- 你生成了可复用的代码模块、配置文件或文档，且用户可能后续使用
- 用户询问"如何保存"或"记不记得"时，建议通过文件保存而非依赖对话记忆
- **所有文件路径必须使用相对路径**，工作目录为当前项目根目录
- 不要在路径中使用反斜杠 `\\`，始终使用正斜杠 `/`（如 `"src/utils/helper.py"`）
- 如果目录不存在，工具会自动创建，你无需提前检查

安全边界
- 不要编造不存在的库、API 或技术方案
- 不要执行可能危害系统的操作（如删除关键文件、修改系统配置）
- 遇到疑似错误的用户指令，友善地指出潜在风险并提供替代方案

你正在为一位开发者提供编程服务，请认真、准确地完成每一次任务。"""

async def chat_loop():
    memory: list[BasicMessage] = [SystemMessage(content=system_prompt)]
    print("langcode已启动。输入 /quit 退出对话。")

    while True:
        user_input = input("\n> ")
        if user_input.strip().lower() == "/quit":
            print("Goodbye!")
            return

        memory.append(HumanMessage(content=user_input))

        # 工具调用循环（支持流式）
        tool_iterations = 0
        final_response = None

        while tool_iterations < MAX_TOOL_ITERATIONS:
            tool_iterations += 1

            # =========================================================
            # 🔥 关键改动：用 astream 替代 invoke，实现流式输出
            # =========================================================
            print("\langcode: ", end="", flush=True)
            full_content = ""
            tool_calls_accumulator = {}  # 按 index 累积 tool_calls

            # 流式接收模型的响应
            async for chunk in model.astream(memory):
                # 1️⃣ 处理文本内容：立即输出
                if chunk.content:
                    print(chunk.content, end="", flush=True)
                    full_content += chunk.content

                # 2️⃣ 处理工具调用：按 index 累积
                if chunk.tool_calls:
                    for tc in chunk.tool_calls:
                        idx = tc.get("index", 0)
                        if idx not in tool_calls_accumulator:
                            tool_calls_accumulator[idx] = {
                                "name": "",
                                "args": {},
                                "id": tc.get("id", ""),
                            }
                        # 合并 name（可能只在第一个 chunk 中出现）
                        if tc.get("name"):
                            tool_calls_accumulator[idx]["name"] = tc["name"]
                        # 合并 args（可能需要跨多个 chunk 拼接）
                        if tc.get("args"):
                            for k, v in tc["args"].items():
                                tool_calls_accumulator[idx]["args"][k] = v
                        # 合并 id
                        if tc.get("id"):
                            tool_calls_accumulator[idx]["id"] = tc["id"]

            print()  # 换行

            # 3️⃣ 将完整的响应（含 content）存入 memory
            if full_content:
                memory.append(AIMessage(content=full_content))

            # 4️⃣ 检查是否有完整的工具调用
            if not tool_calls_accumulator:
                # 没有工具调用，这就是最终答案
                final_response = full_content
                break

            # 5️⃣ 有工具调用 → 执行
            tool_calls = list(tool_calls_accumulator.values())
            print(f"\n🔧 正在执行 {len(tool_calls)} 个工具调用...")

            for tc in tool_calls:
                tool_name = tc["name"]
                tool_args = tc["args"]
                tool_id = tc["id"]

                # 执行工具
                tool_message:ToolMessage = tool_map[tool_name].invoke(tool_args)
                tool_message.id = tool_id
                memory.append(tool_message)
                print(f"  ├─ {tool_name}({tool_args}) → {tool_message.content[:80]}")

            memory = trim_memory(memory)

        # 超限处理
        if tool_iterations >= MAX_TOOL_ITERATIONS and final_response is None:
            print("\n⚠️ 工具调用超过最大轮数，请简化需求。")
            memory.append(AIMessage(content="⚠️ 工具调用超过最大轮数，请简化需求。"))

        memory = trim_memory(memory)

if __name__ == "__main__":
    asyncio.run(chat_loop())

执行结果：

astream 返回的是一个异步迭代器（AsyncIterator），每次迭代产出一个 AIMessageChunk 对象。

async for chunk in model.astream(messages):
    # chunk 的类型是 AIMessageChunk
    print(type(chunk))  # <class 'langchain.messages.AIMessageChunk'>

AIMessageChunk 是 AIMessage 的流式版本，主要包含两个关键属性：

属性	类型	说明
content	str	当前 chunk 中的文本内容片段
tool_call_chunks	List[ToolCallChunk]	当前 chunk 中的工具调用片段
tool_calls	List[ToolCall]	累积后的完整工具调用（只有最后一个 chunk 才有）

关键区别：tool_call_chunks 是增量片段，每个 chunk 可能只包含工具调用的一部分（比如名称在一个 chunk，参数在后续 chunk）。而 tool_calls 是累积后的完整结果，只在流结束时才有。

tool_call_chunks: list[ToolCallChunk] = Field(default_factory=list)

这是最重要的设计。它存放的是工具调用的增量片段，而不是完整工具调用。

为什么要设计成“片段”？

如果模型要调用一个工具，并且参数很长（比如要生成 500 行代码作为 content 参数），一次性传回来会卡住。所以 Ollama/OpenAI 的流式 API 会把工具调用拆成多个小块（chunks），逐块传输：

# chunk 1: 工具调用的开头
AIMessageChunk(
    content="",
    tool_call_chunks=[
        ToolCallChunk(index=0, name="write_file", args='{"path": "test.py", "content": "')
    ]
)

# chunk 2: 部分代码内容
AIMessageChunk(
    content="",
    tool_call_chunks=[
        ToolCallChunk(index=0, args='def hello():\n    print("Hello")\n')
    ]
)

# chunk 3: 剩余代码 + 结束
AIMessageChunk(
    content="",
    tool_call_chunks=[
        ToolCallChunk(index=0, args='    return "World"')
    ]
)

每个 ToolCallChunk 包含：

• index：当模型同时调用多个工具时，用于标识是第几个工具
• name：工具名称（通常在第一个 chunk 出现）
• args：参数的 JSON 字符串片段（后续 chunk 逐步拼接）

实际流式中，args 字段叫 args，是一个 JSON 字符串片段。LangChain 内部会将这些片段累积成完整的 JSON 对象。

chunk_position: Literal["last"] | None = None

这是用于流式聚合的标记。它的作用是标识当前 chunk 在聚合流中的位置：

• None：普通 chunk
• "last"：这是这个流中的最后一个 chunk

使用场景：当你用 chunk1 + chunk2 + ... 累加多个 AIMessageChunk 时，LangChain 内部需要知道哪个 chunk 是最后一个，以便触发 tool_call_chunks → tool_calls 的转换。

对应的 BaseMessageChunk 基类中还有一个对应的设计：

# 在 BaseMessageChunk 中
chunk_position: Literal["first", "middle", "last"] | None = None

AIMessageChunk 只重定义了 "last"，其他位置由基类处理。

流式传输中，AIMessageChunk 的 tool_calls 字段（继承自 AIMessage）只有在所有 chunk 累加完毕后才有意义。在累加过程中，数据存在于 tool_call_chunks 中。当 LangChain 检测到一个 chunk_position="last" 的 chunk 被添加时，它会触发一个内部转换：将 tool_call_chunks 解析并填充到 tool_calls 字段。

LangChain 内部实现了 AIMessageChunk 的 __add__ 方法：

chunk1 = AIMessageChunk(content="你好", tool_call_chunks=[...])
chunk2 = AIMessageChunk(content="，世界", tool_call_chunks=[...])
chunk3 = AIMessageChunk(content="", chunk_position="last")

# 累加
full = chunk1 + chunk2 + chunk3

# 累加后：
# full.content == "你好，世界"
# full.tool_calls == 完整的工具调用列表（由 tool_call_chunks 合并而来）
# full.chunk_position == "last"

更高级的替代方案：astream_events

如果你的场景更复杂（比如需要同时处理多个步骤、工具执行状态等），LangChain 提供了 astream_events，它能发出更细粒度的事件：

async for event in model.astream_events(messages, version="v2"):
    if event["event"] == "on_chat_model_stream":
        # 每个文本 token
        print(event["data"]["chunk"].content, end="")
    elif event["event"] == "on_chat_model_end":
        # 完整的响应，包含 tool_calls
        print(event["data"]["output"].tool_calls)

astream_events 在 LangChain v1.3.0引入了全新的 v3 版本。v3 版本是一次重大的 API 升级，它通过结构化、可独立消费的事件流，彻底改变了我们与 Agent 交互的方式。

简单来说，astream_events v3 版本的核心是，从一个统一的“事件流”，演变成了一组可以独立、并发消费的“数据投影（Projections）”。

在 v1/v2 中，astream_events 返回一个混合类型的事件流，你需要自己在循环中解析不同类型的事件。v3 版本则完全改变了这一模式，它返回的是一个类型安全的流对象，你可以直接从该对象上获取多个独立的“投影（Projections）”。

这些投影就像是多个并行的数据管道，每个管道都专注于提供一种特定类型的数据，而且可以并发地消费它们。这意味着你可以同时接收模型的文本输出、工具调用状态和状态快照，而无需复杂的同步逻辑。

🧩 v3 提供了哪些“投影”？

v3 的 stream 对象提供了多种投影，让你可以精准地获取所需信息：

• stream.messages: 模型消息流。包含模型每次调用产生的文本、推理内容和工具调用参数。你可以从中获取：
  • message.text: 流式输出的文本块。
  • message.reasoning: 模型的推理内容（如果模型支持）。
  • message.tool_calls: 工具调用的参数块。
  • message.output: 本次模型调用完成后的完整消息对象。
• stream.tool_calls: 工具执行生命周期流。用于监控工具从开始到结束的完整过程。
• stream.values: 智能体状态快照。可以获取智能体在运行过程中的状态变化。
• stream.output: 最终状态。获取整个智能体运行的最终结果。
• stream.subgraphs: 子图流。用于处理嵌套运行（如子智能体）的事件。

为了让你更直观地感受差异，这里对比一下两种版本的典型用法。

v2 版本：手动解析事件类型

# v2: 在同一个循环中处理所有事件
async for event in agent.astream_events(input, version="v2"):
    if event["event"] == "on_chat_model_stream":
        # 处理文本块
        print(event["data"]["chunk"].content, end="")
    elif event["event"] == "on_tool_start":
        # 处理工具开始
        print(f"\n调用工具: {event['name']}")

v3 版本：独立消费数据投影

# v3: 分别处理不同的事件流
stream = await agent.astream_events(input, version="v3")

# 独立消费消息流（文本 + 工具参数）
for message in stream.messages:
    for delta in message.text:
        print(delta, end="", flush=True)  # 打字机效果
    # 当工具调用完成时
    finalized_tool_calls = message.tool_calls.get()

# 独立消费工具执行流
for tool_event in stream.tool_calls:
    if tool_event.event == "tool-started":  # 新的事件类型
        print(f"\n[工具开始] {tool_event.tool_name}")
    elif tool_event.event == "tool-finished":  # 新的事件类型
        print(f"\n[工具完成] {tool_event.output}")

✨ v3 的核心优势

• 更清晰的关注点分离：不再需要在一个庞大的事件循环中处理所有逻辑。文本输出、工具状态、状态快照等可以分开处理，代码更清晰、更易维护。
• 支持并发消费：stream.messages 和 stream.tool_calls 等投影可以被并发地 await，这为实现高性能的实时 UI 提供了可能。
• 更细粒度的事件类型：引入了 tool-started、tool-finished、content-block-delta 等更精确的事件，让你能对 Agent 的执行过程进行更精细的控制。
• 原生支持推理内容：message.reasoning 投影为处理模型的“思维链”提供了标准方式。

⚙️ 如何启用与兼容性

启用 v3 非常简单，只需在调用 astream_events 时指定 version="v3" 即可。

• LangChain 版本要求：v3 API 在 LangChain v1.3.0 中引入。建议使用更新版本。
• 与 v1/v2 的兼容性：v3 是一个全新的 API，与旧版本不兼容。不过，LangChain 在设计上保证了 v1/v2 的调用方式依然有效，所以你的旧代码不会因此崩溃。

基础篇总结：Agent 的本质，不过如此

看到这里，你可能会觉得——我们好像什么都没用？没有 create_deep_agent，没有 Backend，没有 Middleware，甚至没有 Skill。我们只是：

1. 初始化了一个模型
2. 定义了几个工具函数
3. 写了一个 while 循环

就这么简单。

一个 Agent 到底需要什么？

回顾一下我们写过的所有代码，核心就三样东西：

# 1. 一个模型
model = ChatOpenAI(model="qwen3.5:4b")

# 2. 一堆工具
tools = [write_file, read_file, ls]

# 3. 一个循环
while True:
    response = model.invoke(memory)
    if response.tool_calls:
        execute_tools(response.tool_calls)
    else:
        print(response.content)
        break

这就是 Agent 的本质。 所有的框架、抽象、中间件，都是在为这三行代码服务。

那 Skill、Backend、脚本执行是什么？

概念	本质	在我们代码中的对应
Skill	解析一个目录，把其中的文档注入到系统提示词中	如果我们在 system_prompt 里动态加载 ./skills/ 目录下的 SKILL.md，就是 Skill
Backend	文件系统操作的抽象层（本地、内存、沙盒、S3…）	我们的 write_file / read_file / ls 就是 Backend 的最简实现
脚本执行	一个 execute 工具，参数是 command，内部用 subprocess 跑命令	加一个 @tool def execute(command: str) -> str: ... 就行了

你看，拆开之后都简单。LangChain 真正让人手足无措的，从来不是“功能复杂”，而是“抽象层的路径和路由”——数据从哪来、到哪去、中间经过了谁、状态存在哪——这些才是让文档战神翻车的真正原因。

在接下来的进阶篇中，我们会做三件事：

1. 接入 create_agent：看看 LangChain 官方是如何把我们的“三行代码”封装成生产级框架的
2. 引入中间件：用 SummarizationMiddleware 替代手写的 trim_memory，用 HumanInTheLoopMiddleware 做人工审批
3. 理解抽象层的路径与路由：这也是最难啃的骨头——我们会把 Backend、Skill、子 Agent 的路由机制讲清楚，让你不再被“抽象层”卡住

记住：LangChain 的所有抽象，都是为了让 Agent 更可控、更可观测、更可扩展。 它不是来颠覆你写的这三行代码的，而是来给这三行代码穿上铠甲的。

进阶篇：从手写循环到生产级 Agent

基础篇中，我们手写了一个 ReAct 循环——模型、工具、while——用不到 200 行代码就让 Agent 跑了起来。它能对话、能读写文件、能调用工具，看起来该有的都有了。

但跑起来之后，你会发现一些问题：

• 流式体验割裂：工具调用阶段是阻塞的，用户看不到模型的思考过程
• 上下文管理粗放：trim_memory 只能按条数裁剪，粗暴地丢掉早期对话
• 没有人工审批：Agent 写了一个 rm -rf / 你只能看着它执行
• 状态无法持久化：程序崩溃，所有对话历史和工具结果全部丢失
• 调试困难：除了 print，你不知道 Agent 内部发生了什么

这些都不是“能不能跑”的问题，而是“能不能用好”的问题。基础篇我们构建了 Agent 的核心引擎，进阶篇我们要为这个引擎装上方向盘、仪表盘、安全气囊和行车记录仪——让 Agent 真正进入生产级。

接下来的章节中，我们会：

1. 读一读 create_agent 的源码——理解官方如何把我们的手写循环“图化”
2. 用 create_agent 重构 Coding Agent——把基础篇的工具迁移到 LangGraph 上
3. 接入中间件——让 Agent 拥有上下文摘要、人工审批、状态持久化等能力
4. 深入流式输出——用 astream_events 实现真正的“打字机 + 工具状态”实时反馈

读一读 create_agent 的源码

在基础篇中，我们手写了一个 ReAct 循环——模型、工具、while。现在，我们把目光投向 LangChain 官方提供的 create_agent，看看它到底做了什么，以及为什么它比我们的手写循环更强大。

先看一个最简的例子：

from langchain.agents import create_agent

agent = create_agent(
    model="openai:gpt-5.5",              # 模型
    tools=[get_weather],                 # 工具列表
    system_prompt="You are a helpful assistant",
)

# 调用
result = agent.invoke({
    "messages": [{"role": "user", "content": "San Francisco的天气怎么样？"}]
})

关键问题：为什么传入的是 dict，而不是像基础篇那样直接传消息列表？

答案：因为 create_agent 返回的不再是一个单纯的“模型”，而是一个 LangGraph 状态机（StateGraph）。而 LangGraph 的核心，就是状态机。

在基础篇中，我们的状态是一个 list——也就是 memory 列表。但 create_agent 把状态升级为一个字典（TypedDict），包含多个字段：

# LangGraph 内部的状态定义（简化版）
class AgentState(TypedDict):
    messages: list[BaseMessage]      # 对话历史（核心）
    next: str                        # 下一步要执行的节点
    memory: dict                     # 跨会话记忆（如果有）
    # ... 其他自定义字段

用户传入的 {"messages": [...]}，本质上是初始化这个状态机的输入状态，告诉它“从这些消息开始运行”。至于其他字段（如 next、memory），LangGraph 会在运行过程中自动填充，用户不需要关心。

为什么要把状态设计成字典？ 因为状态机需要管理多个维度的状态，而不只是消息列表。字典形式为状态机提供了更大的灵活性——你可以随时添加新字段来扩展状态，比如 memory、current_step、tool_results 等。

create_agent 的本质：帮你写了一个 Agent 运行时。而在基础篇中，我们自己写的 while 循环就是 Agent 的运行时（Runtime）：

# 基础篇：我们自己写的运行时
while True:
    response = model.invoke(memory)     # 1. 调用模型
    if response.tool_calls:             # 2. 判断：需要工具吗？
        execute_tools(response)         # 3. 执行工具
        memory.append(ToolMessage(...)) # 4. 把结果放回状态
    else:
        print(response.content)         # 5. 输出最终答案
        break

create_agent 底层就是用 LangGraph 帮我们把上面这个循环写成了一张图（Graph），并编译成可执行的状态机。

具体的对应关系是：

基础篇手写逻辑	create_agent 对应图节点
response = model.invoke(memory)	"model" 节点（调用 LLM）
if response.tool_calls:	条件边（Conditional Edge）
execute_tools(response)	"tools" 节点（执行工具）
memory.append(ToolMessage(...))	工具节点的输出 → 自动更新状态
while 循环	LangGraph 的边（Edge）自动形成循环

# create_agent 底层示意图（伪代码）
graph = StateGraph(AgentState)
graph.add_node("model", call_model)      # 对应我们的 model.invoke
graph.add_node("tools", execute_tools)   # 对应我们的工具执行
graph.add_conditional_edges("model", should_continue)  # 有 tool_calls → tools，否则结束
graph.add_edge("tools", "model")         # 循环回模型

所以，create_agent 的本质是：

用 LangGraph 帮我们构建了一个 Agent 状态机，把基础篇中我们自己写的 while 循环“图化”了。

把 Agent 从 while 循环升级为 LangGraph 状态图，带来的好处是：

1. 状态统一管理：所有数据（消息、中间结果、自定义状态）都在一张图的状态字典中流转，而不是散落在局部变量里。
2. 流程标准化：LLM 调用、工具执行、条件判断——全部以“节点”和“边”的形式纳入同一套调度框架。
3. 扩展方式统一：需要新增功能（比如“调用模型前做点什么”）时，只需要在图中插入新节点或中间件，而不是重写主循环。
4. 底层能力复用：图结构让 LangGraph 的检查点（Checkpoint）、持久化、流式事件、中断恢复等能力能够无缝接入。

中间件（Middleware）是 create_agent 区别于我们手写循环的核心扩展机制。

Middleware hooks run inside the compiled LangGraph that create_agent returns.

换句话说，中间件不是独立于 Agent 之外运行的；它本身就是图中插入的节点和钩子。

中间件可以做的事情很多：

钩子类型	能做什么
before_model	在每次 LLM 调用前修改提示词、注入上下文
wrap_model_call	包裹模型调用，实现缓存、重试、日志
after_model	在模型调用后处理结果
before_tool / after_tool	工具执行前后的拦截和控制
stream_handlers	拦截和转换输出流

LangChain 提供了多个预置中间件：

中间件	作用
SummarizationMiddleware	对话历史过长时自动压缩（替代我们手写的 trim_memory）
HumanInTheLoopMiddleware	敏感工具调用需要人工审批
PIIMiddleware	发送模型前脱敏敏感信息

对比基础篇中我们的手写实现：

• 我们用 trim_memory 按条数裁剪上下文 → SummarizationMiddleware 可以做更智能的摘要压缩
• 我们没有任何人工审批机制 → HumanInTheLoopMiddleware 原生支持
• 我们无法在模型调用前后插入逻辑 → 中间件钩子完美解决

现在我们开始删减代码：

import asyncio

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from pydantic import SecretStr

from tool import write_file,read_file,ls
from prompt import system_prompt

# 汇总所有工具
tools = [write_file, read_file, ls]


agent = create_agent(
        model=ChatOpenAI(
        model="qwen3.5:4b",
        base_url="http://localhost:11434/v1",
        api_key=SecretStr("ollama"),
        temperature=0.1,
    ),
    system_prompt=system_prompt,
    tools=tools
)


async def chat_loop():
    """异步聊天循环，使用 astream_events 实现流式输出与工具状态反馈"""
    print("langcode已启动。输入 /quit 退出对话。")

    while True:
        user_input = input("\n> ")
        if user_input.strip().lower() == "/quit":
            print("Goodbye!")
            break

        # 构造输入状态（字典形式，包含消息列表）
        inputs = {"messages": [{"role": "user", "content": user_input}]}

        print("\langcode: ", end="", flush=True)

        # 使用稳定的 v2 版本事件流
        async for event in agent.astream_events(inputs, version="v2"):
            if event["event"] == "on_chat_model_stream":
                # 文本增量 → 实时打印（打字机效果）
                chunk = event["data"]["chunk"]
                if chunk.content:
                    print(chunk.content, end="", flush=True)

            elif event["event"] == "on_tool_start":
                # 工具开始执行 → 打印提示
                print(f"\n🔧 调用工具: {event['name']}, 参数: {event['data']['input']}")

            elif event["event"] == "on_tool_end":
                # 工具执行完成 → 打印结果摘要
                output = event["data"]["output"]
                preview = str(output)[:80] + ("..." if len(str(output)) > 80 else "")
                print(f"✅ 工具完成: {preview}")

            # （可选）还可以处理 on_chain_end 等事件，但以上已足够

        print()  # 换行，为下一轮输入留空



if __name__ == "__main__":
    asyncio.run(chat_loop())

现在我们来看create_agent的源码，看看他帮我们做了哪些事情！

首先是模型和messages 列表：

和上文所说的一致，我们传入的model字段被封装成了langgraph的一个节点。而messages列表就位于这个state内：

至于ModelRequest则是langchain v1.0模仿web框架搞的一套生命周期，可以暂时不理会！接下来就是工具，了解langgraph的同学都比较清楚，框架帮我们封装了ToolNode来管理工具执行：

最后需要注意的一点是，在基础篇中，我们用 list 存对话历史（短期记忆），用 memory.json 存用户消息（长期记忆）。这种“内存 vs 磁盘”的二分法很容易让人产生一个误解：存在内存里就是短期，存在硬盘里就是长期。

这个理解是错的。

长短期的划分标准不是存储介质，而是会话边界（Session Boundary）。

短期记忆：属于当前会话的消息列表。它只在一次会话中有效，会话结束即失效——无论它存在哪里。

长期记忆：属于用户的、可以跨会话复用的信息。它附着于用户身份，而非某一次对话。

举个例子。假设你的 Agent 用 MongoDB 存储了某次会话中的所有消息：

会话 A（用户：张三）：
  用户：我喜欢吃苹果
  Agent：好的，记下了

会话 B（用户：张三，新会话）：
  用户：我喜欢吃什么水果？
  Agent：我不知道，你没告诉过我

虽然会话 A 的消息持久化到了 MongoDB，但它依然是短期记忆——因为它被锁定在会话 A 的边界内，会话 B 无法访问。

MongoDB 只是存储位置，会话边界才是真正的分界线。

那怎么才算长期记忆？ 当用户说“我喜欢吃苹果”时，你把这条信息写入用户画像表（user_preferences），而不是当前会话的消息列表。这样，无论用户开启多少次新会话，Agent 都能回答：“你喜欢吃苹果。”

长期记忆（跨会话）：
  user_id: "张三"
  preferences: ["喜欢吃苹果"]

会话 A：用户说“我喜欢吃苹果”
  → 写入 user_preferences

会话 B（新会话）：用户问“我喜欢吃什么？”
  → 查询 user_preferences → “你喜欢吃苹果”

文件系统抽象层——Agent 的“手”是怎么长出来的

在基础篇中，我们手写了三个工具：write_file、read_file、ls。Agent 正是通过它们，才能对代码文件进行增删改查。

但你有没有想过一个问题：这些工具是从哪来的？

答案很简单：我们写的。但更深层的问题是：这些工具的本质是什么？

它们的本质，是对文件系统（File System）的操作封装。

基础篇中，我们的工具直接操作本地磁盘：

@tool
def write_file(path: str, content: str) -> str:
    file_path = Path(path)
    file_path.parent.mkdir(parents=True, exist_ok=True)
    file_path.write_text(content, encoding="utf-8")
    return f"✅ 文件已保存到：{path}"

这有什么问题吗？在本地跑跑 demo 没问题。但一旦进入生产环境，问题就来了：

• 如果我想让 Agent 在沙盒（Sandbox） 里操作文件，而不是直接访问我的磁盘呢？
• 如果我想让 Agent 读写 S3 或 MinIO 上的文件呢？
• 如果我想让 Agent 在内存文件系统中测试，避免污染本地目录呢？

基础篇的写法把“文件操作”和“本地磁盘”硬绑定在一起。每换一种存储方式，就要重写一套工具。

create_agent 的设计哲学是：把“文件操作”抽象为一个接口（Interface），具体的存储方式作为实现（Implementation）。

这个接口就是 FilesystemMiddleware，它默认提供了多个工具：

工具	功能
ls	列出目录内容
read_file	读取文件内容
write_file	写入文件内容

而这几个工具的操作对象，不是直接指向本地磁盘，而是指向一个 Backend（后端）。

Backend = 文件系统接口的具体实现

只要实现了 Backend 接口，无论底层是本地磁盘、内存、S3、还是远程沙盒，Agent 都能用同一套工具进行操作。

在 deepagents.backends 中，LangChain 提供了丰富的 Backend 实现：

Backend	数据存储位置	适用场景
StateBackend	LangGraph 状态字典中	默认 Backend。数据随会话存在，会话结束即消失。适合无状态测试。
FilesystemBackend	本地磁盘（指定 root_dir）	本地开发，直接读写项目文件
LocalShellBackend	本地磁盘 + Shell 命令执行	FilesystemBackend 的增强版，额外提供 execute 工具（可直接运行 Shell 命令）
StoreBackend	LangGraph 的 BaseStore（跨会话持久化存储）	跨会话记忆、用户偏好、全局配置
CompositeBackend	多个 Backend 的组合	路由核心——不同路径走不同 Backend。例如 /memories/ 走 StoreBackend，/workspace/ 走 FilesystemBackend
LangSmithSandbox	LangSmith 云沙盒	云端隔离环境，安全执行代码
ContextHubBackend	上下文中心	特殊用途，用于多 Agent 共享文件上下文
BackendContext	上下文对象	辅助类，保存当前 Backend 的运行时上下文
BackendProtocol	Protocol 类（接口定义）	定义 Backend 的“契约”——所有 Backend 必须实现的方法

在进阶篇中，我们不再需要手写 write_file、read_file、ls。FilesystemMiddleware 会帮我们搞定一切。

from langchain.agents import create_agent
from langchain.agents.middleware import FilesystemMiddleware
from deepagents.backends import FilesystemBackend

backend = FilesystemBackend(root_dir="./workspace")

agent = create_agent(
    model=model,
    middleware=[
        FilesystemMiddleware(backend=backend),
    ],
)

当你把 backend 换成 StateBackend 时，Agent 的所有文件操作都会变成内存操作，不会写入磁盘：

from deepagents.backends import StateBackend

backend = StateBackend()

agent = create_agent(
    model=model,
    middleware=[
        FilesystemMiddleware(backend=backend),
    ],
)

切换只需一行代码。这就是抽象的力量。

这里顺便展开说一下“内存文件系统”。它的实现非常简单，本质上就是一个字典：

class MemoryFilesystemBackend:
    def __init__(self):
        self.files: dict[str, str] = {}          # 路径 → 内容
        self.dirs: set[str] = set()              # 已存在的目录

    def write_file(self, path: str, content: str) -> str:
        self.files[path] = content
        # 自动创建父目录
        parent = "/".join(path.split("/")[:-1])
        if parent:
            self.dirs.add(parent)
        return f"✅ 已写入：{path}"

    def read_file(self, path: str) -> str:
        if path not in self.files:
            return f"❌ 文件不存在：{path}"
        return self.files[path]

    def ls(self, path: str = "/") -> list[str]:
        return [f for f in self.files.keys() if f.startswith(path)]

你看，文件系统从来就不是“磁盘”的同义词——它是一个“路径 → 内容”的映射。只要实现了这个映射，无论背后是磁盘、内存、S3 还是数据库，对 Agent 来说都一样。

不同 Backend 的代码示例

StateBackend（默认，纯内存）：

from deepagents.backends import StateBackend

backend = StateBackend()

FilesystemBackend（本地磁盘）：

from deepagents.backends import FilesystemBackend

backend = FilesystemBackend(root_dir="./workspace")

LocalShellBackend（本地磁盘 + Shell 执行）：

from deepagents.backends import LocalShellBackend

backend = LocalShellBackend(root_dir="./workspace")
# Agent 现在可以用 execute 工具运行 Shell 命令

CompositeBackend（路由到不同 Backend）：

from deepagents.backends import CompositeBackend, StateBackend, StoreBackend

backend = CompositeBackend(
    default=StateBackend(),                    # 默认走内存
    namespace_map={
        "/memories": StoreBackend(store=store),  # /memories/ 路径走持久化存储
        "/workspace": FilesystemBackend(root_dir="./workspace"),
    }
)
# Agent 写入 /memories/user_pref.txt → 走 StoreBackend（跨会话）
# Agent 写入 /workspace/main.py → 走 FilesystemBackend（本地磁盘）
# Agent 写入 /tmp/123.txt → 走 StateBackend（默认，纯内存）

LocalShellBackend 的“危险”之处：

LocalShellBackend = FilesystemBackend + execute 工具。它允许 Agent 直接执行 Shell 命令：

# Agent 可以调用 execute 工具
@tool
def execute(command: str) -> str:
    return subprocess.run(command, shell=True, capture_output=True)

这就是为什么我们在基础篇会遇到 D:/xxx 路径错误——LocalShellBackend 的 execute 工具跑在真实操作系统上，而文件操作工具跑在虚拟文件系统上。两条路径，一套工具，路由不清晰，就出 bug 了。

在进阶篇中，我们会更系统地理解这些 Backend 的路由机制，避免类似问题。

核心理解：路径 → Backend 的映射

先看一个关键认知：

所有文件系统操作都基于路径（Path）。而“路由（Routing）”就是根据路径前缀，决定把操作交给哪个 Backend。

DeepAgents 中的 CompositeBackend 就是干这件事的。它本质上是一个 路由表（Routing Table）：

CompositeBackend(
    default=StateBackend(),                    # 默认走内存
    namespace_map={
        "/memories": StoreBackend(store=store),  # 路径前缀匹配
        "/workspace": FilesystemBackend(root_dir="./workspace"),
    }
)

路由逻辑是：

请求路径	匹配哪个 Backend
/workspace/main.py	FilesystemBackend（匹配 /workspace）
/memories/user_pref.txt	StoreBackend（匹配 /memories）
/tmp/cache.json	StateBackend（default，无匹配）

这就是为什么基础篇你设置 virtual_mode=True 还会出现 D:/xxx 路径错误——因为 LocalShellBackend 的 execute 工具跑在真实操作系统上（绝对路径），而文件操作工具跑在虚拟文件系统上（相对路径）。两条路径体系不对齐，就出了 bug。

CompositeBackend 和 execute工具做了什么

我们先来查看FilesystemMiddleware，这个中间件实现的执行工具：_create_execute_tool：

注意源码中红框标记，execute tool的执行能力并非是由FilesystemMiddleware直接实现的，而是借助了backend的抽象，那么我们就需要观察deepagents的backend，到底哪个backend实现了execute功能：

from deepagents.backends import (
    CompositeBackend,LocalShellBackend,StateBackend,FilesystemBackend,ContextHubBackend,StoreBackend
)

首先来看FilesystemBackend：

class FilesystemBackend(BackendProtocol):
    """直接从文件系统读写文件的后端。

    文件通过其真实的文件系统路径进行访问。相对路径基于当前工作目录进行解析。
    内容以纯文本形式读写，元数据（时间戳）来源于文件系统的 stat 信息。

    !!! warning "安全警告"

        此后端赋予 Agent 直接读写文件系统的权限。请谨慎使用，且仅限在适当的
        环境中使用。

        **适当的使用场景：**

        - 本地开发 CLI（编码助手、开发工具）
        - CI/CD 流水线（见下文安全考量）

        **不适当的使用场景：**

        - Web 服务器或 HTTP API —— 请使用 `StateBackend`、`StoreBackend` 或
            `SandboxBackend`

        **安全风险：**

        - Agent 可以读取任何可访问的文件，包括密钥（API Key、凭证、`.env` 文件）
        - 结合网络工具，密钥可能通过 SSRF 攻击被外泄
        - 文件修改是永久性的且不可撤销

        **推荐的防护措施：**

        1. 启用 Human-in-the-Loop（HITL）中间件来审查敏感操作
        2. 将密钥排除在可访问的文件系统路径之外（尤其是在 CI/CD 中）
        3. 生产环境优先使用 `StateBackend`、`StoreBackend` 或 `SandboxBackend`

        总的来说，我们期望此后端与 Human-in-the-Loop（HITL）中间件配合使用，
        或者如果你需要运行不受信任的工作负载，在适当沙盒化的环境中使用。

        !!! note

            `virtual_mode=True` 主要用于虚拟路径语义（例如与 `CompositeBackend`
            配合使用）。它也可以通过阻止路径穿越（`..`、`~`）和 `root_dir` 之外
            的绝对路径来提供基于路径的防护，但它不提供沙盒或进程隔离。
            默认值（`virtual_mode=False`）即使设置了 `root_dir` 也不提供任何安全性。
    """

    def __init__(
        self,
        root_dir: str | Path | None = None,
        virtual_mode: bool | None = None,
        max_file_size_mb: int = 10,
    ) -> None:
        """初始化文件系统后端。

        Args:
            root_dir: 文件操作的根目录（可选）。

                默认为当前工作目录。

                - 当 `virtual_mode=False`（默认）时：仅影响相对路径解析。
                - 当 `virtual_mode=True` 时：作为文件系统操作的虚拟根目录。

            virtual_mode: 启用虚拟路径模式。

                **主要使用场景：** 当与 `CompositeBackend` 配合使用时，提供稳定的、
                与后端无关的路径语义。`CompositeBackend` 会剥离路由前缀，并将
                规范化后的路径转发给路由到的后端。

                当为 `True` 时，所有路径都被视为锚定在 `root_dir` 下的虚拟路径。
                路径穿越（`..`、`~`）会被阻止，所有解析后的路径都会被验证
                保持在 `root_dir` 内。

                当为 `False`（默认）时，绝对路径原样使用，相对路径在 `root_dir`
                下解析。这无法防止 Agent 选择 `root_dir` 之外的路径。

                - 绝对路径（例如 `/etc/passwd`）完全绕过 `root_dir`
                - 带有 `..` 的相对路径可以逃逸出 `root_dir`
                - Agent 拥有不受限制的文件系统访问权限

            max_file_size_mb: 操作（如 grep 的 Python 回退搜索）的最大文件大小（MB）。

                超过此限制的文件在搜索时会被跳过。默认为 10 MB。
        """
        self.cwd = Path(root_dir).resolve() if root_dir else Path.cwd()
        if virtual_mode is None:
            warn_deprecated(
                since="0.5.0",
                removal="0.6.0",
                message=(
                    "`FilesystemBackend` 的 `virtual_mode` 默认值将在 "
                    "deepagents==0.6.0 中变更；请显式指定 `virtual_mode`。"
                    "注意：`virtual_mode` 用于虚拟路径语义（例如 `CompositeBackend` "
                    "路由）和可选的基于路径的防护；它不提供沙盒或进程隔离。"
                    "安全提示：保持 `virtual_mode=False` 允许绝对路径和 `'..'` "
                    "绕过 `root_dir`。详情请参阅 API 参考文档。"
                ),
                package="deepagents",
            )
            virtual_mode = False
        self.virtual_mode = virtual_mode
        self.max_file_size_bytes = max_file_size_mb * 1024 * 1024

这行代码是 FilesystemBackend 初始化时的工作目录（Working Directory）锚点设定。我们把它拆开看：

self.cwd = Path(root_dir).resolve() if root_dir else Path.cwd()

1. if root_dir else ...（条件判断）

   • 如果传入了 root_dir 参数（比如 "./workspace"），走前半段。
   • 如果没传（root_dir=None），走后半段，直接使用当前 Python 进程的工作目录。

2. Path(root_dir).resolve()（传入了路径时）

   • Path(root_dir)：把字符串转成 pathlib.Path 对象。
   • .resolve()：将路径解析为绝对路径，同时会规范化（移除 ..、.）并解析符号链接。
   • 比如传入 "./workspace"，当前在 D:/my_project/，那么 Path("./workspace").resolve() 会得到 D:/my_project/workspace。

3. Path.cwd()（没传入路径时）

   • 直接获取当前 Python 进程的工作目录（Current Working Directory）。

self.cwd 存储了这个绝对化的根目录锚点。后续所有相对路径的解析都会基于它。

• 当 virtual_mode=False 时：相对路径（如 src/main.py）会拼接成 self.cwd / "src/main.py"，但绝对路径（如 /etc/passwd）不受影响，直接使用。
• 当 virtual_mode=True 时：所有路径（包括绝对路径）都会被强制锚定到 self.cwd，并检查是否在其范围内。

假设你的 Python 程序跑在 D:/my_project，代码里写：

backend = FilesystemBackend(root_dir="./workspace")

执行时：

1. root_dir = "./workspace"
2. Path("./workspace").resolve() → D:/my_project/workspace
3. self.cwd = D:/my_project/workspace

之后 Agent 写文件 "test.py"：

• 如果 virtual_mode=False：实际路径是 D:/my_project/workspace/test.py（因为相对路径拼接到 self.cwd）
• 如果 virtual_mode=True：实际路径同样是 D:/my_project/workspace/test.py，但如果 Agent 想写 "../evil.py"，virtual_mode=True 会拦截，virtual_mode=False 则放行，导致路径变成 D:/my_project/evil.py，逃逸出了 workspace。

需要注意的是，FilesystemBackend 没有实现execute接口，这个接口位于沙箱接口中，因此文件增删改查是一个backend类型，文件执行是另外一个backend类型，因为后者风险系数更高！！

而唯一一个实现了沙箱协议的Backend就是LocalShellBackend，如果想让agent有执行代码的能力，你离不开他：

class LocalShellBackend(FilesystemBackend, SandboxBackendProtocol):
    """具备无限制本地 Shell 命令执行能力的文件系统后端。

    此后端在 `FilesystemBackend` 的基础上增加了 Shell 命令执行能力。
    命令直接在主机系统上执行，没有任何沙箱、进程隔离或安全限制。

    !!! warning "安全警告"

        此后端赋予 Agent **同时**操作本地文件系统 **和** 无限制执行 Shell 命令的能力。
        请极度谨慎使用，且仅限在适当的场景中使用。

        **适当的使用场景：**

        - 本地开发 CLI（编码助手、开发工具）
        - 个人开发环境（你信任 Agent 代码）
        - CI/CD 流水线（需配合良好的密钥管理）

        **不适当的使用场景：**

        - 生产环境（如 Web 服务器、API、多租户系统）
        - 处理不可信的用户输入或执行不可信的代码

        生产环境请使用 `StateBackend`、`StoreBackend`，或继承 `BaseSandbox` 自行实现隔离后端。

        **安全风险：**

        - Agent 可以以你的用户权限执行 **任意 Shell 命令**
        - Agent 可以读取 **任何可访问的文件**，包括密钥（API Key、凭证、`.env` 文件、SSH 密钥等）
        - 结合网络工具，密钥可能通过 SSRF 攻击被外泄
        - 文件修改和命令执行是 **永久且不可逆的**
        - Agent 可以安装软件包、修改系统文件、启动进程等
        - **无进程隔离**——命令直接跑在你的主机上
        - **无资源限制**——命令可以消耗无限的 CPU、内存、磁盘

        **推荐的防护措施：**

        由于 Shell 访问无限制，可以绕过文件系统限制，强烈建议：

        1. **启用 Human-in-the-Loop（HITL）中间件**，在执行前审查并批准所有操作。
        2. 仅在专用的开发环境中运行，绝不要在共享或生产系统上使用。
        3. 绝不要暴露给不可信用户，或允许执行不可信代码。
        4. 对于需要执行代码的生产环境，继承 `BaseSandbox` 创建隔离后端（Docker容器、虚拟机或其他沙盒环境）。

        !!! note

            当启用了 Shell 访问时，`virtual_mode=True` 和基于路径的限制**无法提供任何安全性**，
            因为命令可以直接访问系统上的任何路径。
    """

__init__ 参数详细讲解

1. root_dir: str | Path | None = None

• 作用：文件系统操作和 Shell 命令的工作目录。
• 如果未提供，默认为当前 Python 进程的工作目录。
• 所有 Shell 命令都会以这个目录作为当前工作目录（即 subprocess 的 cwd）。
• 对于文件操作（继承自 FilesystemBackend）：
  • 当 virtual_mode=False（默认）时，路径按真实文件系统处理，Agent 可以用绝对路径或 .. 逃逸出 root_dir。
  • 当 virtual_mode=True 时，它成为文件操作的虚拟根目录，路径穿越被阻止。
• 注意：root_dir 对 Shell 命令没有限制作用——命令可以访问系统任何路径。

2. virtual_mode: bool | None = None

• 作用：为文件系统操作启用虚拟路径模式。
• 当为 True 时，root_dir 成为虚拟根目录，所有文件路径（包括绝对路径）都被解释为相对于 root_dir，.. 和 ~ 被阻止。
• 主要设计目的是配合 CompositeBackend 的路由机制，让文件操作在不同 Backend 之间保持路径语义一致。
• 重要：virtual_mode 只影响文件操作，不影响 Shell 命令（execute）。即使设为 True，Shell 命令依然可以访问任何路径。
• 从 deepagents==0.5.0 开始，该参数被标记为弃用（默认值将在 0.6.0 变更），建议显式指定。

3. timeout: int = DEFAULT_EXECUTE_TIMEOUT

• 作用：Shell 命令执行的默认超时时间（秒）。
• 默认值为 120 秒（2 分钟）。
• 如果命令执行超过此时间，进程会被终止。
• 可以在调用 execute() 时通过参数覆盖此默认值。

4. max_output_bytes: int = 100_000

• 作用：捕获命令输出（stdout + stderr）的最大字节数。
• 默认 100,000 字节（约 100KB）。
• 超过此大小的输出会被截断，防止 Agent 产生海量输出导致内存问题。

5. env: dict[str, str] | None = None

• 作用：为 Shell 命令指定的环境变量字典。
• 如果 inherit_env=False（默认），则命令只会获得 env 中定义的变量。
• 如果 inherit_env=True，则 env 中的变量会覆盖继承自父进程的环境变量。

6. inherit_env: bool = False

• 作用：是否继承父进程（你的 Python 程序）的全部环境变量。
• 当为 False（默认）时，命令的环境变量仅限于 env 字典中显式设置的变量（可能是空字典）。
• 当为 True 时，命令会获得 os.environ 的全部变量，然后应用 env 中的覆盖。
• 安全提示：继承父进程环境可能泄露敏感信息（如 AWS_SECRET_ACCESS_KEY），在 CI/CD 或共享环境中建议设为 False 并手动配置所需变量。

然后我们来查看其实现的execute 工具：

构建生产级复合 Backend——Agent 的完整“工具箱”

在了解了 FilesystemBackend、LocalShellBackend、StoreBackend、StateBackend 和 CompositeBackend 之后，是时候把它们组合起来，构建一个真正可用于生产环境的 Agent 文件系统了。

本节我们将从零开始，构建一个生产级的复合 Backend——它能读写文件、能执行脚本、能跨会话记忆，还能支持 Skill 加载和路径路由。

在生产环境中，一个 Coding Agent 的文件系统至少需要满足：

需求	说明	对应的 Backend
文件读写	读取/写入/编辑/删除文件	FilesystemBackend 或 LocalShellBackend
脚本执行	运行 Shell 命令、执行脚本	LocalShellBackend（本地）或沙盒方案
跨会话持久化	用户偏好、项目配置需要跨会话保留	StoreBackend
临时工作区	中间结果、缓存文件随会话自动清理	StateBackend
路径路由	不同用途的路径走不同的存储后端	CompositeBackend
Skill 加载	Agent 能按需加载领域知识和脚本	SkillsMiddleware + Backend
可观测性	能看到 Agent 在文件系统上做了什么	通过 LangSmith、控制台日志或回调实现

在开始构建 Backend 之前，我们需要先理解 Skill 和 Backend 的关系。

Skill 的本质：Skill 是一个目录，里面包含一个 SKILL.md 文件（描述 Skill 的用途、触发条件、使用方式）和若干脚本文件。SkillsMiddleware 会扫描指定目录，读取所有 SKILL.md 文件，将其中的元信息（名称、描述、触发条件）注入到 Agent 的 System Prompt 中，让 Agent 知道“在什么情况下应该加载哪个 Skill”。

Skill 与 Backend 的关系：Skill 的 SKILL.md 和脚本文件存储在某个 Backend 中。具体存在哪里，由 CompositeBackend 的路由规则决定。例如，我们可以把 Skill 放在 /skills/ 路径下，由 FilesystemBackend 或 StoreBackend 托管，让 Agent 可以读取它们。

我们直接去github把anthropic的skills全部拖下来：

简单总结：

• SkillsMiddleware 负责“把 Skill 变成提示词”（控制层）
• Backend 负责“存放 Skill 的文件”（存储层）
• CompositeBackend 负责“路由 Skill 文件的读写请求”（路由层）

我们先设计一套清晰的路径前缀规则：

路径前缀	Backend	用途	生命周期
/workspace/	LocalShellBackend	项目文件读写 + 脚本执行	持久化到本地磁盘
/skills/	FilesystemBackend	Skill 定义文件（SKILL.md + 脚本）	持久化到本地磁盘，只读
/memories/	StoreBackend	用户偏好、长期记忆	跨会话永久持久化
/tmp/（默认）	StateBackend	中间结果、临时缓存	当前会话

第二步：准备各个 Backend 组件

from deepagents.backends import CompositeBackend, FilesystemBackend, LocalShellBackend, StateBackend, StoreBackend
from langgraph.store.memory import InMemoryStore

# 1. 工作区 Backend（本地磁盘 + Shell 执行）
workspace = LocalShellBackend(
    root_dir="./workspace",
    virtual_mode=True,
    timeout=60,
    max_output_bytes=100_000,
    inherit_env=True,
)

# 2. Skill 存储 Backend（只读文件系统）
skills_fs = FilesystemBackend(
    root_dir="./skills",  # 你的 Skill 目录
    virtual_mode=True,
)

# 3. 持久化存储 Backend（跨会话记忆）
store = InMemoryStore()  # 生产环境替换为 RedisStore / PostgresStore
memories = StoreBackend(store=store)

# 4. 临时存储 Backend（会话级）
ephemeral = StateBackend()

第三步：用 CompositeBackend 组合

backend = CompositeBackend(
    default=ephemeral,                 # /tmp/ 或未匹配路径 → 内存
    routes={
        "/workspace/": workspace,      # /workspace/ → 本地磁盘 + Shell
        "/skills/": skills_fs,         # /skills/ → Skill 文件存储
        "/memories/": memories,        # /memories/ → 跨会话持久化
    }
)

注意：Agent 不能在 /skills/ 路径下执行脚本——因为 FilesystemBackend 压根没有 execute 工具。

一个典型的 Skill 执行流程是这样的：

1. SkillsMiddleware 扫描 /skills/ 目录，读取所有 SKILL.md
   → 把 Skill 元信息注入 System Prompt
   → Agent 知道：有一个 "fastapi" Skill 可用

2. 用户说："帮我用 fastapi 写一个接口"
   → Agent 从 System Prompt 中知道 Skill 存在
   → Agent 读取 /skills/fastapi/script.py（通过 FilesystemBackend）

3. Agent 判断需要执行脚本
   → 但 /skills/ 目录没有 execute 工具！
   → Agent 把脚本内容写入 /workspace/run.py（通过 LocalShellBackend）
   → 然后 execute python /workspace/run.py（通过 LocalShellBackend）

4. 脚本在 /workspace/ 中执行
   → 结果返回给 Agent

为什么不直接在 skill 目录执行？

1. 权限分离：/skills/ 被设计为只读知识库，用于存储 Skill 定义。/workspace/ 被设计为可执行工作区，用于运行脚本。职责清晰。

2. 安全性：FilesystemBackend 没有 execute 工具，Agent 无法在 Skill 目录里执行任何命令，防止恶意 Skill 在加载时就执行危险操作。

3. 可追溯性：所有执行产生的文件（输出、日志、中间结果）都在 /workspace/ 下，方便追踪和清理。

假设你有一个 fastapi Skill，目录结构是：

./skills/fastapi/
├── SKILL.md          # Skill 描述
└── templates/
    └── main.py       # 模板代码（不是可执行脚本，只是代码模板）

当用户说“用 fastapi 写一个 hello world 接口”时：

# 1. Agent 从 Prompt 中知道 fastapi Skill 存在
# 2. Agent 读取 /skills/fastapi/templates/main.py（模板）
# 3. Agent 根据模板生成实际代码
# 4. Agent 把代码写入 /workspace/main.py
# 5. Agent 执行 python /workspace/main.py
# 6. 结果返回用户

模板本身不被执行——它只是被读取、参考，然后由 Agent 基于模板生成代码，再在工作区执行。

第四步：添加 SkillsMiddleware

SkillsMiddleware 需要指定 Skill 的存放路径，它会扫描该路径下的所有 SKILL.md 文件，并将元信息注入 System Prompt。

from deepagents.middleware import SkillsMiddleware

skills_middleware = SkillsMiddleware(
    sources=["/skills/"],   # 注意：这是虚拟路径，由 CompositeBackend 路由
    backend=backend,         # 使用我们刚创建的复合 Backend
)
fs_middleware = FilesystemMiddleware(backend=backend)

第五步：创建 Agent

from langchain.agents import create_agent 
from langchain_openai import ChatOpenAI
from pydantic import SecretStr

model = ChatOpenAI(
    model="qwen3.5:4b",
    base_url="http://localhost:11434/v1",
    api_key=SecretStr("ollama"),
    temperature=0.1,
)

agent = create_agent(
    model=model,
    system_prompt="你是一个安全的编码助手。",
    backend=backend,
    middleware=[skills_middleware],  # 👈 注入 SkillsMiddleware
)

当 Agent 使用这个复合 Backend 时，它完全不知道自己面对的是四个不同的存储后端——它只看到“路径”。

场景 1：写代码并运行（走工作区）

用户："帮我写一个 hello world 脚本，保存到 /workspace/hello.py，然后运行它"
Agent：写入 /workspace/hello.py → 路由到 LocalShellBackend
Agent：execute python /workspace/hello.py → 在本地 Shell 执行
输出：Hello, World!

场景 2：加载 Skill 知识（走 Skill 存储）

用户："如何用 FastAPI 创建 REST API？"
Agent：检测到 FastAPI 相关 → 从 /skills/fastapi/ 读取 SKILL.md
Agent：将 Skill 内容注入上下文 → 基于 Skill 指导回答

场景 3：记住用户偏好（走持久化）

用户："记住，我喜欢用 pytest 而不是 unittest"
Agent：写入 /memories/user_prefs.json → 路由到 StoreBackend
# 下次会话，Agent 依然能读到这条偏好

场景 4：临时缓存（走内存）

Agent：写入 /tmp/cache.json → 路由到 StateBackend（默认）
# 会话结束，缓存自动消失，无需手动清理

为了让 SkillsMiddleware 正常工作，你的 Skill 目录结构应该遵循以下规范：

./skills/
├── fastapi/
│   └── SKILL.md              # FastAPI 使用指南
├── pytest/
│   └── SKILL.md              # Pytest 测试指南
├── docker/
│   ├── SKILL.md              # Docker 使用指南
│   └── scripts/
│       └── deploy.sh         # 辅助脚本
└── README.md                 # （可选）Skill 目录说明

每个 SKILL.md 文件的内容格式通常为：

---
name: FastAPI
description: 用于构建快速 API 的现代 Python Web 框架
triggers:
  - "fastapi"
  - "rest api"
  - "web framework"
---

# FastAPI Skill

## 适用场景
当用户需要创建 REST API、Web 服务或后端接口时，使用本 Skill。

## 使用方式
1. 导入 FastAPI 和 uvicorn
2. 创建 FastAPI 应用实例
3. 定义路由和端点
4. 启动服务

## 最佳实践
- 使用 Pydantic 进行数据验证
- 使用依赖注入管理数据库会话
- 使用中间件处理跨域和日志

系统提示词（模仿官方文档）：

system_prompt = """你是 langcode，一个专业的编程助手。

## 核心原则（最重要！！！）

⚠️ **描述 ≠ 执行**

你说"让我读取文件"，就必须**立即调用 `read_file` 工具**。
你说"我来检查一下"，就必须**立即调用 `ls` 工具**。
你说"我来写个代码"，就必须**立即调用 `write_file` 工具**。

**禁止**：仅仅用自然语言描述你的计划而不执行工具。
**正确做法**：直接调用工具，用自然语言解释你通过工具发现了什么。

You have access to a structured virtual filesystem with multiple storage backends mapped to specific paths:

- `/workspace/`: Local disk storage for your active project files, code, and working documents. Use this directory for file creation, editing, and executing shell or tool operations.
- `/skills/`: Read-only or managed storage containing reusable skills and custom agent instructions. Check here when looking for predefined capabilities.
- `/memories/`: Long-term persistent storage backed by a cross-thread store. Save important user preferences, key insights, and accumulated knowledge here (e.g., `/memories/user_preferences.txt`) so they persist across different sessions and threads.
- Default (`/` or other paths): Ephemeral memory scoped strictly to the current thread.

Always use the appropriate directory path according to whether data needs to be transient, local-disk backed, or long-term persistent.

## 工作流程（每次都要走完）

1. 理解用户需求
2. **立即调用工具**获取必要信息（ls、read_file 等）
3. 基于工具返回的结果，给出答案或执行下一步
4. 如果继续需要信息，回到步骤 2

## 可用工具

| 工具 | 用途 | 何时使用 |
| :--- | :--- | :--- |
| `ls(path)` | 列出目录内容 | 第一次接触任何目录时先用 |
| `read_file(path)` | 读取文本文件 | 用户提到任何文件名时 |
| `write_file(path, content)` | 写入文件 | 生成超过 5 行代码时 |
| `grep(pattern, path)` | 搜索文件内容 | 在代码中查找关键字 |
| `glob(pattern, path)` | 按通配符查找文件 | 批量查找文件 |

## 工具执行规则

1. `ls` 必须带参数，如 `ls("/workspace/")`
2. `read_file` 之前先用 `ls` 确认文件存在
3. 路径**禁止**使用 `C:\\`、`D:\\`、`\` 等 Windows 风格
4. 路径始终用正斜杠：`/workspace/src/main.py`
5. **禁止输出超过 5 行的代码到聊天框**，超过 5 行必须用 `write_file` 保存

## 输出风格

- 极其简洁：不要重复用户的问题
- 不要输出"好的，我来帮你..."这类客套话
- 直接给结果：要么是工具调用，要么是简洁的回答
- 如果工具返回了数据，基于数据回答，不要编造

## 安全

- 禁止执行 `rm -rf`、`del /f`、`shutdown` 等危险命令
- 禁止读取 `.env`、`/etc/passwd`、`~/.ssh/` 等敏感路径
- 不确定的路径先用 `ls` 确认

## 立即行动

用户发来请求后，第一句话就调用工具，不要说"我来帮你"之类的废话。

示例：
用户："帮我分析 x.xlsx"
→ 你：[调用 ls("/workspace/")] → 确认文件存在
→ 你：[调用 read_file("/workspace/x.xlsx")] → 读取内容
→ 你：基于读取结果分析

用户："写一个 FastAPI hello world"
→ 你：[调用 write_file("/workspace/main.py", content="...")] → 保存文件
→ 你：文件已保存到 /workspace/main.py，运行 uvicorn main:app 启动

现在开始。每个工具调用都要有明确的目的，执行完工具后基于结果回答用户。"""

一切就绪，但作者遇到了一个让人崩溃的问题

代码写好了，Backend 配置好了，SkillsMiddleware 也挂载上了，CompositeBackend 的路由规则清晰，astream_events 的事件监听也调通了。一切看起来都完美无缺——理论上，Agent 应该能准确识别用户的意图，按需加载 Skill，调用工具，完成编码任务。

但实际上，它就是不干活。

具体表现是：当我输入“workspace 下有一个 x.xlsx，帮我分析下内容”时，Agent 会输出“我来帮你读取这个 Excel 文件”，然后……就没有然后了。它没有调用 read_file，没有调用 ls，更没有加载 xlsx Skill。它只是“说”了它想做什么，却没有真正“做”。

我开始怀疑人生。检查 SkillsMiddleware 的配置，检查 System Prompt 是不是写得不够清晰，检查 CompositeBackend 的路由规则，甚至重写了 astream_events 的循环——结果全都无济于事。我一度以为是代码有问题，或者是 Backend 抽象层的某个环节出了 bug。

直到我换了一个模型。

我把本地的 qwen3.5:4b 换成了 OpenRouter 上的 openai/gpt-oss-120b，其他代码一行没改。结果一切恢复正常了——Agent 开始调用 ls 确认文件存在，然后调用 read_file 读取内容，按 Skill 的指导完成了整个流程。

问题不在代码，不在配置，在于模型本身。

小模型能处理单步任务，能写代码片段，能聊天。但当它需要理解复杂的 System Prompt、结合多个工具、判断何时加载 Skill、如何按步骤执行时，它的“理解力”就跟不上了。它会输出正确的自然语言描述，却不会真正执行动作——这是在“描述计划”，而不是在“执行计划”。

模型能力的天花板，是硬约束。 代码可以调优，Prompt 可以优化，但模型如果根本不知道“调用 Skill”这件事的意义，架构层面做再多也是白搭。

经验教训：先选对模型，再优化架构。 大模型把小模型的逻辑跑通之后，再考虑要不要为了成本降级——至少你知道“什么行为是对的”，而不是在小模型的错误反馈里死循环。

换成 openai/gpt-oss-120b（OpenRouter 上的 120B 大参数模型）后，问题立刻消失。Agent 不再“嘴炮”——它不再用自然语言描述“我来读取文件”，而是直接调用 read_file 和 ls 工具，按 Skill 的指导完成整个流程。

这个是我要求120b模型写的一个前端界面，可以看到页面效果非常的anthropic：

用 TUI 给 Agent 一个“体面的脸”

我们的 Agent 已经能干活了——读文件、写代码、查 Skill、调用工具——但它的“脸面”还停留在 print 阶段：黑白文字、杂乱无章、工具调用和模型输出混在一起，分不清谁是谁。

这一节，我们用 prompt_toolkit + Rich 给它换一张脸。

为什么是 prompt_toolkit + Rich？

这两个库的组合，是目前 Python TUI 生态中最成熟的“黄金搭档”：

库	职责	擅长的
prompt_toolkit	输入交互	多行输入、历史记录、自动补全、快捷键
Rich	输出渲染	彩色文字、表格、Markdown、语法高亮

它们各司其职：prompt_toolkit 管“怎么输入”，Rich 管“怎么展示”。组合起来，就能造出一个类似 Claude Code 风格的终端界面。

完整代码：

import asyncio

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from pydantic import SecretStr
from deepagents.middleware import SkillsMiddleware,FilesystemMiddleware
from langchain.messages import HumanMessage, AIMessage
from rich.console import Console
from rich.live import Live
from rich.markdown import Markdown
from rich.panel import Panel
from rich.spinner import Spinner
from rich.text import Text

from backend import backend
from prompt import system_prompt


skills_middleware = SkillsMiddleware(
    sources=["/skills/"],   # 注意：这是虚拟路径，由 CompositeBackend 路由
    backend=backend,         # 使用我们刚创建的复合 Backend
)
fs_middleware = FilesystemMiddleware(backend=backend)

agent = create_agent(
    model=ChatOpenAI(
        model="openai/gpt-oss-120b",
        base_url="https://openrouter.ai/api/v1",
        api_key=SecretStr(""),
        temperature=0.6,
    ),
    middleware=[fs_middleware,skills_middleware],
    system_prompt=system_prompt,
)

_console = Console()

def _display_banner() -> None:
    """展示启动横幅。"""
    title = Text("Attack Chain Data Production Engine", style="bold bright_white")
    subtitle = Text.assemble(
        ("  /list ", "bold cyan"), ("查看场景  ", "dim"),
        ("  /quit ", "bold cyan"), ("退出", "dim"),
    )
    _console.print(Panel(
        Text.assemble(title, "\n", subtitle),
        border_style="bright_blue",
        padding=(1, 2),
    ))
    _console.print()


def _display_progress(event: dict) -> None:
    """在终端展示生产进度事件。"""
    ts = event.get("timestamp", "")
    content = event.get("content", "")
    event_type = event.get("type", "progress")

    style_map = {"progress": "cyan", "result": "green", "error": "red"}
    style = style_map.get(event_type, "cyan")
    _console.print(f"  [{ts}] {content}", style=style)


async def chat_loop() -> None:
    """交互式对话主循环，流式输出 LLM 回复和生产进度。"""
    messages: list = []

    _display_banner()

    while True:
        try:
            user_input = _console.input("[bold green]> [/]").strip()
        except (EOFError, KeyboardInterrupt):
            _console.print("\n再见!")
            break

        if not user_input:
            continue
        if user_input.lower() in ("/quit", "/exit", "quit", "exit"):
            _console.print("再见!")
            break

        messages.append(HumanMessage(content=user_input))

        _console.print()
        try:
            full_response = ""
            first_token = False

            with Live(
                    Spinner("dots", text="思考中...", style="cyan"),
                    console=_console,
                    refresh_per_second=8,
            ) as live:
                async for mode, chunk in agent.astream(
                        input={"messages": messages},
                        stream_mode=["messages", "custom"],
                ):
                    if mode == "messages":
                        msg_chunk, _meta = chunk
                        content = getattr(msg_chunk, "content", "")
                        if content and isinstance(content, str):
                            full_response += content
                            if not first_token:
                                first_token = True
                            live.update(Markdown(full_response))

                    elif mode == "custom":
                        live.update(Spinner("dots", text="生产中...", style="cyan"))
                        _display_progress(chunk)

            if full_response:
                messages.append(AIMessage(content=full_response))
            print()

        except Exception as e:
            _console.print(f"错误: {e}", style="bold red")
            _console.print()

if __name__ == '__main__':
    asyncio.run(chat_loop())

效果是这样的：

进阶篇结尾：抽象的意义，就是把复杂藏起来

说实话，我之前一直有个疑问没想通。

短期记忆，也就是 Checkpointer，俗称“存档点”——这东西可以放到持久化 DB 里，按 thread_id 区分不同会话。这我能理解。

但长期记忆的 Store，是怎么区分用户的？

直到我翻完官方文档、又让文档 AI 帮我解读了一遍，才算彻底搞清楚：本质上是同一件事，只是传入方式不同。

记忆类型	标识方式	传入位置	生命周期
短期记忆（Checkpointer）	thread_id	config["configurable"]["thread_id"]	会话级
长期记忆（Store）	user_id	runtime.context["user_id"]（或 state）	跨会话

短期记忆通过 config 传入 thread_id，长期记忆通过 runtime.context 或 state 传入 user_id。本质上都是 “给我一个 ID，我帮你把数据隔离开”。

# 短期记忆：通过 config 传入 thread_id
agent.invoke(
    {"messages": [...]},
    config={"configurable": {"thread_id": "session_123"}}
)

# 长期记忆：通过 namespace 函数从 context 获取 user_id
def user_namespace(runtime):
    user_id = runtime.context.get("user_id", "anonymous")
    return ("memories", user_id)

backend = StoreBackend(
    store=store,
    namespace=user_namespace,
)

这里贴出一个官方的案例：

from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend
from deepagents.backends.filesystem import FilesystemBackend
from langgraph.store.memory import InMemoryStore

# 1. 定义你的 CompositeBackend 路由
backend = CompositeBackend(
    default=StateBackend(),                     # /tmp/ 或未匹配路径 → 内存 (StateBackend)
    routes={
        "/workspace/": FilesystemBackend(root_dir="./workspace", virtual_mode=True),  # /workspace/ → 本地磁盘
        "/skills/": FilesystemBackend(root_dir="./skills", virtual_mode=True),        # /skills/ → Skill 文件存储
        "/memories/": StoreBackend(namespace=lambda _rt: ("memories",)),             # /memories/ → 跨会话持久化
    }
)

# 2. 编写引导模型识别路径的系统 Prompt
system_prompt = """
You have access to a structured virtual filesystem with multiple storage backends mapped to specific paths:

- `/workspace/`: Local disk storage for your active project files, code, and working documents. Use this directory for file creation, editing, and executing shell or tool operations.
- `/skills/`: Read-only or managed storage containing reusable skills and custom agent instructions. Check here when looking for predefined capabilities.
- `/memories/`: Long-term persistent storage backed by a cross-thread store. Save important user preferences, key insights, and accumulated knowledge here (e.g., `/memories/user_preferences.txt`) so they persist across different sessions and threads.
- Default (`/` or other paths): Ephemeral memory scoped strictly to the current thread.

Always use the appropriate directory path according to whether data needs to be transient, local-disk backed, or long-term persistent.
"""

# 3. 创建 Agent
store = InMemoryStore()

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    backend=backend,
    store=store,
    system_prompt=system_prompt,
)

但是在使用小模型的时候，遇到一个奇怪的现象：

架构上，我们什么都准备好了——CompositeBackend 路由、FilesystemMiddleware、SkillsMiddleware、StoreBackend、InMemorySaver，甚至 TUI 界面都像模像样了。

但模型就是不用。

它不读 /skills/，不写 /memories/，甚至不调用工具，只输出“我来帮你读取”这种废话。直到换上 120B 的大模型，一切才恢复正常。

架构解决的是“能不能做”，模型解决的是“愿不愿做”。

┌─────────────────────────────────────────────────────────────┐
│  你配置的路径规则                                           │
│  routes={"/workspace/": ..., "/memories/": ...}            │
│                         ↓                                  │
│  FilesystemMiddleware 把它“翻译”成 System Prompt           │
│  "/workspace/ 是你的工作区，/memories/ 存你的偏好..."      │
│                         ↓                                  │
│  小模型：收到了，但我不太懂                                 │
│  大模型：明白了，我按这个规则干活                           │
└─────────────────────────────────────────────────────────────┘

小参数模型（比如 4B、7B 的量化版）天生的设计目标就是“纯文本对话”。

它们的训练数据里，90% 以上是网页文本、书籍、论坛对话。它们学的是“怎么说话”，而不是“怎么做事”。

模型类型	擅长什么	不擅长什么
小参数模型（4B-7B）	日常对话、简单问答、文本补全	多步推理、严格遵循 Prompt、工具调用
大参数模型（70B-120B）	复杂推理、遵循指令、工具编排	部署成本高、推理慢

所以，如果你的 Agent 需要做这些事：

• 根据 System Prompt 的规则行动
• 主动调用工具而不是描述动作
• 理解并遵循 Skill 机制
• 在多轮对话中保持一致的行为模式

那么小模型基本指望不上。 它不是“不愿意”，而是“没学过”。它的基座模型里没有“工具调用”这个概念，强行加 Function Calling 也像在自行车上装喷气发动机——看起来能跑，实际上不动。

如果你只是做纯对话场景（比如聊天机器人），小模型够用。但如果你要做 Coding Agent——需要读写文件、执行脚本、调用工具、加载 Skill——请直接用大参数模型，或者在本地用中等参数模型（如 30B 以上）的量化版本。

不要在小模型上死磕工具调用。那不是 Prompt 能解决的问题，那是模型能力的天花板。

我们搭好了架子，但真正决定 Agent 能不能用好的，还是模型本身。如果你的模型不听话，不要怀疑架构——换模型。

高级篇：当抽象变成障碍，就是该撕开它的时候了

基础篇我们手写了 Agent 的“心脏”——一个 while 循环 + 模型 + 工具，跑通了 ReAct。

进阶篇我们把 while 循环升级成了 LangGraph 状态机，用 CompositeBackend 做了路径路由，用 FilesystemMiddleware 和 SkillsMiddleware 注入了工具和规则，最后用 TUI 给它装上了一张“脸”。

——看起来一切都好，对吧？

但只有真正跑过的人才知道，这套东西有多少让人“无语”的时刻。

小参数模型根本不听使唤，你说“调用工具”它给你输出一长串“我来帮你调用”的废话；create_agent 和 create_deep_agent 都能用，但你不知道差别在哪；interrupt_before= 和 HumanInTheLoopMiddleware 都能做人工审批，但你又得纠结选哪个；store 参数和 StoreBackend 都可以实现长期记忆，但为什么有两种写法？

这就是 LangChain 的“真实面目”：它不是一个人设计出来的优雅框架，而是多个团队并行迭代、快速拼凑出来的“组装车间”。

深入 Checkpoint 和 Store——存档点和数据库

说实话，我每次看到 Checkpoint 这个词，脑子里浮现的就是游戏里的“存档点”。你玩到一半，存个档，下次从这儿接着打。

LangGraph 的 Checkpoint 本质上就是干这个的。

你每一次调用 agent.invoke() 或 agent.astream()，LangGraph 都会在每个节点执行完后自动保存一份“状态快照”。这个快照里包含：

• 当前所有的 messages（对话历史）
• 当前执行到哪个节点了
• 所有中间变量和状态字段

下次你带着同样的 thread_id 来，它就从上次保存的快照里恢复状态，接着往下走。

# 第一次调用：执行完毕后自动存档
agent.invoke(
    {"messages": [HumanMessage(content="你好")]},
    config={"configurable": {"thread_id": "user_123"}}
)

# 第二次调用：从存档恢复，接着往下走
agent.invoke(
    {"messages": [HumanMessage(content="我刚才说了什么？")]},
    config={"configurable": {"thread_id": "user_123"}}  # 同一个 thread_id
)

就是这么简单。它不是什么黑魔法——就是 “存状态 + 恢复状态”。

Checkpoint 存的是什么？ 是一个 StateSnapshot 结构：

StateSnapshot(
    values={
        'messages': [
            HumanMessage(content='你好'),
            AIMessage(content='你好！👋 有什么可以帮你的吗？'),
            HumanMessage(content='我叫123'),
            AIMessage(content='你好！很高兴认识你 😊'),
            HumanMessage(content='我叫什么？'),
            AIMessage(content='您好！我是一个人工智能助手...'),
        ]
    },
    next=(),
    config={'configurable': {'thread_id': 'user1', 'checkpoint_id': 'xxx'}},
    metadata={'step': 5, 'source': 'loop', ...},
    created_at='2024-...',
    parent_config={'configurable': {'thread_id': 'user1', 'checkpoint_id': 'yyy'}},
)

这就是一个完整的“存档点”，包含了整个对话的历史状态，注意这里的状态是既包含历史message又包含下一跳要做什么，所以说也可以叫断点，一个包含静态信息和动态行为的点。

至于Checkpoint 的存储层：

InMemorySaver 把存档存在内存里，程序重启就没了。

PostgresSaver / RedisSaver 把存档存在数据库里，重启还能恢复。

但不管存哪，它存的东西是一样的——都是“状态快照”。

# 内存存档
checkpointer = InMemorySaver()

# 数据库存档（生产环境）
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string("postgresql://...")

至于Checkpoint 和 Store 的区别？

组件	类比	存什么	谁使用
Checkpoint	游戏存档点	执行状态（包括 messages）	框架自动读写
Store	外部数据库	用户自定义数据（偏好、配置、知识）	Agent 通过工具主动读写

Checkpoint 是框架自动帮你存的，你不需要写任何代码，LangGraph 在每个节点执行完后自动保存。

Store 是Agent 主动去读写的，你需要通过 System Prompt 或工具教会模型：“当用户告诉你个人信息时，用 write_file 写到 /memories/ 下。”

一个简化的理解：

┌─────────────────────────────────────────────────────────────┐
│  Checkpoint（存档点）                                       │
│  ├── 谁在管：框架自动                                      │
│  ├── 存什么：执行状态（messages、节点位置、中间变量）      │
│  ├── 怎么用：每次调用带 thread_id 自动恢复                 │
│  └── 生命周期：按 thread_id 隔离，可持久化到 DB            │
├─────────────────────────────────────────────────────────────┤
│  Store（数据库）                                            │
│  ├── 谁在管：开发者 / Agent 主动                          │
│  ├── 存什么：用户偏好、项目配置、跨会话知识                │
│  ├── 怎么用：通过工具（read_file/write_file）读写          │
│  └── 生命周期：按 namespace 隔离，跨会话持久化             │
└─────────────────────────────────────────────────────────────┘

Checkpoint 的本质：图执行引擎的“存档/读档”

值得一提的是，Checkpoint 是 LangGraph 图执行引擎 的能力，不是 create_agent 的。

当你调用 graph.invoke() 或 graph.astream() 时，LangGraph 的执行引擎会在每个“超级步”（super-step）结束后自动调用 checkpointer.put() 保存当前状态快照。

下次你用同样的 thread_id 调用时，执行引擎会先调用 checkpointer.get() 恢复状态，然后从上次中断的地方继续执行。

create_agent 的函数签名里明确包含了这两个参数：

def create_agent(
    model: str | BaseChatModel,
    tools: Sequence[BaseTool | Callable[..., Any] | dict[str, Any]] | None = None,
    *,
    system_prompt: str | SystemMessage | None = None,
    middleware: Sequence[AgentMiddleware[...]] = (),
    checkpointer: Checkpointer | None = None,  # ← 存档点
    store: BaseStore | None = None,           # ← 数据库
    interrupt_before: list[str] | None = None,
    interrupt_after: list[str] | None = None,
    ...
) -> CompiledStateGraph: ...

create_agent 内部会构建一个 StateGraph（状态图），把所有节点、边、中间件都挂上去，最后调用：

graph = builder.compile(
    checkpointer=checkpointer,  # ← 你传的存档点
    store=store,               # ← 你传的数据库
)

这就是全部的秘密。create_agent 只是个“脚手架”，它帮你搭好图结构，然后原封不动地把 checkpointer 和 store 传给 LangGraph 的 compile() 方法。

LangGraph 官方文档也明确写着同样的用法：

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore

checkpointer = InMemorySaver()
store = InMemoryStore()

graph = builder.compile(checkpointer=checkpointer, store=store)

	Checkpoint	Store
谁在用	LangGraph 执行引擎	LangGraph 执行引擎
传给谁	graph.compile(checkpointer=...)	graph.compile(store=...)
存什么	图执行状态（messages、节点位置）	用户自定义键值数据
隔离方式	thread_id	命名空间（namespace）
谁读写	框架自动读写	节点函数通过 runtime.store 读写
生命周期	会话级（按 thread）	跨会话（按 namespace）

至于为什么**thread_id 被用作 Checkpoint 隔离标识，主要是因为它最初是为了映射“聊天对话线程（Conversation Thread）”这一最常见的 Agent 应用场景而设计的。**

在标准的 Chatbot 或多轮对话 Agent 中，用户的每一轮对话都属于同一个逻辑会话（Thread）。LangGraph 将底层的状态持久化和多轮交互抽象为 thread_id，使得同一个对话下的所有 Checkpoint（历史状态快照）都能被自动归类和串联起来。

然而，这种命名确实容易引起混淆，特别是在以下场景中：

1. 非对话场景： 当你运行的不是聊天机器人，而是批量数据处理、工作流自动化（Workflow Automation）或者后台任务时，把隔离 ID 叫 thread_id 就会让人觉得莫名其妙——这里根本没有“线程”或“聊天”的概念。
2. 多 ID 共存的混乱感： LangGraph 的配置中确实存在多个不同的 ID，它们各自承担不同的职责：
   • thread_id：会话/隔离空间 ID。用于隔离不同的运行实例或用户会话。同一个 thread_id 下可以有多条历史记录（即多个 Checkpoint）。
   • checkpoint_id：版本/历史快照 ID。用于标识某一个特定步骤执行完后的精确状态（支持时间旅行和状态回滚）。
   • run_id（或 task_id）：单次执行 ID。用于区分某一次具体的 invoke 或内部任务运行。

为什么不换个更直观的名字？

虽然从纯架构角度来看，thread_id 可以被看作是 session_id、namespace_id 或 partition_key，但 thread_id 这个名字在 SDK 和 LangGraph API 中已经作为核心约定俗成概念固定下来了。它对应了 LangGraph 平台中“Threads”的概念（即一组有状态的连续执行历史）。

如果你觉得 thread_id 容易产生歧义，完全可以把它当作你的业务会话 ID、用户 ID 或任务隔离 Key 来使用——只要保证每次需要读取或延续某一个独立状态时传入相同的字符串即可。

还有另外一个关键概念是 “超级步”（Super-step）

“超级步”这个概念，你可以把它想象成游戏里一轮一轮的“回合制”调度。它借鉴自 Google 的分布式图计算框架 Pregel，指的是图执行过程中的一个“批次轮次”，也就是一轮统一调度的执行阶段。

它的核心要点有两个：

• 并发执行：在同一个超级步内，所有可以被执行的节点（比如你图里的多个分支）会同时（并发地） 运行。你可以把它理解成一局游戏里的“回合”，所有能行动的单位都在这个回合里一起行动。
• 全局同步：每个超级步结束后，会有一个全局同步点，用来汇总和同步所有节点的执行结果。同步完成之后，才会进入下一个超级步。

LangGraph 的检查点（Checkpoint）只在每个超级步（super-step）结束时写入。

这里的“每个超级步的边界（boundary）”，指的就是一个超级步结束、下一个超级步开始前的那个同步点。它保证了状态的完整性和一致性。每完成一个“回合”，系统就会自动保存一次“存档”。

以 START -> A -> B -> END 这个顺序图为例：

1. 第一个超级步：执行输入节点 START。结束后，创建第一个检查点（存档1）。
2. 第二个超级步：执行节点 A。结束后，创建第二个检查点（存档2）。
3. 第三个超级步：执行节点 B。结束后，创建第三个检查点（存档3）。
4. 第四个超级步：执行结束节点 END。结束后，创建第四个检查点（存档4）。

可以看到，每一步执行完，系统都会自动存一个档。而如果图中存在可以并行执行的分支，那么它们会在同一个超级步内被执行，并在该超级步结束时，只创建一个检查点。

假设你的 Agent 需要同时做两件事：搜索天气和计算一个数学表达式。在 LangGraph 中，这两种操作如果互不依赖，就可以放在同一个超级步中并行执行。

用户请求: "搜索北京的天气，并计算 123 * 456"
         ↓
    ┌────┴────┐
    ↓         ↓
 天气工具    计算器工具
    ↓         ↓
    └────┬────┘
         ↓
  回答用户: 天气+计算结果

from langgraph.graph import StateGraph, START, END

class AgentState(TypedDict):
    messages: list[BaseMessage]
    weather: str | None
    calc_result: int | None

def call_weather(state: AgentState):
    # 模拟调用天气 API（耗时操作）
    return {"weather": "北京今日晴，25°C"}

def call_calculator(state: AgentState):
    # 模拟计算（耗时操作）
    return {"calc_result": 123 * 456}

graph = StateGraph(AgentState)
graph.add_node("weather", call_weather)
graph.add_node("calculator", call_calculator)
graph.add_edge(START, ["weather", "calculator"])  # 启动两条并行边
graph.add_edge("weather", END)
graph.add_edge("calculator", END)

执行过程（超级步维度）:

第一个超级步：

• START 节点执行完成
• 图发现两条并行边都准备就绪
• 同时启动 weather 和 calculator 两个节点
• 这两个节点并发执行（真正的并行，不是交替执行）
• 等待两个节点都执行完毕
• 超级步结束

关键点：虽然 weather 可能先执行完，calculator 慢一点，但它们仍然在同一个超级步内完成。系统不会在 weather 结束后就立刻进入下一个超级步，而是等待 calculator 也完成。

第一个超级步结束后：

checkpointer.put({
    "weather": "北京今日晴，25°C",
    "calc_result": 56088  # 123 * 456
})

只创建一个检查点，包含了两个并发节点的最终结果。

如果是串行执行（走两条边但不是并行），超级步会是这样：

超级步1: START → weather
超级步2: weather → calculator
超级步3: calculator → END

每个节点各自在自己的超级步中单独执行，每次都会创建一个检查点。这就是并行的价值——两个节点共享一个检查点，而串行则是各自的检查点。

并发模式:
┌─────────────────────────────────────────┐
│  超级步 1                                │
│  ┌──────────┐    ┌─────────────┐       │
│  │ 天气工具  │    │ 计算器工具  │       │
│  └──────────┘    └─────────────┘       │
│         ↓            ↓                  │
│         一次检查点（两个结果合并）        │
└─────────────────────────────────────────┘

串行模式:
┌──────────────────────────────────────────┐
│  超级步 1: START → 天气工具 → 检查点 1   │
│  超级步 2: 天气工具 → 计算器 → 检查点 2  │
│  超级步 3: 计算器 → END → 检查点 3       │
└──────────────────────────────────────────┘

还有一个与此相关的机制值得一提：如果在一个超级步中，部分节点成功执行，而另一个节点执行失败，LangGraph 会存储成功节点的待处理写入（pending writes）。

这样，当你从这个超级步恢复执行时，就不需要重新运行那些已经成功的节点了。这保证了高效的故障恢复能力。

同时这种设计也引出一个比较严重的问题！！

每个超级步结束，LangGraph 就写一次完整的 Checkpoint。 如果你的图里有 10 个节点，一次完整执行就会有 10 个存档点。

简单对话几次，数据库就有 1GB 数据——这完全不是人的幻觉或者程序的BUG，这是 LangGraph 默认行为的“副作用”。

因为每个 Checkpoint 保存的是整个状态快照（State Snapshot），不是“差异”。换句话说：

# 存档点 1
{
    "messages": [HumanMessage("你好")],
    "current_step": 1
}

# 存档点 2（完整保存，不是增量）
{
    "messages": [HumanMessage("你好"), AIMessage("你好！")],
    "current_step": 2
}

# 存档点 3
{
    "messages": [HumanMessage("你好"), AIMessage("你好！"), HumanMessage("我叫123")],
    "current_step": 3
}

每次都是完整拷贝整个 messages 列表，而不是只存“这次新增了哪个节点”。

如果你的 messages 里有：

• 一段 2000 字的代码
• 几十轮对话记录
• 工具调用的 JSON 结果

那么每次 Checkpoint 写入，这些数据都会被完整写一遍。你聊 10 轮，数据量就是 单次状态大小 × 10，而不是 单次状态大小 + 增量。

这就是为什么你简单对话几次，数据库就膨胀到 1GB。

从“实现简单”的角度看，合理——每次直接存完整状态，恢复时直接加载，不需要做任何 diff 或合并逻辑，LangGraph 执行引擎不需要处理“增量恢复”的复杂性。

但从“存储效率”的角度看，非常不合理——尤其是对 Coding Agent 来说，每次工具调用都可能返回大段代码或日志，存储开销是巨大的。

LangGraph 有解决吗？

有。PostgresSaver 和 RedisSaver 都支持异步写入和压缩存储，但架构层面没有从根本上解决“存全量”的问题。1GB 数据大概率来自：

1. 工具调用结果：每次 read_file 返回的代码被完整写进 Checkpoint
2. 长对话历史：几十轮对话堆叠在 messages 里，每次都被完整拷贝
3. 没有配置清理策略：Checkpoint 默认不会自动清理旧的存档点

对于大多数 Coding Agent 场景，你需要的只是最近的一个或几个存档点（比如用于恢复中断），而不是完整的历史存档链。

LangGraph 官方文档也提到了这一点：“在开发环境下，考虑限制 checkpoint 的数量来避免存储膨胀。”

可以做的优化：

1. 配置保留策略：使用支持 ttl 或 max_checkpoints 的 checkpointer，只保留最近 N 个存档点。
2. 避免超大消息体：如果你让 read_file 读取大型文件，Checkpoint 会完整保存这部分内容。考虑对文件内容做摘要或分块处理。
3. 使用异步压缩：PostgresSaver 支持 serde 序列化，可以配置压缩算法（如 zlib）来减少存储。

一个更真实的比喻：

Checkpoint 不是“增量存档”，而是“每走一步，都把整个游戏进度重新存一遍”。

如果你在游戏里存了 50 次档，每次存档都包含整个地图数据、所有任务状态、所有装备——那 1GB 真的不奇怪。

所以问题不是“为什么有 1GB”，而是“你真的需要存 50 个完整的存档吗”。

如果需要精细控制，可以使用 LangGraph 提供的 prune 方法。它允许你针对指定的 thread_ids 执行清理：

from langgraph.checkpoint.postgres import PostgresSaver

# 假设 checkpointer 是 PostgresSaver 实例
checkpointer.prune(
    thread_ids=["thread_id_1", "thread_id_2"],
    strategy="keep_latest"  # 或 "delete"
)

• strategy="keep_latest": 只保留每个线程的最新一个检查点。
• strategy="delete": 删除指定线程的所有检查点。

另外，你还可以使用 delete_thread 方法，删除某个线程的所有检查点和写入数据。

⚠️ 重要提醒：keep_latest 与人工干预的冲突

清理策略需要特别小心。如果你的 Agent 使用了人工干预（Human-in-the-loop），特别是多工具中断（multi-tool interrupt） 机制，简单粗暴地只保留最新检查点（keep_latest）可能会中断流程。

在这种场景下，LangGraph 需要依赖中间检查点来追踪在一个批次中哪些工具已经执行完毕。如果删除了这些检查点，恢复时可能会重复执行已经完成的任务。

因此，对于有人工干预的复杂场景，建议保留更多的最近检查点（例如 keep_last=20），或采用更精细的按时间清理策略。

Store 的本质：图执行引擎的“数据库”

Store 比较简单，他是 LangGraph 的另一个持久化能力。

Checkpoint 是按 thread_id 隔离的——thread-1 的存档和 thread-2 的存档互不干扰。Store 是跨线程的——它存的数据不绑定任何 thread_id，而是通过命名空间（namespace） 来组织。

当你在 compile() 时传了 store，LangGraph 执行引擎会把它注入到每个节点的 Runtime 对象中。节点函数可以通过 runtime.store 来读写数据：

def my_node(state, runtime):
    # 写入长期记忆
    runtime.store.put(("memories", "user_123"), "preferences", {"theme": "dark"})
    # 读取长期记忆
    prefs = runtime.store.get(("memories", "user_123"), "preferences")

这就是为什么你在 create_agent 里传了 store，但模型不会自动用它——Store 是给“节点函数”用的，不是给“模型”用的。模型要通过工具（比如 write_file / read_file）间接访问 Store，而这需要 StoreBackend + CompositeBackend 把 Store 包装成文件系统路径。

通过阅读源码，我们发现langgraph的Node是一个类型集合，参数带runtime的是一种Node，带config的是一种Node，都带的也算，不带的也算，哪怕是一个实现了__call__的类也算Node：

StateNode: TypeAlias = (
    _Node[NodeInputT]
    | _NodeWithConfig[NodeInputT]
    | _NodeWithWriter[NodeInputT]
    | _NodeWithStore[NodeInputT]
    | _NodeWithWriterStore[NodeInputT]
    | _NodeWithConfigWriter[NodeInputT]
    | _NodeWithConfigStore[NodeInputT]
    | _NodeWithConfigWriterStore[NodeInputT]
    | _NodeWithRuntime[NodeInputT, ContextT]
    | Runnable[NodeInputT, Any]
)

所以我们通过runtime.操作可以点出很多属性，其中一个属性就是store！ 然后就可以自由的CRUD，但在langchain框架和deepagents中，这个CRUD也不用你操作了。

当然你通过中间件的runtime去获取node依然可以操作，但StoreBackend已经做了这件事情！

class StoreBackend(BackendProtocol):
    """Backend that stores files in LangGraph's BaseStore (persistent).

    Uses LangGraph's Store for persistent, cross-conversation storage.
    Files are organized via namespaces and persist across all threads.

    The namespace can include an optional assistant_id for multi-agent isolation.
    """

    def __init__(
        self,
        runtime: object = None,
        *,
        store: BaseStore | None = None,
        namespace: NamespaceFactory | None = None,
        file_format: FileFormat = "v2",
    ) -> None:
        r"""Initialize `StoreBackend`.

        Args:
            runtime: Deprecated - accepted for backward compatibility but
                ignored. Store and context are now obtained via
                `get_store()` / `get_runtime()`.
            store: Optional `BaseStore` instance. When provided, this store
                is used directly. When `None` (the default), the store is
                obtained at call time via `get_store()`, which requires
                a LangGraph graph execution context.
            namespace: Optional callable that receives a `Runtime` and returns
                a namespace tuple for scoping store operations.
                Wildcards (`*`) are forbidden.
                If `None`, uses legacy assistant_id detection from metadata (deprecated).

                Old-style callables that accept `BackendContext` still work
                but are deprecated and will be removed in `deepagents==0.7.0`.

            file_format: Storage format version. `"v1"` (default) stores
                content as `list[str]` (lines split on `\\n`) without an
                `encoding` field. `"v2"` stores content as a plain `str`
                with an `encoding` field.

        Example:
            `namespace=lambda rt: (rt.server_info.user.identity, "filesystem")`
        """

StoreBackend实现了Backend协议，说白了就是Store已经变成了一个抽象文件系统了。

上下文管理：Trim、Delete 与摘要

既然 Checkpoint 保存了完整的 messages 列表，随着对话轮次增加，messages 会越来越长，针对这个问题，LangChain 提供了几种上下文管理策略：

注意：这里只是针对状态机中的messages字段做操作，而这个操作也会被记录到快照中，因此上下文管理不会导致快照变小。

1. Trim（裁剪）

   简单粗暴：保留最近 N 条消息，删掉更早的。

   @before_model
   def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
       """Keep only the last few messages to fit context window."""
       messages = state["messages"]
   
       if len(messages) <= 3:
           return None  # No changes needed
   
       first_msg = messages[0]
       recent_messages = messages[-3:] if len(messages) % 2 == 0 else messages[-4:]
       new_messages = [first_msg] + recent_messages
   
       return {
           "messages": [
               RemoveMessage(id=REMOVE_ALL_MESSAGES),
               *new_messages
           ]
       }
   

2. Delete（删除）

   更粗暴：完全清空上下文，开始新对话。

   from langgraph.graph.message import REMOVE_ALL_MESSAGES  
   
   def delete_messages(state):
       return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)]}
   

   这通常在用户主动说“重新开始”时才使用。

3. 摘要（Summarization）

   这才是真正的“智能压缩”。不删消息，而是让模型把历史对话压缩成一段摘要，用摘要代替原始对话。

   summary = llm.invoke("请总结以下对话：{history}")
   messages = [SystemMessage(content=f"对话历史摘要：{summary}"), current_user_message]
   

   这样既保留了关键信息（用户说了什么、做了什么决策），又大幅减少了上下文长度。

   摘要的本质是用“知识密度”换取“Token 数量”。

有人可能好奇，为什么用：RemoveMessage，这是因为langchain的状态机AgentState采用字段+处理器的方式去更新字段，

messages: Required[Annotated[list[AnyMessage], add_messages]]

如果你不返回一个RemoveMessage，而是返回一个你自己清洗后的list，那么这个list会被完全追加到历史messages。

其次是为什么现代上下文管理不再简单粗暴地删除？

因为 Agent 需要记忆的连贯性。

如果直接删除旧消息，Agent 会“失忆”——用户之前说过的需求、做过的决策、提供的信息，全部丢失。这在 Coding Agent 场景中尤其致命：

用户：帮我写一个 FastAPI 项目，用 PostgreSQL
Agent：好的，我创建了项目结构...
（30 轮对话后）
用户：帮我加一个用户认证
Agent：PostgreSQL 是什么？你没告诉过我啊。（因为历史被删了）

所以现代上下文管理的策略是：

1. 保留关键信息：用摘要压缩历史，而不是删除
2. 分级存储：热数据（最近几轮）保持完整，冷数据（早期对话）存摘要
3. 按需检索：用 RAG 从外部检索相关历史

LangChain 内置了 SummarizationMiddleware，它的工作流程是：

1. 每次调用模型前，检查当前 messages 的 token 数
2. 如果超过阈值，调用模型生成历史摘要
3. 用摘要替换旧的对话历史（或作为 SystemMessage 追加）

agent = create_agent(
    model="gpt-5.5",
    tools=[...],
    middleware=[
        SummarizationMiddleware(
            model="gpt-5.4-mini",
            trigger=("tokens", 4000),
            keep=("messages", 20)
        )
    ],
    checkpointer=checkpointer,
)

但注意：摘要本身也是一次模型调用，会产生额外的 token 消耗和时间开销。这就是“用钱换空间”的策略。

然后我们来看一下SummarizationMiddleware内部是如何做的：

@override
    async def abefore_model(
        self, state: AgentState[Any], runtime: Runtime[ContextT]
    ) -> dict[str, Any] | None:
        """Process messages before model invocation, potentially triggering summarization.

        Args:
            state: The agent state.
            runtime: The runtime environment.

        Returns:
            An updated state with summarized messages if summarization was performed.
        """
        messages = state["messages"]
        self._ensure_message_ids(messages)

        total_tokens = self.token_counter(messages)
        if not self._should_summarize(messages, total_tokens):
            return None

        cutoff_index = self._determine_cutoff_index(messages)

        if cutoff_index <= 0:
            return None

        messages_to_summarize, preserved_messages = self._partition_messages(messages, cutoff_index)

        summary = await self._acreate_summary(messages_to_summarize)
        new_messages = self._build_new_messages(summary)

        return {
            "messages": [
                RemoveMessage(id=REMOVE_ALL_MESSAGES),
                *new_messages,
                *preserved_messages,
            ]
        }

这段代码是 SummarizationMiddleware 的核心——它在模型推理之前，检查消息列表的长度，如果超过阈值就自动触发摘要压缩。

1. 获取消息列表并确保 ID 完整

   messages = state["messages"]
   self._ensure_message_ids(messages)
   

   从状态中取出所有消息，并为每条消息生成唯一 ID。LangGraph 的消息系统依赖 ID 来追踪消息的添加和删除。RemoveMessage 需要通过 ID 来定位要删除的消息。

2. 计算 Token 数并判断是否需要摘要

   total_tokens = self.token_counter(messages)
   if not self._should_summarize(messages, total_tokens):
       return None
   

   用 token_counter 统计所有消息的 token 总数。如果没超过阈值（如 16000 token），直接返回 None，表示“不需要做任何事”。

3. 确定从哪里开始切割

   cutoff_index = self._determine_cutoff_index(messages)
   if cutoff_index <= 0:
       return None
   

   _determine_cutoff_index 决定从哪条消息开始保留。通常的策略是：

   • 从最新的消息往前数，直到 token 数达到阈值
   • 之前的消息都交给摘要，之后的消息保留

   如果 cutoff_index <= 0，说明消息太短或无法切割，直接返回。

4. 分割消息

   messages_to_summarize, preserved_messages = self._partition_messages(messages, cutoff_index)
   

   将消息分成两组：

   • messages_to_summarize：要压缩成摘要的旧消息
   • preserved_messages：保留的最近消息（直接保留，不摘要）

5. 调用 LLM 生成摘要

   summary = await self._acreate_summary(messages_to_summarize)
   

   异步调用模型，把 messages_to_summarize 压缩成一段摘要文本。

6. 构建新的消息列表并返回更新

   new_messages = self._build_new_messages(summary)
   return {
       "messages": [
           RemoveMessage(id=REMOVE_ALL_MESSAGES),
           *new_messages,
           *preserved_messages,
       ]
   }
   

   这是最关键的代码。它的逻辑是：

   • RemoveMessage(id=REMOVE_ALL_MESSAGES)：删除所有现有的消息
   • *new_messages：插入新的摘要消息（通常是一个 SystemMessage，内容是摘要文本）
   • *preserved_messages：插入最近保留的消息

   最终效果是：

   [原始消息列表] → [摘要 SystemMessage] + [最近 N 条消息]
   

为什么用 RemoveMessage 而不是直接赋值？

在 LangGraph 的状态管理中，messages 字段是追加模式（append-only）。你直接赋值会被忽略，必须通过 RemoveMessage 显式删除旧消息，再添加新消息。

REMOVE_ALL_MESSAGES 是一个特殊 ID，表示“删除所有消息”。

理解了 SummarizationMiddleware 的原理后，你会发现一个重要的模式：

中间件可以在模型调用前/后，对状态（state）进行任意操作。

SummarizationMiddleware 做的事情是：在模型调用前，检测上下文长度，如果超限，就修改 state["messages"]（用摘要替换原始消息）。

同理，你可以写一个自定义中间件来操作 Store——比如在对话结束后，自动把关键信息写入 Store，实现跨会话记忆：

from langchain.agents.middleware import AgentMiddleware

class MemoryMiddleware(AgentMiddleware):
    """自动将用户偏好写入 Store"""

    def before_model(self, state, runtime):
        # 检测是否包含“我喜欢”“我习惯”等信号词
        last_msg = state["messages"][-1].content if state["messages"] else ""
        
        if "我喜欢" in last_msg or "我习惯" in last_msg:
            # 写入 Store
            runtime.store.put(
                ("memories", "user_preferences"),
                "preference",
                {"value": last_msg}
            )
        
        return state

然后把它加入 middleware 列表：

agent = create_agent(
    model=model,
    middleware=[
        fs_middleware,
        skills_middleware,
        SummarizationMiddleware(model=model, max_tokens_before_summary=16000),
        MemoryMiddleware(),  # ← 自定义中间件
    ],
    checkpointer=checkpointer,
    store=store,  # ← 需要把 Store 传进去，让 Runtime 能访问
)

这就是从“上下文管理”到“自定义行为扩展”的顺理成章的演进：

Trim（裁剪，丢失信息）
    ↓
Summary（摘要压缩，保留信息）
    ↓
SummarizationMiddleware（官方实现，自动触发）
    ↓
自定义中间件（在摘要基础上，把信息写入 Store）
    ↓
跨会话记忆（Store 保存摘要，下次会话加载）

集成生态：LangChain 的“武器库”

在深入 Checkpoint 和 Store 的本质后，我们终于可以把它们组装起来，为我们的 Agent 配备完整的“记忆系统”。同时，理解 LangChain 庞大的集成生态（Ecosystem），是让这套记忆系统从“玩具”走向“生产”的关键。

LangChain 拥有一个庞大的集成生态，也就是 langchain-<provider> 这样的集成包。这个生态的核心价值在于：它通过一套标准接口，屏蔽了不同底层服务商的差异。这意味着，你可以先用 InMemorySaver 在本地测试，然后无缝切换到生产级的 PostgresSaver，而无需修改业务逻辑代码。

在开发阶段，InMemorySaver 非常方便，但它的数据存储在 RAM 中，进程重启即丢失。要让短期记忆持久化，我们需要一个持久化的 Checkpointer。

LangChain 为生产环境提供了多种选择，例如 SqliteSaver 就是一种基于本地文件的轻量级方案。

通过 LangChain 官方的 AI 助手（chat.langchain.com），我们可以快速获得这类问题的具体实现。例如，询问“如何替换为使用文件存储短期记忆？”，官方助手给出了如下方案：

1. 安装必要的包：

   pip install langgraph-checkpoint-sqlite
   

2. 使用同步 SqliteSaver：

   import sqlite3
   from langgraph.checkpoint.sqlite import SqliteSaver
   
   # 创建或连接到本地 SQLite 文件
   conn = sqlite3.connect("checkpoints.db", check_same_thread=False)
   checkpointer = SqliteSaver(conn)
   
   # 编译图时传入 checkpointer
   graph = workflow.compile(checkpointer=checkpointer)
   

3. 或使用异步 AsyncSqliteSaver：

   import aiosqlite
   from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
   
   async with aiosqlite.connect("checkpoints.db") as conn:
       checkpointer = AsyncSqliteSaver(conn)
       graph = workflow.compile(checkpointer=checkpointer)
   

就这样，通过替换一个 checkpointer 实现，我们就将 Agent 的短期记忆从易失的内存，迁移到了持久的本地文件。

将这个理念应用到我们的 Agent 中，代码结构会非常清晰：

import aiosqlite
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver

async def chat_loop() -> None:
    async with aiosqlite.connect(AGENT_ROOT / "memory" / "short_memory.db") as conn:
        sqlite_memory = AsyncSqliteSaver(conn)

        # 确保表结构已创建（建议调用一次 setup）
        # 注意：setup 是异步的，且只需要调用一次（如果表不存在则创建）
        await sqlite_memory.setup()

        agent = create_agent(
            model=ChatOpenAI(
                model="glm-4.7-flash",
                base_url="https://open.bigmodel.cn/api/paas/v4",
                api_key=SecretStr("x"),
                temperature=0.7,
            ),
            middleware=[
                fs_middleware,
                skills_middleware
            ],
            system_prompt=system_prompt,
            checkpointer=sqlite_memory,
        )

        """交互式对话主循环，流式输出 LLM 回复和生产进度。"""
        _display_banner()

就这样，我们的 Agent 同时具备了会话内的短期记忆和跨会话的长期记忆能力，并且短期记忆已经可以持久化到文件。

然后当然，我们的目标是一个coding agent，那么我们定义如下三个变量：

from pathlib import Path
from uuid import uuid4

# ① Agent 安装根目录（固定，不受用户执行位置影响）
AGENT_ROOT = Path(__file__).resolve().parent

# ② 用户项目根目录（即用户执行命令时所在的目录）
USER_PROJECT = Path.cwd()   # 或者 Path()，效果相同

# 当前用户
USER_ID:str = str(uuid4())

同时，使用backend router划分区域，聪明的你肯定想到了，USER_PROJECT是可以作为文件系统的根目录的，也就是当用户说，帮我生成一个文件时，就会把文件写到这个目录下：

我们定义了路径是不够的，因为模型不知道，所以需要配套修改prompt：

system_prompt = f"""你是 langcode，一个可以直接读写用户项目文件的编程助手。

## 你的工作目录

用户的当前项目目录是 `/workspace/`。你在这个目录下拥有完整的文件读写权限：
- 可以查看项目结构（`ls`）
- 可以读取任何文本文件（`read_file`）
- 可以创建、修改、删除文件（`write_file`）
- 可以搜索代码内容（`grep`）
- 可以执行 Shell 命令（`execute`）

**目标**：帮助用户完成实际的编程任务，而不是空谈。

## 核心工作流

收到用户请求
↓
理解需求（这是什么类型的任务？）
↓
【必须】先用 ls 查看当前项目结构，了解已有文件
↓
根据需求读取相关文件（read_file）
↓
执行操作：
- 写新代码 → write_file
- 修改现有代码 → write_file（覆盖）
- 运行测试/脚本 → execute
- 搜索内容 → grep
↓
输出：简洁说明你做了什么，结果如何

## 工具使用指南

### 第一步永远先 ls
每次用户提出新需求，**先 `ls("/workspace/")`**，了解项目结构后再行动。不要凭空猜测。

### 文件操作要主动
- 生成任何代码 → 直接 `write_file` 保存到 `/workspace/` 下的合适位置
- 不要只输出代码块让用户手动复制保存
- 文件路径使用正斜杠 `/`，如 `/workspace/main.py`

### 执行命令要谨慎
- 可以执行 Shell 命令来运行脚本、安装依赖、启动服务
- 危险命令（rm -rf、shutdown 等）自动拒绝
- 长时间运行的命令（如启动 Web 服务）建议不要执行，或提醒用户手动运行

## 输出风格

- **简洁直接**：不要重复用户的问题，不要客套话
- **先说做了什么，再说结果**：“已创建 main.py，包含 FastAPI 示例代码”
- **代码块只用于展示**：如果你确实需要向用户展示代码片段，用 Markdown 代码块，但默认应该直接写文件

## 安全边界

- 禁止读取 `.env`、`/etc/passwd`、`~/.ssh/` 等敏感文件
- 禁止执行 `rm -rf`、`del /f` 等危险命令
- 不要编造不存在的库、API 或技术方案
- 遇到不确定的路径，先用 `ls` 确认

## 示例对话

用户："帮我写一个 Go 的爬虫 demo"
→ ls("/workspace/") → 看到项目结构
→ write_file("/workspace/main.go", content="...") → 创建 Go 爬虫文件
→ 输出："已创建 /workspace/main.go，包含一个基础的 HTTP 爬虫示例。执行 `go run main.go` 即可运行。"

用户："这个项目用了什么框架？"
→ ls("/workspace/") → 查看文件列表
→ 看到 go.mod、main.go 等 → 基于文件结构判断
→ 输出："看起来是一个 Go 项目，使用了标准库 net/http，没有使用第三方框架。"

用户："帮我给 main.go 加一个 User-Agent 头"
→ read_file("/workspace/main.go") → 读取当前代码
→ 定位到 HTTP 请求部分 → 修改代码
→ write_file("/workspace/main.go", content="...") → 覆盖保存
→ 输出："已在 main.go 的 HTTP 请求中添加了 User-Agent 头。"

现在开始为用户服务。每次操作前确认路径正确，执行后说明结果。"""

当我们让其写一个文件的时候：

当我们让其写一个小demo项目的时候：

由于我们是在当前目录的控制台执行的agent，因此所谓的用户项目目录其实就是我们的项目，所以agent可能会修改我们本身的源码。。。。QAQ

如果你的agent在文件操作上“装唐”，那么正如前文所言，换一个模型，这里推荐glm的免费模型glm-4.7-flash，好歹是一个30B级别的模型，比ollama本地的强多了：

agent = create_agent(
        model=ChatOpenAI(
            model="glm-4.7-flash",
            base_url="https://open.bigmodel.cn/api/paas/v4",
            api_key=SecretStr(""),
            temperature=0.7,
        ),
        middleware=[
            fs_middleware,
            skills_middleware
        ],
        system_prompt=system_prompt,
        checkpointer=sqlite_memory,
    )

还有一点需要注意的是，异步的checkpoint是需要asyncio的事件循环的，因此你需要将其放入整个agent loop中。

权限控制——工具级审批 vs 路径级权限

Agent 能读写文件、执行命令之后，下一个必须面对的问题是：如何防止它乱来？

LangChain 提供了两个不同层级的控制机制，分别对应两种不同的“乱来”：

控制维度	对应机制	控制对象
工具级审批	interrupt_on（HumanInTheLoopMiddleware 的入参）	哪些工具需要人工审批
路径级权限	_permissions（FilesystemMiddleware 的入参）	哪些路径允许什么操作

它们互不替代，可以叠加使用。一个是“这把刀能不能用”，一个是“这把刀能砍哪块砧板”。

interrupt_on 是 HumanInTheLoopMiddleware 的入参，当你在 create_agent 或 create_deep_agent 中传入它时，框架会自动注入 HumanInTheLoopMiddleware。

它的核心逻辑是：按工具名称控制是否需要人工审批。

agent = create_agent(
    model="gpt-5.5",
    tools=[write_file, execute_sql, read_data],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "write_file": True,  # All decisions (approve, edit, reject, respond) allowed
                "execute_sql": {"allowed_decisions": ["approve", "reject"]},  # No editing allowed
                "read_data": False, # Safe operation, no approval needed
            },
            # Prefix for interrupt messages - combined with tool name and args to form the full message
            # e.g., "Tool execution pending approval: execute_sql with query='DELETE FROM...'"
            # Individual tools can override this by specifying a "description" in their interrupt config
            description_prefix="Tool execution pending approval",
        ),
    ],
    checkpointer=InMemorySaver(),
)

当 Agent 尝试调用 delete_file 时，执行会暂停，等待人类 reviewer 审批。reviewer 可以做四种操作：

决策	含义
approve	使用 Agent 提议的原始参数执行工具
edit	修改工具参数后再执行
reject	拒绝执行，跳过此工具调用
respond	直接返回人类的消息作为工具结果

适用场景：你担心 Agent 会误删文件、误执行危险命令，或者你想在关键操作前加一道人工审核。

限制：它只能按工具名称控制，粒度较粗。write_file 要么全部需要审批，要么全部不需要——你无法区分“写 /workspace/”和“写 /secrets/”。

_permissions 是 FilesystemMiddleware 的入参，用于控制 不同路径下的读写权限。它的粒度比 interrupt_on 更细，支持 Glob 通配符匹配。

它的核心逻辑是：按路径控制操作类型（read / write）的权限。

from deepagents import FilesystemPermission, create_deep_agent

# Read-only agent: deny all writes
agent = create_deep_agent(
    model=model,
    backend=backend,
    permissions=[
        FilesystemPermission(
            operations=["write"],
            paths=["/**"],
            mode="deny",
        ),
    ],
)

或

fs_middleware = FilesystemMiddleware(
    backend=backend,
    _permissions=FilesystemPermission(
    operations=["write"],
    paths=["/**"],
    mode="deny",
))

FilesystemPermission 的三个核心字段：

字段	说明
operations	["read"] 或 ["write"]。read 覆盖 ls、read_file、glob、grep；write 覆盖 write_file、edit_file、delete
paths	Glob 模式列表，如 ["/workspace/**"]，支持 ** 递归匹配
mode	"allow"（允许）、"deny"（拒绝并返回错误）、"interrupt"（暂停等待人工审批）

适用场景：你希望 Agent 对某些敏感目录（如 /secrets/、/etc/）完全没有访问权限，或者对工作区外的路径执行更严格的审批。

限制：它只能控制文件系统路径，无法控制非文件系统相关的工具（如网络请求、API 调用）。

两者如何协同工作？

用户请求："删除 /secrets/credentials.json"
         ↓
    触发 delete_file 工具调用
         ↓
    【工具级审批】interrupt_on 检查
    └── delete_file 是否需要审批？
        ├── 是 → 暂停 → 人类审批
        └── 否 → 继续
         ↓
    【路径级权限】_permissions 检查
    └── /secrets/credentials.json 是否匹配规则？
        ├── 匹配 "deny" → 返回权限错误，不执行
        ├── 匹配 "interrupt" → 暂停 → 人类审批
        └── 匹配 "allow" 或无匹配 → 自动执行
         ↓
    执行工具（仅当两级检查都通过）

两者是 AND 关系：Agent 必须同时通过工具级审批和路径级权限检查，工具才能被执行。

场景	interrupt_on	_permissions	结果
写 /workspace/main.py	未配置 write_file	allow（或无匹配）	✅ 自动执行
写 /workspace/main.py	write_file: True	allow（或无匹配）	⏸️ 需要审批
写 /secrets/password.txt	未配置 write_file	deny	❌ 拒绝执行
写 /secrets/password.txt	write_file: True	interrupt	⏸️ 需要审批（但最终可能被拒）

实际场景中，你很可能同时使用两种控制。

一个重要的提醒

所有中断机制都依赖 checkpointer。没有 checkpointer，Agent 无法在中断后保存状态，也就无法恢复执行。无论是 interrupt_on 还是 _permissions 的 mode="interrupt"，都必须配合 checkpointer 使用。

# ✅ 正确：配置了 checkpointer
agent = create_deep_agent(
    model="...",
    interrupt_on={"delete_file": True},
    checkpointer=MemorySaver(),  # 必须！
)

# ❌ 错误：没有 checkpointer，中断无法恢复
agent = create_deep_agent(
    model="...",
    interrupt_on={"delete_file": True},
    # 没有 checkpointer → 中断后状态丢失
)

选择指南

你的需求是什么？
    │
    ├─ 想控制“哪些工具”需要审批？
    │   └─ 用 interrupt_on（工具级审批）
    │
    ├─ 想控制“哪些路径”能做什么？
    │   └─ 用 _permissions（路径级权限）
    │
    └─ 两个都想控制？
        └─ 两个都用，互不冲突

工具级审批和路径级权限是互补的，而不是互斥的。理解了它们的区别，你就能根据具体场景灵活组合使用。

然后我们来改造下自己的程序：

import asyncio
import aiosqlite

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from pydantic import SecretStr
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
from deepagents.middleware import SkillsMiddleware,FilesystemMiddleware
from langchain.agents.middleware import SummarizationMiddleware,HumanInTheLoopMiddleware

from settings import AGENT_ROOT,USER_ID
from tui.ui import tui
from router.routers import backend
from prompt_builder.system_prompt import system_prompt

model = ChatOpenAI(
    model="glm-4.7-flash",
    base_url="https://open.bigmodel.cn/api/paas/v4",
    api_key=SecretStr(""),
    temperature=0.7,
    max_retries=3,
    timeout=5
)

skills_middleware = SkillsMiddleware(
    # 这里的sources是为了SkillsMiddleware初始化的时候用/skills/路由匹配到backend进行查阅skills
    sources=["/skills/"],
    backend=backend,
)
fs_middleware = FilesystemMiddleware(backend=backend)

summary_middleware = SummarizationMiddleware(
    trigger=("tokens", 200000),
    model=model
)

human_in_the_loop_middleware = HumanInTheLoopMiddleware(
    interrupt_on={
        # 默认为approve，edit，reject，respond
        "write_file": True,
        "execute_command": {"allowed_decisions": ["approve", "reject"]},
    },
    description_prefix="需要人工审批",
)


async def loop() -> None:
    async with aiosqlite.connect(AGENT_ROOT / "memory" / "short_memory.db") as conn:
        sqlite_memory = AsyncSqliteSaver(conn)

        # 确保表结构已创建（建议调用一次 setup）
        # 注意：setup 是异步的，且只需要调用一次（如果表不存在则创建）
        await sqlite_memory.setup()

        agent = create_agent(
            model=model,
            middleware=[
                # 1. 文件系统中间件（最先注册工具，供后续中间件使用）
                fs_middleware,
                # 2. 上下文与知识注入（信息输入）
                skills_middleware,
                # 3. 上下文管理（信息压缩）
                summary_middleware,
                # 4. 行为控制与拦截（行动输出）
                human_in_the_loop_middleware,
            ],
            system_prompt=system_prompt,
            checkpointer=sqlite_memory,
        )

        await tui(agent,USER_ID)




if __name__ == '__main__':
    asyncio.run(loop())

这里我们虽然加入了HumanInTheLoopMiddleware中间件，但我们的tui还是不支持中断的，因此tui程序也需要修改：

from langchain.messages import HumanMessage
from langgraph.graph.state import CompiledStateGraph
from langgraph.types import Command
from rich.console import Console
from rich.live import Live
from rich.markdown import Markdown
from rich.panel import Panel
from rich.spinner import Spinner
from rich.text import Text

from langchain_core.runnables import RunnableConfig

_console = Console()

def _display_banner() -> None:
    """展示启动横幅。"""
    title = Text("Attack Chain Data Production Engine", style="bold bright_white")
    subtitle = Text.assemble(
        ("  /list ", "bold cyan"), ("查看场景  ", "dim"),
        ("  /quit ", "bold cyan"), ("退出", "dim"),
    )
    _console.print(Panel(
        Text.assemble(title, "\n", subtitle),
        border_style="bright_blue",
        padding=(1, 2),
    ))
    _console.print()


def _display_progress(event: dict) -> None:
    """在终端展示生产进度事件。"""
    ts = event.get("timestamp", "")
    content = event.get("content", "")
    event_type = event.get("type", "progress")

    style_map = {"progress": "cyan", "result": "green", "error": "red"}
    style = style_map.get(event_type, "cyan")
    _console.print(f"  [{ts}] {content}", style=style)


async def tui(agent: CompiledStateGraph, user_id: str) -> None:
    """交互式对话主循环，使用 astream v2 格式与 HumanInTheLoopMiddleware 完美适配。"""
    _display_banner()

    config: RunnableConfig = RunnableConfig(configurable={"thread_id": user_id})

    while True:
        try:
            # 1. 检查当前图是否处于中断状态
            state = await agent.aget_state(config)
            if state.tasks and any(task.interrupts for task in state.tasks):
                _console.print("\n[bold yellow][!] 智能体请求人工审批 (Human-in-the-Loop):[/]")

                # 解析 HumanInTheLoopMiddleware 抛出的中断内容
                for task in state.tasks:
                    for intr in task.interrupts:
                        payload = intr.value
                        _console.print(f"审批详情: {payload}")

                # 根据决策提示用户
                choice = _console.input("[bold yellow]请输入决策 (approve / reject / edit): [/]").strip().lower()

                if choice in ("approve", "yes", "y"):
                    resume_payload = {"decisions": [{"type": "approve"}]}
                elif choice in ("reject", "no", "n"):
                    resume_payload = {"decisions": [{"type": "reject"}]}
                else:
                    resume_payload = {"decisions": [{"type": "approve"}]}

                run_input = Command(resume=resume_payload)
            else:
                # 正常对话输入
                user_input = _console.input("[bold green]> [/]").strip()
                if not user_input:
                    continue
                if user_input.lower() in ("/quit", "/exit", "quit", "exit"):
                    _console.print("再见!")
                    break
                run_input = {"messages": [HumanMessage(content=user_input)]}

            _console.print()
            try:
                full_response = ""
                first_token = False

                with Live(
                        Spinner("dots", text="思考中...", style="cyan"),
                        console=_console,
                        refresh_per_second=8,
                ) as live:
                    # 传入 version="v2" 开启统一流式格式
                    async for chunk in agent.astream(
                            input=run_input,
                            config=config,
                            stream_mode=["messages", "custom"],
                            version="v2",
                    ):
                        chunk_type = chunk.get("type")
                        data = chunk.get("data")

                        # v2 格式下通过 chunk["type"] 进行精确分发
                        if chunk_type == "messages":
                            # data 是一个 (msg_chunk, metadata) 元组
                            msg_chunk, _meta = data
                            content = getattr(msg_chunk, "content", "")
                            if content and isinstance(content, str):
                                full_response += content
                                if not first_token:
                                    first_token = True
                                live.update(Markdown(full_response))

                        elif chunk_type == "custom":
                            live.update(Spinner("dots", text="生产中...", style="cyan"))
                            _display_progress(data)

                _console.print()

            except Exception as e:
                _console.print(f"错误: {e}", style="bold red")
                _console.print()
        except (EOFError, KeyboardInterrupt):
            _console.print("\n再见!")
            break

当我们使用astream的时候，中断就需要自行获取state = await agent.aget_state(config)，这里有两点需要讲清楚，第一点是中间件的顺序，第二点是langchain调用方式。

“基础的，兜底的在前面，增加上下文的在中间，后置处理的在后面。”

这就是中间件顺序的核心原则。它和面包、蔬菜、酱料的顺序逻辑如出一辙：

顺序	层定位	包含的中间件	职责
① 最前	基础设施（基础，兜底）	FilesystemMiddleware TodoListMiddleware	提供底层能力：工具、状态管理、权限控制。必须最先就绪，后续中间件才能依赖它们工作。
② 中间	上下文准备（增加上下文）	SkillsMiddleware MemoryMiddleware	丰富信息源：注入技能、记忆、指令到 System Prompt。在模型“思考”前准备好一切所需信息。
③ 最后	行为控制（后置处理）	HumanInTheLoopMiddleware SummarizationMiddleware	拦截和压缩：在模型决策后、工具执行前进行审批，或在模型调用前压缩上下文。“后置”指的是紧邻模型调用/工具执行的那个位置。

可以灵活调整，但要遵循逻辑依赖关系：

1. 不能颠倒依赖关系：如果一个中间件依赖另一个提供的工具或状态，依赖者必须放在被依赖者后面。
2. 稳定的放前面：不变的内容（如工具定义）适合放前面，有利于 Prompt Caching，节省成本。
3. 易变的放后面：变化频繁的上下文或动态指令放后面，不影响缓存。
4. 拦截放最后：HumanInTheLoopMiddleware 一定要放在工具执行前的最末尾，确保在 Agent 真正行动前拦截。

这个原则不仅能帮你理解现有的中间件配置，也能指导你未来编写自定义中间件时的顺序决策。简单、清晰、实用。

其次是langchain调用方法，我们可以把这些方法分成三组来看：一次性调用 (invoke)、状态流式 (stream) 和 事件流式 (stream_events)。每个组里的 v1、v2、v3 代表了演进的过程。

第一组：一次性调用 —— invoke / ainvoke

• v1（默认传统版）
  • 行为：执行完成后，直接返回一个包含所有图状态的 Python 字典。
  • 痛点：如果中途触发了中断（Human-in-the-Loop），中断信息会被硬塞进字典的 __interrupt__ 键里，状态和元数据混在一起，不够规范。
• v2 (version="v2")
  • 行为：返回一个专用的 GraphOutput 对象。
  • 优势：它将图状态和中断元数据彻底解耦。你可以通过 result.value 获取纯净的状态，通过 result.interrupts 干净地获取中断信息。

第二组：状态流式 —— stream / astream

它关心的是：“图在执行的每一步吐出了什么数据”（基于 stream_mode，如 values, updates, messages, custom）。

• v1（默认传统版）
  • 行为：输出格式不固定。只传 1 个模式时直接吐数据；传多个模式时吐 (mode, data) 元组；有子图时吐 (namespace, data) 元组。
  • 痛点：写代码时必须写一堆 if/else 或元组解包，极其容易出错。
• v2 (version="v2")
  • 行为：引入统一的 StreamPart 结构。无论你开几个模式、有没有子图，每个 chunk 的格式永远是固定字典：

    {"type": "...", "ns": (...), "data": ...}
    

  • 优势：格式高度统一，完美支持 IDE 的类型提示和自动补全。

第三组：全量事件与智能体流式 —— stream_events / astream_events

它关心的是：“整个应用内部发生了什么事件”（包括模型生成每一个 Token、工具开始/结束调用、子智能体嵌套、以及人类中断）。它不依赖 stream_mode，而是直接捕捉底层所有运行事件。

• v1 / v2（早期版本）
  • 行为：主要输出大而全的底层原始事件流（如 on_chat_model_stream、on_tool_start 等），格式较底层，开发者需要自己写很多过滤逻辑去拼凑 Token 或处理中断。
• v3 (version="v3")
  • 行为：现代 Agent 的首选方案。它在底层事件之上提供了类型化投影（Typed Projections）。
  • 优势：
    • 直接提供 stream.messages，让你可以像操作普通异步迭代器一样直接消费 LLM 生成的 Token。
    • 直接提供 stream.interrupted 和 stream.interrupts，让你可以一行代码判断智能体是不是暂停了，并直接对接 Command(resume=...)。

如何选择？

• 做简单的后端逻辑/不需要管中断：用 invoke() 或 stream() 并加上 version="v2"。
• 做终端 TUI / 聊天机器人 / 智能体（需要流畅的 Token 输出和人工审批中断）：直接上 stream_events(..., version="v3")。

做完这一切呢，效果是这样的：

中断：Agent 的“暂停键”——interrupt() 原理解析

上一节我们讲了 HumanInTheLoopMiddleware 的配置与用法，但你可能好奇：Agent 在执行过程中是怎么“暂停”下来等人类审批的？

答案藏在 LangGraph 的底层机制里：interrupt() 原语。

interrupt() 是 LangGraph 提供的一个中断原语，它的作用就是让图在执行到一半的时候停下来，等待外部输入。

from langgraph.types import interrupt

def some_node(state):
    # 执行到这里，图会暂停
    user_input = interrupt("请确认是否继续")
    # 恢复后，user_input 就是外部传入的值
    return {"confirmed": user_input}

当图执行到 interrupt() 时，会发生三件事：

1. 图暂停执行：当前图执行到此停止，不再继续
2. 状态被保存：LangGraph 利用 checkpointer 把当前状态完整保存下来
3. 返回中断信息：interrupt() 中传入的值（如 "请确认是否继续"）会作为中断信息返回给调用方

HumanInTheLoopMiddleware 本质上就是在工具执行前调用 interrupt()。

Agent 决定调用 write_file
        ↓
Middleware 检查 interrupt_on 配置
        ↓
write_file 需要审批？
        ↓
    是 → 调用 interrupt()，图暂停，等待人类决策
        ↓
    否 → 直接执行工具

当 interrupt() 被调用时：

• 状态被持久化：checkpointer 保存当前图状态，包括所有 messages 和执行位置
• 中断信息返回：调用方可以通过 state.tasks[0].interrupts 或 GraphOutput.interrupts 获取中断信息（包含工具名、参数、提示语等）
• 图等待恢复：执行无限期暂停，直到收到 Command(resume=...)

打开HumanInTheLoopMiddleware的源码，我们可以看到langgraph的中断原语被导入：

顺藤摸瓜下去，会发现在after_model中被调用：

在 LangGraph 的底层机制中，interrupt(value) 是实现“人类在环（Human-in-the-Loop）”的核心函数。简单来说，它就像是代码里的**“急刹车”和“邮递员”**。

当你在图的某个节点或工具里调用 interrupt("请问是否批准？") 时，它会分步执行以下三件事：

1. 瞬间“急刹车”（暂停执行）

   • 它会立刻中断当前节点的运行，并且抛出一个特殊的控制流信号（GraphBubbleUp），强行让整个 LangGraph 停止往下跑。
   • 任何后面的代码都不会执行，当前节点直接挂起。

2. 把数据“打包带走”（序列化保存）

   • 你传给 interrupt(value) 的参数（比如上面代码里的字符串、或者包含工具调用详情的字典），会被原封不动地交个底层的 Checkpointer。
   • Checkpointer 会把这个数据连同当前的整个图状态一起永久保存在数据库或内存中（也就是你在前端或 TUI 里看到的“审批详情/Payload”）。

3. “死等”人类的回复（挂起并释放）

   • 此时，程序不会报错，而是直接退出当前的 invoke 或 stream 运行，把控制权交还给你的主程序（比如你的 TUI 界面）。
   • 图就停在这里，可以静静地等几分钟、几小时甚至几天。

当人类在 TUI 里输入了选择（比如批准、拒绝、或是修改参数），你需要通过 Command(resume=用户输入) 再次唤醒图：

1. 当你带着 Command(resume="你的选择") 重新调用图时，LangGraph 会通过 thread_id 找到之前暂停的那个点。
2. 神奇的事情发生了：之前卡住的那行 value = interrupt(...) 会被“复活”，而你刚才传进去的 resume="你的选择" 会直接变成这个 interrupt() 函数的返回值，赋给左边的变量（即 value = "你的选择"）。
3. 节点拿到这个返回值后，带着人类的意见继续往下执行。

而langchain的任何一个调用的input都是支持一个状态机复制或者Command结构的：

为什么必须要有 checkpointer？

因为 interrupt() 依赖 checkpointer 来保存状态：

• 没有 checkpointer，图暂停后状态无法保存，下次调用时无从恢复
• 没有 checkpointer，Command(resume=...) 无法定位到之前暂停的位置

# ✅ 必须配置 checkpointer
agent = create_agent(
    model=...,
    middleware=[HumanInTheLoopMiddleware(interrupt_on={"delete_file": True})],
    checkpointer=MemorySaver(),  # 必须！
)

中断好理解，就是checkpointer直接保存当前state的快照，然后退出上下文，那么中断是如何做到的呢？

这就是 LangGraph 的精髓所在。

当你把 Command(resume=...) 传给 agent.astream(..., config=config) 时：

1. 框架在启动的一瞬间，会先去读取 checkpointer。
2. 它通过 thread_id 找到了上一次被保存的那个节点的快照（Snapshot）。
3. 框架发现这次输入的不是普通的文本或状态，而是一个 Command(resume=...)，它就知道：“哦！这是在恢复中断。”
4. 于是，它跳过了前面所有已经执行完的节点，直接把指针精准地拨回到上一次调用 interrupt() 的那个节点，并把你的 Command 里的值塞进去，让节点继续向下运行。

而Command 里的 resume 数据没有被直接塞进对话历史的 messages 里。

它的流转过程是这样的：

1. 中断发生时：
   中间件（如 HumanInTheLoopMiddleware）在底层调用了 interrupt()。此时大模型刚刚决定要调用某个工具（比如 write_file），但工具还没有执行。此时的图状态里，只有“AI 想要调用工具”的 AIMessage。
2. 人类审批通过时：
   你传的 Command(resume={"decisions": [{"type": "approve"}]}) 穿透到了中间件。
3. 中间件如何处理：
   • 如果是 approve（批准）：中间件直接在后台替你把那个原本被拦住的工具真正执行了，然后把工具返回的结果包装成一个标准的 ToolMessage 写入到图的 messages 历史中，再让图继续往下跑。
   • 如果是 reject（拒绝）：中间件不会去执行工具，而是直接伪造一个“用户拒绝执行”的 ToolMessage 写入历史，告诉大模型：“用户拒绝了，你看着办吧。”

所以，最终真正被写进 messages 对话历史的是工具的执行结果（ToolMessage），而你输入的 Command 只是用来驱动中间件做出“放行”还是“拦截”动作的指令（Instruction）。

比如这个案例：

from langgraph.types import interrupt

def approval_node(state: State):
    # Pause and ask for approval
    approved = interrupt("Do you approve this action?")

    # When you resume, Command(resume=...) returns that value here
    return {"approved": approved}

用 Python 内置的 input() 来类比 interrupt() 再恰当不过了：

• 传统的 input()：
  程序卡在命令行死等，期间整个进程阻塞，无法响应其他事情，用户一旦关掉终端或重启服务，输入的内容就丢了。
• LangGraph 的 interrupt()：
  它是一个**“有记忆、可持久化、带存档点”的超级版 input()**。
  1. 它同样会暂停（但不是死等阻塞，而是把整个图的运行状态打包存进 checkpointer 数据库，然后安全地退出程序/释放连接）。
  2. 当你稍后（无论是几秒钟后，还是几天后）带着用户的输入通过 Command(resume=...) 重新唤醒程序时，它会从数据库里复原一切，让 interrupt() 优雅地“苏醒”，并把用户输入的值赋给 approved 变量，继续往下执行。

这就是 LangGraph 做 Human-in-the-Loop（人机协同）最核心、最优雅的设计。

我们大致知道了理论，但interrupt()的内部又是如何具体代码实现的呢？

def foo():
    print("starting...")
    while True:
        res = yield 4        # ← 暂停点：返回 4，等待外部发送值
        print("res:", res)

g = foo()
print(next(g))               # 启动生成器，执行到 yield 暂停，返回 4
print("*" * 20)
print(g.send(7))             # 从暂停点恢复，yield 返回 7，继续执行到下一个 yield


starting...
4
********************
res: 7
4

我们先来看这个python的生成器，这个生成的执行流程如下，

步骤	调用	发生了什么	结果
1	next(g)	进入 foo()，执行 print("starting...")，遇到 yield 4	暂停，返回 4
2	g.send(7)	从暂停点恢复，yield 4 返回值变为 7，赋值给 res	打印 res: 7，继续循环，再次遇到 yield 4，再次暂停，返回 4

核心机制：

• yield 是“暂停点”，函数在 yield 处暂停，把控制权交还给调用者
• send() 从暂停点恢复，同时向生成器内部“注入”一个值，作为 yield 表达式的返回值

LangGraph 的 interrupt() 是一模一样的模式

def some_node(state):
    user_input = interrupt("请确认是否继续")  # ← 暂停点
    return {"confirmed": user_input}

步骤	LangGraph 调用	对应生成器的行为
1	图执行到 interrupt("请确认是否继续")	相当于生成器的 yield 4——暂停执行，返回中断信息（"请确认是否继续"）
2	调用方拿到中断信息，等待人类决策	相当于生成器的 next(g) 拿到了 4
3	人类决策后，用 Command(resume=resume_payload) 恢复	相当于生成器的 g.send(7)——恢复执行，interrupt() 的返回值变为 resume_payload
4	图继续执行，返回结果	相当于生成器继续运行到下一个 yield

它们的本质是一样的：都是“暂停-恢复”的流程控制模型。

区别在哪里？

	生成器	LangGraph interrupt()
暂停点	yield	interrupt()
恢复方式	send()	Command(resume=...)
状态保存	生成器的局部变量在内存中	状态通过 checkpointer 持久化到数据库
跨进程/跨时间恢复	❌ 不支持，进程重启后丢失	✅ 支持，只要有 thread_id 和 checkpointer

生成器是“内存级”的暂停-恢复，进程重启就丢了。而 interrupt() 配合 checkpointer，把状态持久化到了数据库，所以可以跨进程、跨时间恢复——这是 interrupt() 在生成器之上增加的“工程化能力”。

LangGraph 的 interrupt() 在底层的设计思想，几乎就是对 Python 这种生成器 send() 机制的工业级、分布式、持久化封装：

1. 第一次运行（对应你的 next(g)）：
   • 图运行到 interrupt()，相当于执行到了 yield 4。
   • 它把值（比如审批提示词）抛给外部，然后暂停并退出。
2. 挂起等待（对应你的 print("*"*20)）：
   • 此时 Python 的生成器如果关掉进程就没了，但 LangGraph 把这个“生成器的当前状态（断点位置、局部变量）”序列化后存进了 SQLite 数据库。
3. 恢复运行（对应你的 g.send(7)）：
   • 当你调用 Command(resume="你的选择") 时，LangGraph 从数据库里把状态捞回来，恢复那个“虚拟生成器”，并把你的选择通过 send() 塞了进去。
   • interrupt() 函数瞬间拿到了这个值，赋给左边的变量，然后继续往下执行。

LangGraph 极其精妙的地方，就在于它把 Python 这种单机内存里的 yield / send 魔法，搬到了分布式、可持久化的有状态工作流（StateGraph）里！

对于python中断，一直 next(g) 每次都会得到 4

def foo():
    print("starting...")
    while True:
        res = yield 4
        print("res:", res)

g = foo()

print(next(g))  # starting... \n 4
print(next(g))  # res: None \n 4
print(next(g))  # res: None \n 4
print(next(g))  # res: None \n 4

执行过程：

调用	发生了什么	输出
第一次 next(g)	启动生成器 → 遇到 yield 4 → 暂停，返回 4	starting... 4
第二次 next(g)	从暂停点恢复 → yield 4 返回 None → 赋值给 res → 打印 res: None → 继续循环 → 再次遇到 yield 4 → 暂停，返回 4	res: None 4
第三次 next(g)	同第二次	res: None 4

关键点：

• 没调用 send() 时，yield 4 的返回值是 None
• 然后继续执行 print("res:", res)，打印 res: None
• 然后进入下一次循环，再次遇到 yield 4，每次都返回 4

和 LangGraph interrupt() 的类比

• next(g) = 第一次执行时，图运行到 interrupt() 暂停，返回中断信息
• 再次 next(g) = 没有用 Command(resume=...) 恢复，而是重新调用了 invoke()，相当于“重新开始”，而不是“从暂停点继续”

如果你想真正“恢复”生成器，必须用 send()。同理，在 LangGraph 中，如果你想“恢复”中断的图，必须用 Command(resume=...)。

子代理——Agent 可以调用 Agent

聊完权限控制，我们进入另一个关键话题：子代理（Subagent）。

你可能会问：“一个 Agent 能干活就够了，为什么要多个？”

原因是：当单个 Agent 能力过载时，子代理是天然的分解方案。

想象一下，你让一个 Agent 做一个完整的 Web 项目：

• 它需要懂前端（React）
• 它需要懂后端（FastAPI）
• 它需要懂数据库（PostgreSQL）
• 它需要懂部署（Docker）

如果所有这些能力都塞进一个 Agent，它的上下文会越来越臃肿，工具列表会越来越长，模型在选择工具时也容易犯错。

子代理的核心思路就是：让一个 Agent 负责调度，让专门的子 Agent 负责执行特定领域的任务。

Deep Agents 提供了三种核心的多智能体协作模式，它们的核心区别在于：“谁来掌控对话的流向（Control Flow）” 以及 “智能体之间是如何组织和通信的”。

我们可以用“公司组织架构与协作方式”来理解它们：

模式	一句话概括	控制流	类比
Subagents	主 Agent 把子 Agent 当作工具来调用	集权制，主 Agent 一手掌控	经理 + 专员
Handoffs	子 Agent 主动将任务交接给其他 Agent	动态流转，平级转移	部门转接
Router	分类器将请求分发到对应的 Agent	一次性分发，分诊台指路	医院分诊台

💡 如果你觉得以上三种范式都不符合你的业务场景，你完全可以用 LangGraph 自行编排逻辑——把 Agent 当作节点嵌入图中，配合确定性代码（数据校验、条件分支、循环），实现完全自定义的协作流程。这也是 LangGraph 作为底层框架的真正威力所在。

模式一：Subagents（经理与专员模式）

工作机制：有一个主 Agent（Main Agent） 作为总指挥。子 Agent 被包装成普通工具（Tool）注册给主 Agent。主 Agent 根据用户需求，决定何时调用哪个子 Agent、传什么参数。

用户："帮我创建一个有登录功能的 Web 应用"
        ↓
主 Agent 接收任务，拆解需求
        ↓
调用 "frontend_subagent" → 生成 React 登录页面
调用 "backend_subagent" → 生成 FastAPI 认证接口
调用 "database_subagent" → 生成 PostgreSQL 表结构
        ↓
主 Agent 汇总所有结果，返回给用户

特点：

• 集权制，主 Agent 掌控所有路由决策
• 子 Agent 被封装为工具，主 Agent 不知道“对面是一个 Agent”
• 子 Agent 之间不直接通信，所有协调通过主 Agent

适用场景：层级分明的任务，如总客服按需调用“退款专员”和“物流专员”。

配置示例：

from deepagents import create_deep_agent
from deepagents.middleware import SubAgentMiddleware

frontend_subagent = {
    "name": "frontend_expert",
    "description": "专门负责前端开发，擅长 React、Vue、HTML/CSS",
    "system_prompt": "你是一个前端专家，只处理前端相关任务...",
    "tools": [write_file, read_file, ...],
}

backend_subagent = {
    "name": "backend_expert",
    "description": "专门负责后端开发，擅长 FastAPI、Django、Go",
    "system_prompt": "你是一个后端专家，只处理后端相关任务...",
    "tools": [write_file, read_file, execute_command, ...],
}

agent = create_deep_agent(
    model=model,
    middleware=[
        SubAgentMiddleware(
            subagents=[frontend_subagent, backend_subagent],
            default_model=model,
        ),
    ],
)

模式二：Handoffs（部门转接模式）

工作机制：没有绝对的“主”和“从”，大家平级。当当前 Agent 发现任务超出能力范围时，它会通过更新状态（State）或调用转接工具，把对话的控制权和上下文完整交接（Handoff） 给另一个 Agent。

用户："我想咨询产品功能"
        ↓
售前 Agent 接待 → 回答产品问题
        ↓
用户："我决定购买了，怎么付款？"
        ↓
售前 Agent 判断：需要转接到售后/财务
        ↓
Handoff 转接 → 控制权交给财务 Agent
        ↓
财务 Agent 接管对话，继续处理

特点：

• 动态流转，平级转移
• 交接时带上下文，用户不需要重复描述问题
• 交接后，原 Agent 不再参与对话

适用场景：长流程、多阶段且需要 Agent 直接和用户对话的场景（如：售前咨询 ➡️ 售后客服 ➡️ 财务专员）。

配置示例：

from deepagents import create_deep_agent
from deepagents.middleware import HandoffMiddleware

agent = create_deep_agent(
    model=model,
    middleware=[
        HandoffMiddleware(
            handoffs={
                "frontend_expert": {
                    "description": "前端专家，处理 UI 相关任务",
                    "system_prompt": "你是一个前端专家...",
                },
                "backend_expert": {
                    "description": "后端专家，处理 API 和数据相关任务",
                    "system_prompt": "你是一个后端专家...",
                },
            },
            default_handoff="frontend_expert",
        )
    ],
)

模式三：Router（路由器模式 / 分诊台）

工作机制：输入进来后，首先经过一个分类器（Routing Step）。分诊台判断这条请求该归哪个专业 Agent，然后一键直达，必要时甚至可以把多个专家的结果汇总（Synthesize）。

用户请求："写一个爬虫程序"
        ↓
   Router 分类器
        ↓
   判断：这是编程任务
        ↓
   路由到 "coding_expert" Agent
        ↓
   coding_expert 执行任务，返回结果

特点：

• 一次性分发，分诊台只负责指路，不参与后续多轮对话
• 支持并行分发：一个请求可以同时路由到多个 Agent，汇总结果
• 适合高并发场景

适用场景：高并发、分类明确的独立任务（如：用户发来的可能是写代码、写小说、查天气，路由器直接分发到对应的独立 Agent）。

配置示例：

from deepagents import create_deep_agent
from deepagents.middleware import RouterMiddleware

agent = create_deep_agent(
    model=model,
    middleware=[
        RouterMiddleware(
            routers={
                "coding": {
                    "description": "处理编程相关任务",
                    "system_prompt": "你是一个编程专家...",
                    "tools": [write_file, execute_command, ...],
                },
                "writing": {
                    "description": "处理文案创作相关任务",
                    "system_prompt": "你是一个文案专家...",
                    "tools": [write_file, ...],
                },
            },
            default_router="coding",
        )
    ],
)

模式对比与选型

模式	控制流	分布式开发	并行化	多跳支持	直接用户交互
Subagents	集权	⭐⭐⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐⭐⭐	⭐
Handoffs	动态流转	-	-	⭐⭐⭐⭐⭐	⭐⭐⭐⭐⭐
Router	一次性分发	⭐⭐⭐	⭐⭐⭐⭐⭐	-	⭐⭐⭐

• 分布式开发：不同团队能否独立维护各自的能力？
• 并行化：多个 Agent 能否同时执行？
• 多跳支持：是否支持串联调用多个子 Agent？
• 直接用户交互：子 Agent 能否直接与用户对话？

选型口诀

上下级汇报、把子 Agent 当工具：适合层级分明的任务。比如总客服（主智能体）按需调用“退款专员”和“物流专员”。 ➡️ Subagents

多轮对话中途转台、交接权限：长流程、多阶段且需要智能体直接和用户对话的场景（如：售前咨询 ➡️ 售后客服 ➡️ 财务专员）。 ➡️ Handoffs

进门先分诊、各走各的路：高并发、分类明确的独立任务（如：用户发来的可能是写代码、写小说、查天气，路由器直接分发到对应的独立 Agent）。 ➡️ Router

三种模式是怎么跑起来的？

理解了三种模式的应用场景后，我们再深入一层：它们底层到底是怎么实现的？

Subagents 是“主 Agent 调子 Agent”，Handoffs 是“Agent 自己换岗”，Router 是“门口分诊台”。它们的实现机制各不相同，我们逐一拆解。

Subagents：task 工具 + 独立上下文

说白了subagents就是按照职责把上下文拆分出去，我们先不管deepagents框架是怎么实现的，就按照自己的想法去发展！！

比如，我想把互联网搜索这个功能交给一个子agent去做，那么我只需要1. 初始化一个agent 2. 把主和子想办法关联起来。

from langchain.tools import tool
from langchain.agents import create_agent

# 创建子 Agent
subagent = create_agent(model="google_genai:gemini-3.6-flash", tools=[...])

# 将子 Agent 包装成工具
@tool("research", description="Research a topic and return findings")
def call_research_agent(query: str):
    result = subagent.invoke({"messages": [{"role": "user", "content": query}]})
    return result["messages"][-1].content

# 主 Agent，将子 Agent 作为工具之一
main_agent = create_agent(model="google_genai:gemini-3.6-flash", tools=[call_research_agent])

主 Agent 视角：我有个工具叫 “research”，描述是 "Research a topic…"用户需要研究时我就调用它，传入 query，拿到结果就行。

子 Agent 视角：我收到了一个用户消息（从主 Agent 转发过来的），我用我的工具和模型独立完成这个任务，把最终结果返回。

这就是 Subagents 模式的手工实现——把一个 Agent 包装成另一个 Agent 的工具，这样主agent就不需要关心子agent是怎么完成任务的，比如子agent第一次搜索失败了，子agent觉得信息不完美，于是去搜索了多个网站…

这么多的上下文信息如果放到主agent中，没啥用处占地方不说，还会污染上下文导致频繁的摘要！！

随着agent职责的增加，子agent越分越多，你需要不断去改代码，于是你想到了我能否用一个去分发任务！

于是第二种实现如下：

from typing import Literal
from langchain.agents import create_agent
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from pydantic import BaseModel, Field

# 1. 定义子智能体字典（也就是我们的“专家注册表”）
sub_agents = {
    "fruit": create_agent(
        model="openai:gpt-4o",
        prompt="你是一个水果专家。用一句话回答关于水果的问题。",
    ),
    "veggie": create_agent(
        model="openai:gpt-4o",
        prompt="你是一个蔬菜专家。用一句话回答关于蔬菜的问题。",
    ),
}

# 2. 定义统一的输入参数 Schema
class DelegateInput(BaseModel):
    expert_name: Literal["fruit", "veggie"] = Field(
        description="要咨询的专家名字：'fruit' (水果) 或 'veggie' (蔬菜)"
    )
    question: str = Field(description="向专家提出的具体问题")

# 3. 编写【单一分发工具】（Single Dispatch Tool）
@tool("delegate_to_expert", args_schema=DelegateInput)
async def delegate_to_expert(expert_name: str, question: str) -> str:
    """将问题动态分发给对应的专家智能体处理。"""
    # 动态从注册表中获取对应的 Agent
    agent = sub_agents.get(expert_name)
    if not agent:
        return f"错误：找不到名为 {expert_name} 的专家。"
    
    # 统一调用子 Agent
    response = await agent.ainvoke({
        "messages": [{"role": "user", "content": question}]
    })
    
    # 返回最后一个 AI 消息的内容
    return response["messages"][-1].content

# 4. 创建主 Agent，并把这个万能分发工具挂载上去
main_agent = create_agent(
    model="openai:gpt-4o",
    tools=[delegate_to_expert],
    prompt="你是一个总调度员。当用户问及水果或蔬菜时，必须使用 delegate_to_expert 工具将问题转交给对应的专家。",
    checkpointer=InMemorySaver(),
)

这是一个典型的 Single dispatch tool（单一分发工具） 的 Python 代码案例。

在这个例子中，我们只定义了一个万能工具 delegate_to_expert。主 Agent 只要在参数里指定 expert_name（比如 “fruit” 或 “veggie”），系统就会动态把任务分发给对应的子智能体：

为什么说它是“Single dispatch”？

1. 工具只有一个：主 Agent 的工具列表里永远只有 delegate_to_expert 这一根“光杆司令”。
2. 无限扩展性：如果明天公司新招了“肉类专家（meat）”，你完全不需要改动主 Agent 的 Prompt 或工具定义，只需要在上面的 sub_agents 字典里加一行 "meat": create_agent(...)，并在 Literal 里加一个选项，主 Agent 就自动学会找肉类专家了！
   Subagents 的实现核心是 SubAgentMiddleware。这个中间件会给主 Agent 添加一个叫 task 的工具——主 Agent 调用 task 工具时，实际上是在启动一个子 Agent。

到目前，你恍然大悟：“所谓的 Subagent，本质上不就是把另一个 Agent 包装成了一个 Tool 吗？”

确实如此。在这两种模式下，底层的技术本质都是“主 Agent 触发一个 Tool，这个 Tool 去运行一段子智能体逻辑”。它们的区别在于**“代码组织形式”和“定制化粒度”**：

1. Tool per agent（每个 Agent 一个专属工具）

   • 怎么写：你明确写出好几个独立的工具函数，例如 ask_fruit_expert()、ask_veggie_expert()。每个函数内部显式调用对应的子智能体。
   • 特点：
     • 高度定制：你可以针对水果专家定制专属的 Prompt、专属的输入 Schema（比如水果专属参数）、专属的输出拦截。
     • 代码量稍大：每增加一个子 Agent，就要写一个新的 @tool 函数。
   • 适用场景：子智能体之间差异很大，需要对它们的输入/输出做精细化控制。

2. Single dispatch tool（单一分发工具 / 动态调度工具）

   • 怎么写：只写一个通用的分发工具（比如叫 delegate_to_agent(agent_name: str, question: str)）。主 Agent 想找谁，就调用这个万能工具，把目标专家的名字和问题传进去。
   • 特点：
     • 极简优雅（约定优于配置）：团队里如果突然新增了 5 个子 Agent，你完全不需要修改主 Agent 的工具代码，只要把新 Agent 注册进注册表，万能工具会自动根据名字把任务派发过去。
     • 定制化低：所有子 Agent 走的是同一套输入/输出标准模板，无法针对某个特定 Agent 做高度定制的参数校验。
   • 适用场景：子智能体数量非常多、经常动态增减、或者由分布式不同团队分别开发（团队 A 只要把自己的 Agent 往注册表里一扔，主 Agent 就能直接用了）。

💡 总结

• Tool per agent 像是公司里给每个专员设独立分机号（找财务拨 101，找法务拨 102），安全、可控、但前台得记住一堆分机。
• Single dispatch tool 像是公司前台设了一个万能转接台（你在对讲机里喊：“帮我转财务部的张三”），灵活、好扩展、适合动态生长的多Agent系统。

SubAgentMiddleware 属于我们前面提到的第一种范式 —— Subagents（子智能体 / 经理与专员模式），具体实现上它属于 Tool per agent（每个 Agent 一个专属工具） 的变体。

在 Deep Agents 框架中，SubAgentMiddleware 的底层逻辑如下：

1. 自动挂载 task 工具：
   当你把 SubAgentMiddleware(subagents=[...]) 放入中间件栈时，它会在后台自动为主 Agent 注入一个名为 task 的工具。
2. 专属工具化：
   每一个在 subagents 列表里定义的子智能体（例如专属的 visualizer 绘图专员、reviewer 审查专员），都会被封装成独立的执行上下文。主 Agent 通过调用 task 工具并指定子专员的名字，把复杂的子任务外包出去。
3. 上下文隔离（Context Isolation）：
   子专员在它自己的独立上下文窗口（Context Window）里跑完所有的多轮对话和工具调用，最后只把精简后的最终报告（Final Report）返回给主 Agent，从而完美保持主 Agent 聊天历史的干净整洁。

我们进入到SubAgentMiddleware的源代码中可以看到，SubAgentMiddleware提供的仅仅是一个task工具：

Handoffs：状态变量 + Command 切换

Handoffs 的实现与 Subagents 完全不同——它不是通过“工具调用”来切换，而是通过“状态变更”来触发行为变化。

实现细节：

1. 核心机制是状态变量：系统维护一个状态变量（如 current_step 或 active_agent）。工具调用时，除了执行自己的逻辑，还会更新这个状态变量。系统在下一轮执行前读取这个变量，决定使用哪套配置（system_prompt、tools）。

2. 工具返回 Command：Handoffs 的核心是工具返回一个 Command 对象：

   return Command(
       update={
           "messages": [new ToolMessage({...})],
           "currentStep": "specialist"  # 触发行为切换
       }
   )
   

   Command 包含 update 字段，用于更新状态。框架读取更新后的状态，决定下一步行为。

3. 状态跨轮次持久化：一次 transfer_to_specialist 后，后续所有用户消息都由 specialist 配置处理，直到下一次状态变更。

4. 直接用户交互：Handoffs 模式支持子 Agent 直接与用户对话，这是它与 Subagents 的关键区别之一。

这是一个典型的 Handoffs（交接模式） 案例。

在这个案例中，我们构建了一个客户支持工作流状态机。它没有“主 Agent”去指挥小弟，而是只有同一个 Agent 随着状态（State）的变化，动态切换自己的 Prompt、规则和可用工具（即实现了权力的交接）：

from typing import Literal
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage, ToolMessage
from langchain.tools import tool, ToolRuntime
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
from typing_extensions import NotRequired
from langchain.agents import AgentState

model = init_chat_model("google_genai:gemini-3.6-flash")

# 1. 定义工作流的三个阶段（状态交接的阶梯）
SupportStep = Literal["warranty_collector", "issue_classifier", "resolution_specialist"]

class SupportState(AgentState):
    current_step: NotRequired[SupportStep]
    warranty_status: NotRequired[Literal["in_warranty", "out_of_warranty"]]
    issue_type: NotRequired[Literal["hardware", "software"]]

# 2. 阶段 A 的专有工具：记录保修状态，并【交接】到下一个阶段
@tool
def record_warranty_status(
    status: Literal["in_warranty", "out_of_warranty"],
    runtime: ToolRuntime[None, SupportState],
) -> Command:
    """记录保修状态，并将控制权交接给 issue_classifier 阶段。"""
    return Command(
        update={
            "messages": [ToolMessage(content=f"保修状态已记录: {status}", tool_call_id=runtime.tool_call_id)],
            "warranty_status": status,
            "current_step": "issue_classifier",  # 👈 状态交接点！
        }
    )

# 3. 阶段 B 的专有工具：记录问题类型，并【交接】到下一个阶段
@tool
def record_issue_type(
    issue_type: Literal["hardware", "software"],
    runtime: ToolRuntime[None, SupportState],
) -> Command:
    """记录问题类型，并将控制权交接给 resolution_specialist 阶段。"""
    return Command(
        update={
            "messages": [ToolMessage(content=f"问题类型已记录: {issue_type}", tool_call_id=runtime.tool_call_id)],
            "issue_type": issue_type,
            "current_step": "resolution_specialist",  # 👈 状态交接点！
        }
    )

@tool
def provide_solution(solution: str) -> str:
    """提供解决方案。"""
    return f"已提供方案: {solution}"

# 4. 不同阶段对应的 Prompt 配置
STEP_CONFIG = {
    "warranty_collector": {
        "prompt": "当前阶段：【保修验证】。请问用户的设备是否在保修期内？使用 record_warranty_status 记录。",
        "tools": [record_warranty_status],
    },
    "issue_classifier": {
        "prompt": "当前阶段：【问题分类】。保修状态：{warranty_status}。请询问用户故障并分类为 hardware 或 software，使用 record_issue_type 记录。",
        "tools": [record_issue_type],
    },
    "resolution_specialist": {
        "prompt": "当前阶段：【给出解法】。保修状态：{warranty_status}，类型：{issue_type}。请使用 provide_solution 给出对应方案。",
        "tools": [provide_solution],
    },
}

# 5. 用 Middleware 根据当前 `current_step` 动态切换 Agent 的性格与工具
@wrap_model_call
def apply_step_config(request: ModelRequest, handler) -> ModelResponse:
    current_step = request.state.get("current_step", "warranty_collector")
    config = STEP_CONFIG[current_step]
    
    # 动态切换当前阶段的系统提示词和工具
    system_prompt = config["prompt"].format(**request.state)
    return handler(request.override(system_prompt=system_prompt, tools=config["tools"]))

# 6. 创建支持 Handoffs 的单 Agent 状态机
agent = create_agent(
    model,
    tools=[record_warranty_status, record_issue_type, provide_solution],
    state_schema=SupportState,
    middleware=[apply_step_config],
    checkpointer=InMemorySaver(),
)

为什么这就是 Handoffs（交接模式）？

• 控制权的交接：在第一阶段结束时，record_warranty_status 工具返回的 Command 里把 current_step 从 "warranty_collector" 改成了 "issue_classifier"。
• 行为动态剧变：当下一轮对话开始时，apply_step_config 发现状态变了，立刻无缝切换了 Agent 的 Prompt 和可用工具。虽然名字还叫同一个 Agent，但它的“灵魂”已经完全变成另一个岗位（从前台接待无缝切换成了技术分诊员）。

相当于主agent在回答问题前，内部已经让不同的流程处理过了。 同一个模型，通过override的方式切换了提示词和工具。

这就是 Handoffs（交接模式）最妙的地方：

• 表面上看：用户在和同一个客服对话，感觉不到断层。
• 实际上：在底层，随着每次工具调用修改了状态（比如 current_step），系统在回答用户之前，就已经在后台通过 override 动态换掉了大模型的**“灵魂（System Prompt）”和“武器库（Tools）”**。

它既节省了多个独立大模型互相调用带来的 Token 开销，又通过状态机完美实现了复杂业务流程的有序交接！

Router：条件边 + 分类器

Router 是最简单直接的模式——一个分类步骤将请求分发到对应的专业 Agent。

实现细节：

1. 分类器决定去向：请求进来后，首先经过一个分类器（Routing Step），判断请求类型。分类后直接分发给对应的专业 Agent。

2. 一次性分发：Router 不维护对话状态。分发完成后，Router 不再参与后续对话。如果需要多轮对话，由被分发的 Agent 自己处理。

3. LangGraph 条件边实现：Router 通常通过 LangGraph 的 条件边（conditional edges） 实现——在图中加一个路由节点，根据输入内容走不同的分支：

   graph.add_conditional_edges("router", routing_function, {
       "coding": "coding_agent",
       "writing": "writing_agent",
       "analysis": "analysis_agent",
   })
   

4. 结果汇总：如果一个请求需要分发到多个 Agent，可以并行执行，然后汇总（Synthesize） 结果。

这是一个典型的 Router（路由器模式） 案例。

在这个例子中，有一个专门的分类/分诊节点，它接收用户的提问，通过结构化输出（Structured Output）判断这条请求应该交给哪个专家处理，然后利用 LangGraph 的 Command(goto=...) 将控制权直接导向对应的专业 Agent：

from typing import Literal
from pydantic import BaseModel, Field
from langchain_core.messages import HumanMessage
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from langgraph.checkpoint.memory import InMemorySaver

model = init_chat_model("google_genai:gemini-3.6-flash")

# 1. 定义状态
class RouterState(BaseModel):
    messages: list
    selected_agent: str | None = None

# 2. 定义分类模型输出的结构
class RouteQuery(BaseModel):
    destination: Literal["math_expert", "code_expert"] = Field(
        description="将用户的查询路由到对应的专家：'math_expert'(数学专家) 或 'code_expert'(代码专家)"
    )

# 3. 路由器节点：负责分类
def router_node(state: RouterState) -> Command:
    messages = state["messages"]
    
    # 让大模型判断应该去哪位专家那儿
    structured_llm = model.with_structured_output(RouteQuery)
    result = structured_llm.invoke(messages)
    
    # 使用 LangGraph 的 Command(goto=...) 进行动态路由跳转
    return Command(
        update={"selected_agent": result.destination},
        goto=result.destination # 👈 直接把控制权交给目标节点/Agent
    )

# 4. 创建两个专业的子 Agent 节点
math_agent = create_agent(model, prompt="你是一个数学专家，擅长解奥数和微积分。")
code_agent = create_agent(model, prompt="你是一个资深程序员，擅长 Python 和架构设计。")

def call_math_expert(state: RouterState):
    # 调用数学专家
    res = math_agent.invoke(state)
    return {"messages": res["messages"]}

def call_code_expert(state: RouterState):
    # 调用代码专家
    res = code_agent.invoke(state)
    return {"messages": res["messages"]}

# 5. 用 LangGraph 组装路由图
workflow = StateGraph(RouterState)

workflow.add_node("router", router_node)
workflow.add_node("math_expert", call_math_expert)
workflow.add_node("code_expert", call_code_expert)

# 入口直接指向分诊台 (router)
workflow.add_edge(START, "router")
# 专家处理完后直接结束
workflow.add_edge("math_expert", END)
workflow.add_edge("code_expert", END)

graph = workflow.compile(checkpointer=InMemorySaver())

为什么这就是 Router 模式？

• 分诊台机制：它有一个独立的 router 步骤（或节点），专门做“分流”动作，自身不回答具体业务问题。
• 精准直达：通过 Command(goto=...)（或 Send），输入一旦被分类，就会直奔目标专家而去，中间没有任何拖泥带水，非常适合多领域知识库检索、分类客服等场景。

需要注意，对于Handoffs和Router模式，deepagents没有直接提供中间件。

这是因为它们的定位和设计哲学不同：

1. Deep Agents 的强项是“深度单兵作战”（比如通过 SubAgentMiddleware 挂载代码解释器、文件系统、TodoList、子智能体等），它的核心关注点是上下文隔离与长任务分解。
2. Handoffs 和 Router 则属于“宏观图编排（Graph-level Orchestration）”：
   • Handoffs 通常通过编写自定义的 @wrap_model_call 拦截器（如刚才的案例）来实现，或者直接用标准的 LangGraph StateGraph 把多个独立 Agent 连起来。
   • Router 则天然更适合直接用 LangGraph 的 StateGraph + Command(goto=...) 来搭。

所以在实际开发中：

• 如果你想做复杂的、多专家自主交接或分流的系统，直接上手写标准的 LangGraph 状态图往往比硬套 Deep Agents 的中间件要更直观、更灵活！

动态子代理：脚本驱动的子代理编排

到目前为止，我们介绍的所有多智能体模式——Subagents、Handoffs、Router——都有一个共同点：子代理是静态配置的。你在初始化 Agent 时定义好所有子 Agent，之后它们就固定了。

但 Claude Code 和现代 AI 编程工具展示了一种更灵活的方式：动态子代理（Dynamic Subagents）。

静态子代理：

• 你在代码中预先定义好所有子 Agent：frontend_expert、backend_expert、database_expert……
• 主 Agent 通过 task 工具逐个调用它们
• 每个子 Agent 的职能是固定的，主 Agent 只能从预设列表中选择

# 静态：预先定义好所有子 Agent
subagents = [
    {"name": "frontend_expert", "system_prompt": "你是前端专家..."},
    {"name": "backend_expert", "system_prompt": "你是后端专家..."},
]

动态子代理：

• 子 Agent 按需创建，而不是预先定义
• 主 Agent 用代码脚本（而非逐个工具调用）来编排子 Agent
• 可以循环、分支、并发地创建和使用子 Agent

// 动态：用脚本驱动子代理执行
const results = await Promise.all(
    pages.map(page => task({
        description: `Summarize page ${page.number}`,
        subagentType: "summarizer"  // 按需指定类型
    }))
);

为什么需要动态子代理？

静态子代理在小规模场景下够用。但当任务规模变大时，问题就暴露了：

问题	静态子代理	动态子代理
大规模并行	主 Agent 需要逐个调用 300 次，每次都消耗一次推理	写一个 for 循环，一次性派发 300 个子 Agent
条件分支	无法根据前一个结果决定是否调用下一个	脚本中有 if/else，自然支持条件逻辑
覆盖保证	Agent 可能“偷懒”，只处理了 75/500 项就说完成了	循环结构保证每一项都被处理
上下文膨胀	每个子 Agent 的结果都回到主 Agent 的上下文	只有最终结果回到上下文，中间结果留在脚本变量中

核心洞察：让模型在“每次决策都调用一次工具”和“写一段脚本一次性编排所有任务”之间选择，后者的可靠性和效率都更高。

LangChain 通过 CodeInterpreterMiddleware 实现了动态子代理。

第一步：安装 QuickJS 解释器

pip install "deepagents[quickjs]"

第二步：配置 Interpreter 中间件和子代理

from deepagents import create_deep_agent
from langchain_quickjs import CodeInterpreterMiddleware

agent = create_deep_agent(
    model="openai:gpt-5.5",
    subagents=[
        {
            "name": "reviewer",
            "description": "Review code for security issues, citing lines and severity",
            "system_prompt": "You are a security-focused code reviewer...",
        },
        {
            "name": "summarizer",
            "description": "Summarize documents or code files",
            "system_prompt": "You are a summarization expert...",
        }
    ],
    middleware=[CodeInterpreterMiddleware()],
)

第三步：触发动态子代理

当用户提示中包含“workflow”或需要大规模编排的任务时，Agent 会编写 JavaScript 脚本来驱动子代理执行：

// Agent 生成的脚本示例
const files = ["file1.py", "file2.py", "file3.py", /* ... 50 files */];
const results = [];

for (const file of files) {
    // 每个文件启动一个 reviewer 子代理
    const result = await task({
        description: `Review ${file} for security issues`,
        subagentType: "reviewer"
    });
    results.push(result);
}

// 汇总所有结果
return results.filter(r => r.hasIssues).map(r => r.summary);

工作流程：

1. 主 Agent 理解任务（如“审查 50 个 Python 文件的安全问题”）
2. 主 Agent 用 JavaScript 写一段编排脚本（循环 + task() 调用）
3. 脚本在 QuickJS 解释器中执行
4. 解释器根据脚本逻辑，动态创建并调用子 Agent
5. 所有子 Agent 执行完毕，脚本返回最终结果给主 Agent

关键区别：脚本在解释器中执行，中间结果不会回到模型的上下文窗口，只有最终结果返回。这意味着动态子代理可以在不撑爆上下文的前提下，处理成百上千个子任务。

Claude Code 也有类似的概念，叫做 “动态工作流（Dynamic Workflows）” 。

在 Claude Code 中，动态工作流是一个 JavaScript 脚本，Claude 为描述的任务编写脚本，运行时在后台执行，同时会话保持响应。核心差异在于“谁持有计划”：

	普通子代理	动态工作流
谁决定下一步	Claude，每轮决策	脚本，预编排
中间结果在哪	Claude 的上下文窗口	脚本变量
可重复性	每次重新决策	脚本可保存、可重跑
规模	每轮几个任务	几十到几百个 Agent

Claude Code 的动态工作流可以用于代码库全面审计、500 个文件的迁移、多源交叉验证的研究等场景。

需要注意的是，动态子代理不是替代之前的三种模式，而是在它们之上增加了一个“脚本化编排”的维度：

┌─────────────────────────────────────────────────────────────┐
│  静态子代理（之前讲的）                                      │
│  └── Subagents：主 Agent 逐个调用子 Agent                   │
│  └── Handoffs：子 Agent 之间交接控制权                     │
│  └── Router：分类器一次性分发                               │
├─────────────────────────────────────────────────────────────┤
│  动态子代理（本节）                                         │
│  └── 主 Agent 写 JavaScript 脚本，脚本驱动子 Agent         │
│  └── 支持循环、分支、并发、大规模并行                      │
│  └── 中间结果不进上下文，只有最终结果返回                   │
└─────────────────────────────────────────────────────────────┘

它们是可以组合的：一个动态子代理脚本中调用的子 Agent，本身也可以是使用 Handoffs 或 Router 模式构建的复杂 Agent。

什么时候用动态子代理？

场景	静态子代理	动态子代理
三五步就能完成的任务	✅ 合适	❌ 过度设计
需要处理几十上百个独立单元	❌ 上下文爆炸	✅ 脚本循环搞定
逻辑有分支、重试、依赖关系	❌ 模型很难可靠执行	✅ 脚本的 if/else 天然支持
需要保证“每一项都处理到”	❌ 模型可能“偷懒”	✅ 循环结构保证全覆盖
需要可重复执行（CI/CD）	❌ 每次重新决策	✅ 脚本可保存重跑

Deep Agents 的思想：不是新框架，是“预配置全家桶”

当你看到 create_deep_agent 的时候，你可能会想：这又是个什么新的东西？LangChain 已经够复杂了，为什么还要再包一层？

但当你理解了 create_agent + 中间件的工作方式后，会发现一个事实：

create_deep_agent 就是 create_agent + 一套预配置的中间件栈。

你完全可以自己用 create_agent 拼出一个和 create_deep_agent 功能一样的 Agent。

# 你可以这样写（使用 create_deep_agent）
from deepagents import create_deep_agent

agent = create_deep_agent(
    model=model,
    backend=backend,
    system_prompt=system_prompt,
    checkpointer=checkpointer,
)

# 也可以这样写（使用 create_agent + 中间件）
from langchain.agents import create_agent
from deepagents.middleware import (
    TodoListMiddleware,
    FilesystemMiddleware,
    SkillsMiddleware,
    SubAgentMiddleware,
    SummarizationMiddleware,
)

agent = create_agent(
    model=model,
    middleware=[
        TodoListMiddleware(),          # 任务规划
        FilesystemMiddleware(backend),  # 文件系统
        SkillsMiddleware(...),          # 技能加载
        SubAgentMiddleware(...),        # 子代理
        SummarizationMiddleware(...),   # 上下文摘要
    ],
    system_prompt=system_prompt,
    checkpointer=checkpointer,
)

两种写法，本质上是一样的。

Deep Agents 默认带了什么？

Deep Agents 的“全家桶”包含了这些中间件：

中间件	作用
TodoListMiddleware	任务列表规划，让 Agent 学会“先列计划再执行”
FilesystemMiddleware	文件系统操作（ls、read_file、write_file、edit_file、glob、grep）
SkillsMiddleware	Skill 技能加载（按需加载领域知识）
SubAgentMiddleware	子代理调用（主 Agent 可以把任务委派给子 Agent）
SummarizationMiddleware	上下文摘要（防止上下文撑爆）
AnthropicPromptCachingMiddleware	提示缓存（Anthropic 模型专用，节省成本）
PatchToolCallsMiddleware	工具调用修补（兼容不同模型的工具调用格式差异）
MemoryMiddleware	记忆加载（注入 AGENTS.md 等指令）

为什么叫“Deep”？

“Deep”不是指模型深度，而是指 Agent 的“深度能力”：

• 深度规划（TodoList）：Agent 不只是“反应”，而是“先规划再行动”
• 深度上下文（Summarization）：Agent 在上下文快满时自动压缩，而不是直接崩溃
• 深度文件系统（Filesystem）：Agent 能像人类一样浏览、读写、编辑项目文件
• 深度知识注入（Skills）：Agent 按需加载领域知识，而不是把全部信息塞进初始 Prompt
• 深度委托（SubAgent）：Agent 把复杂任务拆解给专门的小组

这些能力叠加在一起，Agent 就不再只是一个“对话模型 + 工具调用”，而是一个可以独立完成复杂项目的 Agentic 系统。

那为什么还要有 Deep Agents？

如果 create_deep_agent 只是 create_agent 加中间件的语法糖，为什么不直接用 create_agent 自己拼？

因为 Deep Agents 提供了一套经过验证的默认配置。

你不需要自己去研究“哪些中间件组合在一起才能让 Agent 稳定工作”——Deep Agents 已经帮你配好了。就像你不需要自己编译 Linux 内核才能用 Ubuntu——别人已经帮你打包好了，你开箱即用。

	create_agent + 手拼中间件	create_deep_agent
学习成本	高，需要了解所有中间件的作用和顺序	低，一个参数搞定
配置复杂度	高，需要自己组装	低，内置默认值
灵活度	高，可以任意组合	中，可以覆盖默认配置
出错的概率	高，中间件顺序/配置不对容易出 bug	低，经过官方验证
适用场景	需要精细控制每一个组件	快速构建生产级 Agent

Deep Agents 的思想

Deep Agents 不是对 LangChain 的“重写”，而是对 LangChain 的“产品化”。

它把 LangChain 的各种底层组件（中间件、Backend、Checkpointer、Store）按照一个“可以用于生产环境”的最佳实践打包在一起。

你在进阶篇和高级篇里做的一切——FilesystemMiddleware、SkillsMiddleware、SummarizationMiddleware、HumanInTheLoopMiddleware——其实都在做同一件事：把 create_agent 拼装成 create_deep_agent。

Deep Agents 的意义不是“替代” create_agent，而是 “让你不用自己拼”。

是否应该用 Deep Agents？

我的看法是：

• 如果你清楚每个中间件的作用，并且需要精细控制 → 用 create_agent + 中间件，就像你在博客里做的那样
• 如果你只想快速开箱即用，不想折腾配置 → 用 create_deep_agent
• 如果两者之间的边界模糊了 → 恭喜你，你已经理解了 LangChain 的设计哲学

说到底，create_deep_agent 只是帮我把博客里写的那些中间件提前打包好了。

发布：从项目到 uvx 命令

博客写完了，代码也跑通了。下一步：怎么让其他人用上它？

你不可能让每个读者都去手动配置 LangChain、Deep Agents、OpenRouter 的 API Key，再自己写 main.py 把各个中间件组装起来。你需要把它变成一个一条命令就能安装、一条命令就能跑的工具。

这正是 uv 最擅长的场景。

完整发布流程

第一步：配置 pyproject.toml

你的项目目前可能只有 pyproject.toml 中的依赖声明。要让它成为可安装、可分发的包，需要补充三样东西：

1. 项目元数据：名称、版本、作者、描述
2. 依赖列表（你已经有了）
3. 入口点（entry point）：别人输入 langcode 后，执行哪个函数

[project]
name = "langcode-agent"
version = "0.1.0"
description = "A coding agent powered by LangChain and Deep Agents"
readme = "README.md"
requires-python = ">=3.11"
authors = [{name = "Your Name", email = "you@example.com"}]
dependencies = [
    "langchain>=1.0.0",
    "langgraph>=0.6.0",
    "deepagents>=0.1.0",
    "langchain-openai>=0.3.0",
    "rich>=13.0.0",
    "prompt-toolkit>=3.0.0",
    # ... 你的所有依赖
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project.scripts]
langcode= "langcode.main:main"

[project.scripts] 是关键——它把 langcode 这个命令指向了你 TUI 的入口函数。

注意路径写法：langcode.main:main 表示“从 langcode/main.py 中导入 main 函数”。你需要确保 main.py 里确实有 def main()。

第二步：调整项目结构，让 uv 能找到你的代码

uv 打包要求代码必须放在一个可导入的包（package）里。如果你的 main.py 直接放在项目根目录，uv build 会找不到它。

方案一：src 布局（推荐）

langcode/
├── src/
│   └── langcode/
│       ├── __init__.py          # 可以是空文件，也可以是导出声明的文件
│       ├── main.py              # 你的 TUI 入口
│       ├── agent.py             # 你的 create_agent 逻辑
│       ├── settings.py
│       ├── memory/
│       │   └── short_memory.py
│       └── ...
├── pyproject.toml
├── README.md
└── .python-version

__init__.py 内容：

# src/langcode/__init__.py
from .main import main

__all__ = ["main"]

这样 langcode 命令才能找到 main 函数。

方案二：扁平布局（也支持）

如果你不想调整目录结构，也可以保持现有结构，但需要把项目根目录作为包目录：

langcode/
├── main.py              # 入口函数在这个文件里
├── settings.py
├── memory/
├── ...
├── pyproject.toml
└── README.md

然后修改 pyproject.toml 的 [build-system] 配置（或者添加 [tool.hatch.build.targets.wheel]）：

[tool.hatch.build.targets.wheel]
packages = ["."]   # 把整个项目根目录打包为一个包

但这种方式容易把不需要的文件（如 .venv、__pycache__）也打包进去。需要配置 [tool.hatch.build.targets.wheel] 的 exclude 来排除。

推荐方案一（src 布局）：更干净，更标准，不容易出错。

第三步：迁移现有代码到 src 布局

假设你当前的目录结构是：

langcode/
├── main.py
├── settings.py
├── middleware/
├── memory/
├── prompt_builder/
├── router/
├── skills/
├── tui/
├── pyproject.toml
└── ...

移动代码到 src/langcode/ 下，并调整导入路径。

修改前：

# main.py
from middleware import fs_middleware
from memory.short_memory import SHORT_MEMORY_DB_PATH

修改后：

# src/langcode/main.py
from langcode.middleware import fs_middleware
from langcode.memory.short_memory import SHORT_MEMORY_DB_PATH

或者用相对导入（如果模块之间关系紧密）：

from .middleware import fs_middleware
from .memory.short_memory import SHORT_MEMORY_DB_PATH

第四步：确认入口函数

你的 main.py 里有一个 TUI 入口函数。可能是 async def main() 。在 [project.scripts] 里指向正确的函数名。

假设你的 TUI 入口是 tui（在 src/langcode/main.py 中）：

[project.scripts]
langcode= "langcode.main:main"

第五步：构建分发包

# 构建源码分发包（sdist）和二进制分发包（wheel）
uv build

uv build 会在 dist/ 目录生成两个文件：

• langcode-0.1.0.tar.gz（源码包）
• langcode-0.1.0-py3-none-any.whl（wheel 包）

第六步：发布到 PyPI

发布前，先去 pypi.org 注册账号，然后在 “Account settings” → “API tokens” 生成一个 token。

# 发布到 PyPI（需要 API token）
uv publish --token pypi-YOUR_TOKEN_HERE

如果不想直接发布到正式 PyPI，可以先推到测试环境验证：

# 先发布到 Test PyPI
uv publish --publish-url https://test.pypi.org/legacy/ --token pypi-YOUR_TEST_TOKEN

# 从测试环境安装验证
uvx --index https://test.pypi.org/simple/ langcode

第七步：验证安装

发布成功后，你的读者只需要一行命令：

uvx langcode

uvx 会自动完成三件事：

1. 从 PyPI 下载你的包
2. 在临时虚拟环境中安装所有依赖
3. 执行 [project.scripts] 里定义的 langcode 命令

无需手动 pip install，无需手动配置虚拟环境——这就是 uvx 的魔力。

发布前检查清单：

检查项	原因
[build-system] 已配置	缺少 build-system，uv build 无法执行
[project.scripts] 指向正确函数	入口点错了，用户执行 langcode 会报 ModuleNotFoundError
__init__.py 存在	Python 包必须包含 __init__.py 才能被导入
requires-python >= 3.11	你的代码用到了 async with aiosqlite.connect 等 Python 3.11+ 特性
所有依赖都在 dependencies 中声明	缺少依赖会导致用户无法运行
dist/ 目录已生成	确认 uv build 成功

假设你的最终结构是：

langcode/
├── src/
│   └── langcode/
│       ├── __init__.py
│       ├── main.py              # 包含 async def tui()
│       ├── agent.py
│       ├── settings.py
│       ├── backend.py
│       ├── memory/
│       │   └── short_memory.py
│       └── middleware/
│           ├── __init__.py
│           └── ...
├── pyproject.toml
├── README.md
└── .python-version

pyproject.toml：

[project]
name = "langcode"
version = "0.1.0"
description = "A coding agent powered by LangChain and Deep Agents"
readme = "README.md"
requires-python = ">=3.11"
authors = [{name = "Your Name", email = "you@example.com"}]
dependencies = [
    "langchain>=1.0.0",
    "langgraph>=0.6.0",
    "deepagents>=0.1.0",
    "langchain-openai>=0.3.0",
    "rich>=13.0.0",
    "prompt-toolkit>=3.0.0",
    "aiosqlite>=0.20.0",
    "pydantic>=2.0.0",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project.scripts]
langcode= "langcode.main:main"

src/langcode/__init__.py：

from .main import tui

__all__ = ["tui"]

src/langcode/main.py：

# 你的 TUI 实现
async def tui():
    # ...

构建并发布：

uv build
uv publish --token pypi-YOUR_TOKEN

读者使用：

uvx langcode

一次发布，所有人受益。博客教他们造车，uvx 直接送他们一辆能开的车。

本地验证：一条命令跑起你的 Agent

如果你只是想本地测试，不打算发布到 PyPI，根本不需要 uv build 和 uv publish。只需要两步：配置入口点，然后用 uv 工具模式运行。

第一步：在 pyproject.toml 中添加入口点

[project.scripts]
langcode= "langcode.main:main"

这里的格式是 "命令名 = "模块:函数名"。

但注意：uv 工具模式要求代码必须在一个可导入的包中。如果你的 main.py 在根目录，uv 会找不到它。最简单的做法是把它移到一个包目录下，比如 src/langcode/main.py，然后入口点写成 "langcode= "langcode.main:tui"。

注意这里要修改入口函数：

def main():
    """同步入口，供 uv tool 调用"""
    asyncio.run(loop())


if __name__ == '__main__':
    main()

项目目录：

第二步：用 uv tool 本地运行

有两种方式：

方式一：直接运行（不安装）

uv tool run langcode

uv tool run 会从当前目录加载 pyproject.toml，解析入口点，并执行对应的函数。但这种方式每次都要写 uv tool run，稍微有点长。

方式二：以可编辑模式安装（推荐）

uv tool install --editable .

这条命令会：

1. 将当前项目注册为一个名为 langcode 的工具
2. 根据 [project.scripts] 生成 langcode 命令，并添加到 PATH
3. --editable 表示“可编辑模式”，你修改代码后命令立即生效，不需要重新安装

安装后，在任何目录下打开终端，直接输入：

langcode

就能启动你的 Agent 了。

验证是否安装成功

which langcode   # 查看命令位置
langcode --help  # 如果你的入口函数支持参数，可以试试

卸载（如果需要）

uv tool uninstall langcode

结果如下：

尾声：从文档战神到工具制造者

到这里，我们的博客之旅终于走到了终点。

我从一个读了 200 万字文档、自以为“LangChain 我熟了”的文档战神，变成了一个在真实项目里被路径问题、小模型翻车、中间件顺序逼疯的实战者，最后又变成了一个能把代码打包成 uvx 工具、让别人一条命令跑起来的工具制造者。

这篇博客记录的不是一个“完美的 Agent 架构”，而是一个真实的、完整的踩坑与成长过程。

我们做了什么？

• 基础篇：从零手写 ReAct 循环，搞懂 Agent 的本质就是 model + tools + while
• 进阶篇：用 create_agent + 中间件把 Agent 工程化，接入 CompositeBackend、SkillsMiddleware、Checkpointer
• 高级篇：撕开抽象，理解权限控制、interrupt()、子代理、动态子代理，最终看透 Deep Agents 的本质——预配置全家桶
• 发布篇：把代码变成 uvx 命令，让 Agent 真正可用

最重要的三件事

1. 模型能力是天花板

小参数模型（4B-7B）只擅长纯文本对话，做不了 Coding Agent。它不是“不愿意”，而是“没学过”。如果你也想做一个能真正干活的 Agent，先用 120B 级别的大模型把逻辑调通，再考虑要不要降级。不要在模型能力的天花板上死磕。

2. 抽象解决的是“能不能做”，模型解决的是“愿不愿做”

LangChain 的所有抽象——Backend、Middleware、Checkpointer、Store——提供的都是“能力”。路径路由、文件系统、跨会话记忆——这些是“能做的事”。但模型到底“愿不愿”按照你的规则去做，取决于模型本身的指令遵循能力。你配置了 CompositeBackend，模型不读也没用——它需要中间件把路由规则翻译成它能理解的 System Prompt。

3. 先跑通，再优化，最后封装

这是写这篇博客最大的心得。如果一开始就用 create_deep_agent，我可能永远不知道它里面装了什么。从手写 while 循环开始，逐步替换成中间件，最后用 uv tool 打包——每一步都是可验证的，每一步都是可理解的。

“先跑通一个破烂的 Demo，再考虑完美的架构。”

关于 LangChain

坦率地说，LangChain 确实存在抽象冗余、文档滞后、不同团队各造各的轮子等问题。作为用户，感觉确实是“有点恶心”——同一个功能有三四种实现方式，路径路由和 Store 配置让人摸不着头脑。

但换个角度看：LangChain 是当前生态里唯一能把这些能力（文件系统、子代理、技能、记忆、审批）统一封装进一个 Agent 的方案。 它的混乱来自它要覆盖的场景太多，而不是它设计得差。

能驾驭 LangChain 的人，是能驾驭复杂性的人。

给读者的建议

如果你读完这篇博客，只记住一件事，我希望是这句话：

Agent 的本质就是“模型 + 工具 + 循环”。 所有框架、中间件、抽象，都是为这三样东西服务的。当你被抽象层搞得晕头转向时，回到这三样东西，你就不会迷失。

模型决定智商，工具决定能力，循环决定自动化程度。三者组合起来，就是 Agent 的全部。剩下的，都是封装。

最后的最后

两百万字的文档笔记，几十次失败的调试，无数个被 tool_calls 为空逼疯的夜晚——今天能把这些写成一篇完整的博客，我觉得值了。

如果你也正在读 LangChain 文档读到怀疑人生，或者被 Agent 的某个抽象层卡住了，希望这篇博客能帮你省下几万 Token 的学费。

代码会过时，模型会迭代，但 Agent 的本质不会变。

祝你好运，也希望你能造出属于自己的 Agent。

—— 一个从文档战神变回学生的开发者

🛠️ 完整代码已发布为 uvx 工具，你可以在任意目录执行 uvx langcode 体验。如果觉得好用，欢迎给项目点个 Star。

📦 完整代码已开源

这篇博客的所有示例代码、最终实现的 Agent、以及完整的项目结构，都已经整理好放到了 GitHub 上：

👉 https://github.com/QwQzy/langcode

仓库里包含了：

• 完整的 src/ 目录结构（所有模块和中间件）
• 可直接运行的 uvx 工具配置
• 示例 Skill 和配置文件
• 快速上手指南

你可以直接 clone 下来：

git clone https://https://github.com/QwQzy/langcode
cd langcode
uv tool install --editable .

然后在任意项目目录下执行 langcode，就能拥有一个属于自己的 Coding Agent。