前面的文章介绍了 Deep Agents 的架构，并实现了一个简单的 Research Agent。

Research Agent 的执行过程很明确：搜索资料、读取网页、整理结果。

而实际任务中，有些往往要较为复杂。

例如，让 Agent 检查某个版本是否具备上线条件。它需要查看代码变更、检查测试结果、调查异常，最后给出发布建议。如果发现新的问题，还需要调整后续工作。

Deep Agents 通过 Planning 机制管理这类任务，使 Agent 能够制定计划、记录进度，并根据执行结果调整安排。

本文重点介绍 TodoListMiddleware、write_todos 及其运行原理。

一、Planning 的基本原理

普通 Agent 通过循环完成任务：

代码片段Text
LLM 推理 → 工具调用 → 获取结果 → 继续推理

模型根据已有消息和工具结果，决定下一步行动。

复杂任务可能涉及几十次工具调用。执行过程中，模型既要处理当前问题，又要记住哪些工作已经完成。

这些信息通常散布在消息历史中。随着任务推进，准确掌握进度会变得越来越困难。

Planning 为此引入了任务列表。

以版本发布检查为例：

代码片段Text
[进行中] 查看版本变更
[待执行] 检查自动化测试
[待执行] 评估发布风险
[待执行] 形成上线建议

任务列表保存在 Agent 的 State 中，供模型在执行过程中读取和更新。

State 是 LangGraph 保存运行状态的数据结构。例如，messages 保存消息历史，todos 保存任务计划。

Todo 的数据结构

Deep Agents 使用的 Todo 可以表示为：

代码片段Python
from typing import TypedDict, Literal
 
 
class Todo(TypedDict):
    content: str
    status: Literal[
        "pending",
        "in_progress",
        "completed",
    ]

例如：

代码片段Python
todo: Todo = {
    "content": "检查自动化测试",
    "status": "in_progress",
}

status 支持三种状态：

状态含义
pending待执行
in_progress进行中
completed已完成

Todo 只包含任务内容和状态，没有专门的优先级、截止时间或依赖关系字段。这样的结构已经能够支持基本的任务规划和进度管理。

二、Planning 的实现机制

Deep Agents 使用 TodoListMiddleware 提供 Planning 能力。

Middleware 是中间件，用于向 Agent 添加额外功能，例如任务规划、工具管理和错误处理。

从 Deep Agents 0.7 开始，Planning 改为按需启用，需要显式配置：

代码片段Python
from deepagents import create_deep_agent
from langchain.agents.middleware import (
    TodoListMiddleware,
)
 
agent = create_deep_agent(
    model="openai:gpt-5.5,
    middleware=[
        TodoListMiddleware(),
    ],
)

create_deep_agent() 用于创建 Agent。middleware 接收中间件列表。

启用 TodoListMiddleware 后，Agent 获得以下能力：

  1. 通过 write_todos 创建和更新任务列表。
  2. 在 State 的 todos 字段中保存计划。
  3. 根据中间件提供的指令管理任务进度。

write_todos 的参数

write_todos 接收一个 todos 参数，类型为：

代码片段Python
todos: list[Todo]

list[Todo] 表示由多个 Todo 字典组成的列表。

例如，模型可能生成以下工具参数：

代码片段JSON
{
  "todos": [
    {
      "content": "查看版本变更",
      "status": "in_progress"
    },
    {
      "content": "检查自动化测试",
      "status": "pending"
    },
    {
      "content": "评估发布风险",
      "status": "pending"
    }
  ]
}

工具输入通过 Pydantic 模型校验，确保数据结构符合要求。

write_todos 每次接收完整的任务列表，并用新列表更新 State 中的 todos。

例如，第一项任务完成后，模型需要保留其他任务，同时将对应状态改为 completed。

由于每次更新都会替换整份列表，并行调用可能导致冲突。因此，中间件会检查并阻止同一轮模型响应中的多个 write_todos 调用。

自定义 Planning 指令

TodoListMiddleware 支持两个配置参数：

参数作用
system_prompt指导模型何时以及如何维护计划
tool_description自定义 write_todos 的工具描述

例如：

代码片段Python
planning = TodoListMiddleware(
    system_prompt=(
        "处理多阶段任务时，使用write_todos"
        "制定计划并记录进度。"
        "发现新问题时及时调整计划。"
    ),
)

自定义 system_prompt 会替换中间件默认的 Planning 指令。一般情况下，保留默认配置即可。

三、一个完整的 Planning 示例

下面实现一个版本发布检查 Agent。

我们要实现的需求如下：

检查本次软件版本是否具备上线条件，并给出发布建议。

Agent 需要查询代码变更和测试结果。如果发现失败项，还要调查具体原因。

为了理解 Planning，示例中的工具均使用模拟数据。

安装依赖：

代码片段Bash
pip install -U deepagents langchain-openai

配置 OPENAI_API_KEY 环境变量后，运行以下代码：

代码片段Python
from deepagents import create_deep_agent
from langchain.agents.middleware import (
    TodoListMiddleware,
)
 
 
def get_release_changes() -> str:
    """查询本次版本的代码变更。"""
    return (
        "本次修改涉及登录状态续期、"
        "订单折扣计算和个人中心页面样式。"
    )
 
 
def get_test_report() -> str:
    """查询自动化测试结果。"""
    return (
        "共120项测试，118项通过，2项失败。"
        "失败项：login_refresh、"
        "discount_calculation。"
    )
 
 
def get_failure_details(test_name: str) -> str:
    """根据测试名称查询失败详情。"""
    failures = {
        "login_refresh":
            "Token过期后未能正常续期。",
        "discount_calculation":
            "部分订单折扣金额计算错误。",
    }
 
    return failures.get(
        test_name,
        "没有找到对应测试",
    )
 
 
agent = create_deep_agent(
    model="openai:gpt-5.5",
    tools=[
        get_release_changes,
        get_test_report,
        get_failure_details,
    ],
    middleware=[
        TodoListMiddleware(),
    ],
    system_prompt=(
        "你是一名软件发布检查助手。"
        "处理多步骤任务时，使用write_todos"
        "制定计划并更新进度。"
        "发现测试失败时，调查具体原因。"
        "根据工具结果评估发布风险，"
        "不得将失败的检查视为通过。"
    ),
)
 
request = {
    "messages": [
        {
            "role": "user",
            "content": (
                "检查本次版本是否具备上线条件，"
                "并给出发布建议。"
            ),
        }
    ]
}
 
result = agent.invoke(request)
 
todos = result.get("todos") or []
 
if todos:
    print("任务计划：")
    for todo in todos:
        print(todo["status"], todo["content"])
else:
    print("本次执行没有创建任务计划。")
 
print("\n检查结论：")
print(result["messages"][-1].content)

这里通过 create_deep_agent() 创建 Agent，并传入四个主要参数：

参数作用
model指定语言模型
tools注册业务工具
middleware配置中间件
system_prompt定义 Agent 的工作要求

最后，通过 agent.invoke(request) 执行 Agent，正常结束后返回 State。

其中，result["messages"] 保存消息记录，result.get("todos") 用于读取任务列表。

一次可能的执行过程如下：

代码片段Text
创建任务计划
     ↓
查询版本变更
     ↓
检查自动化测试
     ↓
发现两项测试失败
     ↓
调整任务计划
     ↓
调查失败原因
     ↓
评估发布风险
     ↓
输出检查结论

根据示例中模拟的数据，Agent 有理由建议暂缓发布，因为存在影响发布的问题。

实际执行顺序由模型决定，任务列表也可能有所不同。

四、为什么有时没有 todos

运行上面的代码时，可能发现一个现象：

有些模型会返回完整的 todos，有些模型则没有。即使使用同一个模型，多次执行相同任务，也可能出现不同结果。

这是 Planning 的正常行为。

启用了 TodoListMiddleware，并不意味着模型一定会调用 write_todos。

中间件提供任务规划能力，模型根据任务和工具说明决定是否使用。

例如，模型可能认为当前任务只需几次工具调用，于是直接查询数据并给出结论，没有创建任务列表。

PlanningState 的可选字段

从 Deep Agents 的源码可以看到，Planning 的 State 定义包含：

代码片段Python
class PlanningState(AgentState[ResponseT]):
    todos: Annotated[
        NotRequired[list[Todo]],
        OmitFromInput,
    ]

这里有一个类型值得关注。

NotRequired 表示字段可以不存在。

因此，todos 是一个可选的状态字段。

如果模型始终没有调用 write_todos，State 中就可能不存在该字段。

这也是示例采用以下写法的原因：

代码片段Python
todos = result.get("todos") or []

无论字段不存在、值为 None，还是空列表，最终都能得到可遍历的列表。

为什么模型行为不同

模型是否使用 Planning，主要受以下因素影响：

  • 任务复杂度：简单任务可能直接执行。
  • 模型能力与工具选择策略：不同模型对规划必要性的判断存在差异。
  • Prompt 与上下文：指令可以引导工具使用，但不能保证模型一定调用。

同一模型的不同运行结果，也可能因生成过程中的选择而发生变化。

如果强制必须使用 Planning，也可以尝试在 system_prompt 中加强约束，要求必须调用 write_todos。

检查实际执行情况

我们可以在 invoke() 之后增加：

代码片段Python
print("State字段：", list(result.keys()))
 
for message in result.get("messages", []):
    for call in getattr(
        message, "tool_calls", []
    ):
        print("工具调用：", call["name"])

通过这些记录，可以判断模型是否调用了 write_todos。

如果没有调用，通常说明模型选择了直接执行。

如果存在调用，但最终没有 todos，则需要进一步检查对应工具是否成功执行，以及运行过程中是否发生中断或异常。

五、观察 Planning 的执行过程

invoke() 返回最终结果。如果希望了解 Agent 怎样创建和修改计划，可以进一步查看工具调用记录。

查看 write_todos 调用

LangChain 使用 AIMessage 保存模型消息，其中的 tool_calls 记录模型发起的工具调用。

在执行结束后增加：

代码片段Python
for message in result["messages"]:
    for call in getattr(
        message, "tool_calls", []
    ):
        if call["name"] == "write_todos":
            print(call["args"]["todos"])

这样可以看到模型历次提交的任务列表。运行结果如下：

代码片段JSON
[{'content': '获取版本变更与测试报告概况', 'status': 'completed'}, {'content': '调查 login_refresh 失败详情', 'status': 'in_progress'}, {'content': '调查 discount_calculation 失败详情', 'status': 'pending'}, {'content': '评估风险并给出发布建议', 'status': 'pending'}]
[{'content': '获取版本变更与测试报告概况', 'status': 'completed'}, {'content': '调查 login_refresh 失败详情', 'status': 'completed'}, {'content': '调查 discount_calculation 失败详情', 'status': 'completed'}, {'content': '评估风险并给出发布建议', 'status': 'completed'}]

使用 stream() 实时观察

如果希望在执行过程中获取计划变化，可以使用 stream()。

LangGraph 支持多种流式输出模式：

模式返回内容
values每个步骤结束后的完整 State
updates每个步骤产生的状态更新
messages模型生成过程中的消息数据

观察 Planning 时，可以使用 values：

代码片段Python
last_todos = None
 
for state in agent.stream(
    request,
    stream_mode="values",
):
    todos = state.get("todos") or []
 
    snapshot = [
        (t["content"], t["status"])
        for t in todos
    ]
 
    if todos and snapshot != last_todos:
        print("\n计划更新：")
 
        for todo in todos:
            print(
                todo["status"],
                todo["content"],
            )
 
        last_todos = snapshot

stream() 逐步返回执行状态。

last_todos 保存上一次的任务列表，避免重复输出。

例如，当 Agent 发现测试失败后，可能产生这样的更新：

代码片段Text
completed   查看版本变更
completed   检查自动化测试
in_progress 调查失败测试
pending     评估发布风险
pending     形成上线建议

通过这种方式，我们可以观察任务进度，也可以将状态展示在前端界面中。

六、write_todos 的底层实现

write_todos 如何将计划保存到 State？

关键在于 LangGraph 的 Command。

中间件内部的核心逻辑可以简化为：

代码片段Python
from langgraph.types import Command
from langchain_core.messages import ToolMessage
 
 
def update_todos(runtime, todos):
    return Command(
        update={
            "todos": todos,
            "messages": [
                ToolMessage(
                    content="Updated todo list",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
        }
    )

这里省略了部分类型声明和处理逻辑，仅展示状态更新过程。

Command 用于向 LangGraph 提交状态更新。

其中，update 指定需要修改的字段：todos 保存新任务列表，messages 记录工具执行结果。

ToolMessage 表示工具返回的消息，tool_call_id 用于关联模型发起的工具调用。

ToolRuntime

中间件内部通过 ToolRuntime 获取运行时信息。

相关函数签名可以表示为：

代码片段Python
def _write_todos(
    runtime: ToolRuntime,
    todos: list[Todo],
) -> Command:
    ...

ToolRuntime 由框架注入，包含当前工具调用 ID 等信息。模型只需要提供 todos 参数。

-> Command 表示函数返回一个 Command 对象。

因此，Planning 的状态更新过程可以概括为：

代码片段Text
LLM 生成计划
     ↓
调用 write_todos
     ↓
校验工具参数
     ↓
返回 Command
     ↓
更新 State.todos
     ↓
Agent 继续执行

总结下来，模型负责生成和调整计划，write_todos 提交更新，LangGraph 保存状态。

七、计划调整与能力边界

Planning 支持根据执行结果调整任务。

例如，原来的发布检查计划为：

代码片段Text
查看变更 → 检查测试 → 评估风险 → 给出建议

发现测试失败后，可以增加调查任务：

代码片段Text
查看变更
   ↓
检查测试
   ↓
调查失败原因
   ↓
评估风险
   ↓
给出建议

这样的调整，同样可以通过 write_todos 完成。

当然，需要注意，计划也不宜频繁修改。只有新信息影响执行方向、任务范围或完成条件时，调整才有必要。

completed 只表示任务状态已经更新，不能直接证明工作结果正确。

此外，Planning 也不负责中断恢复。对于需要长期运行的任务，还要结合 Checkpointer 等持久化机制。

总结

Deep Agents 通过 TodoListMiddleware 提供 Planning 能力，由 write_todos 创建和更新任务列表，并将计划保存在 LangGraph State 中。

模型可以根据任务制定计划，调用工具完成工作，再依据执行结果调整安排。开发者则可以通过 invoke()、stream() 和工具调用记录观察执行过程。

Planning 是一项可选能力。启用中间件后，模型仍可能选择直接执行，此时 State 中不一定存在 todos 字段。

这套机制改善了复杂任务的完成方式，但执行质量仍然取决于任务安排、工具结果和完成标准。