Appearance
计划模式
当请求从"一次模型调用就能回答"升级到"需要多步、多工具、还要在执行中追踪进度"时,让模型靠自由推理 兼顾所有步骤很快就会失控:它会遗忘早期决策、重复劳动、或在中途丢失目标。AgentScope 2.x 的计划 模式(Plan)把这件事显式化--给 Agent 一组操作 Task 列表的工具,让它在每一步都看清"还剩什么、 被什么阻塞、下一步做什么"。
版本基线:本文以 Python
agentscope 2.0.4(PyPI,2026-07-07)为准,要求 Python>=3.11。 计划模式作为 building block 在2.0.3引入,2.0.4保持稳定。Task 工具与Task/TaskContext/AgentState的导出路径已与main分支源码核对,但 API 仍可能随小版本演进, 生产接入前请固定版本号并跟踪 release notes。
什么是计划模式
计划模式不是一个新的运行时,也不是一个独立的编排引擎。它是一组被注入到 Agent 状态里的工具 (state-injected tools):TaskCreate / TaskGet / TaskList / TaskUpdate。Agent 在 ReAct 循环中像调用普通工具一样调用它们,把一个复杂请求拆成一个有序、可追踪、可依赖的任务列表。任务 列表本身存在 agent.state.tasks_context(一个 TaskContext Pydantic 模型)里,随 AgentState 一起序列化,跨轮次可见。
关键在于"显式化"三个字:自由推理把计划藏在模型内部的思维链里,一旦上下文被压缩或会话恢复,计划 就丢了;计划模式把计划变成 Agent 可以读写、应用代码可以观测、会话可以持久化的外部状态。
前端类比
如果你来自前端:
- 把
tasks_context类比为应用里的 TODO list + 看板状态--它不是 UI 组件,而是被组件消费的 数据源,Task就是其中一条卡片。 - 把四个 Task 工具类比为 CRUD 接口:
TaskCreate是 POST、TaskGet是 GET 单条、TaskList是 GET 列表、TaskUpdate是 PATCH。Agent 通过 tool call 调用它们,就像前端调用 REST。 - 把
blocks/blocked_by类比为看板上的阻塞箭头:A 阻塞 B,等价于 B 被 A 阻塞,是同一条 依赖关系的两个方向。
AgentScope 原生语义:tasks_context 不是 React 的组件 state,而是挂在 AgentState 上的 Pydantic 模型,序列化由 AgentState 统一负责;四个 Task 工具不是 HTTP 接口,而是被 Toolkit 装配、受 Permission 拦截的 FunctionTool 子类,只是它们的 check_permissions 被硬编码为 ALLOW(详见下文"常见错误")。blocks / blocked_by 是 Task 上的两条双向边,AgentScope 不会自动同步它们,需要调用方手动保持一致。这与前端的"单向数据流"不同:这里的关系是图,不是树。
四个工具
| 工具 | 作用 |
|---|---|
TaskCreate | 创建一条新 Task,含 subject / description / 可选 blocked_by |
TaskGet | 按 id 读取单条 Task 的当前状态与依赖 |
TaskList | 列出 tasks_context 中的全部 Task,供 Agent 回顾全局进度 |
TaskUpdate | 更新 Task 的状态(pending -> in_progress -> completed)或字段 |
它们都从 agentscope.tool 导入:
python
from agentscope.tool import TaskCreate, TaskGet, TaskList, TaskUpdateTask / TaskContext / AgentState 从 agentscope.state 导入:
python
from agentscope.state import Task, TaskContext, AgentState这组导入路径已与 main 分支的 tool/__init__.py 与 state/__init__.py 核对一致。
用法:装进 Toolkit
Task 工具本身是普通的 FunctionTool 子类,使用方式和其他工具完全一样--装进 Toolkit,交给 Agent,模型在 ReAct 循环中自主决定何时创建、查询、更新任务。
python
import os
from agentscope.agent import Agent
from agentscope.tool import Toolkit, TaskCreate, TaskGet, TaskList, TaskUpdate
from agentscope.model import DashScopeChatModel # 或 OpenAIChatModel / AnthropicChatModel 等
from agentscope.credential import DashScopeCredential
model = DashScopeChatModel(
credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
model="qwen-plus", # 所选模型需支持 tool calling
)
toolkit = Toolkit(
tools=[
TaskCreate(),
TaskGet(),
TaskList(),
TaskUpdate(),
# 这里可以继续混入其他工具:Bash、Read、Write、MCPTool、FunctionTool ...
],
)
agent = Agent(
name="planner",
system_prompt="你是一个会把复杂请求拆成 Task 列表并逐步执行的 Agent。",
model=model,
toolkit=toolkit,
)
# Agent 通过 reply / reply_stream 自主调用 Task 工具
# result = await agent.reply(
# UserMsg(name="user", content="调研 AgentScope 2.x 的沙箱能力,产出一份中文摘要")
# )模型在执行过程中会先 TaskCreate 拆分任务,再 TaskList 回顾进度,每开始一项就 TaskUpdate 把它推进到 in_progress,完成后推进到 completed。应用代码不需要手工驱动这个循环。
编程式 seed 任务
并非所有任务都该让模型从零创建。很多时候你希望在 Agent 启动时就给它一个已规划好的任务骨架,模型 只负责执行与细化。这时可以直接操作 agent.state.tasks_context.tasks:
python
from agentscope.state import Task
# 在 Agent 创建后、调用 reply 之前 seed 初始任务
agent.state.tasks_context.tasks.extend(
[
Task(
id="t1",
subject="收集资料",
description="检索 AgentScope 2.x 沙箱相关文档与源码",
),
Task(
id="t2",
subject="整理摘要",
description="把收集到的资料整理成中文摘要",
blocked_by=["t1"], # t2 等 t1 完成
),
Task(
id="t3",
subject="输出终稿",
description="复核摘要并产出最终回复",
blocked_by=["t2"],
),
]
)seed 后,Agent 调用 TaskList 就能立即看到这份计划,并按照依赖关系选择下一个未被阻塞的任务推进。 注意:这里只写了 blocked_by,没有写 blocks--两者是同一条依赖的两个方向,下一节说明如何保持 一致。
Task 状态机与依赖图
状态机
每个 Task 的生命周期遵循下面的状态机:
pending --(start)--> in_progress --(complete)--> completed
| |
| +----(delete)----> deleted
+-----------------(delete)-----------------> deletedpending:已创建但未开始。in_progress:Agent 正在处理。completed:已完成。deleted:被删除,不再出现在活跃列表中。
状态转移通过 TaskUpdate 触发。Agent(或应用代码)调用 TaskUpdate(id=..., status=...) 即可 推进状态。
依赖:blocks 与 blocked_by
Task 上有两条互补的字段:
blocked_by:本任务被哪些任务阻塞(要等它们完成)。blocks:本任务阻塞了哪些任务(本任务完成后它们才能开始)。
两者描述的是同一条依赖关系的两个方向:A blocks B 等价于 B blocked_by A。AgentScope 不会 自动同步这两个字段--当你通过 TaskCreate 或编程式 seed 写入 blocked_by=["t1"] 时,t1 的 blocks 不会自动追加 t2。需要调用方手动保持双向一致:
python
# seed t2 时声明它被 t1 阻塞
Task(id="t2", subject="...", blocked_by=["t1"])
# 同时应手动在 t1 上记录它阻塞了 t2
# (找到 t1,把 t2 追加进 t1.blocks)
t1 = next(t for t in agent.state.tasks_context.tasks if t.id == "t1")
if "t2" not in (t1.blocks or []):
t1.blocks = (t1.blocks or []) + ["t2"]单向写 blocked_by 通常足以让 Agent 做出正确的调度决策(它会查 TaskList 找未完成的前置), 但如果你的应用代码会读 blocks 做依赖分析或可视化,就必须双向同步,否则会读到残缺的图。
存储:随 AgentState 序列化
tasks_context 不是临时内存里的列表,而是 AgentState 上的 Pydantic 模型。这意味着:
- 任务列表会随
AgentState一起被序列化、检查点、恢复。 - 跨多轮
reply调用,Agent 看到的是同一份任务列表。 - 会话恢复后,计划不会丢失。
序列化与恢复的细节由 AgentState 与会话层负责,详见 状态与会话。计划模式本身不关心存储后端,它只保证把 任务状态挂在 AgentState 上,由上层统一持久化。
与 1.x Pipeline 的区别
这是一个极易踩坑的迁移点,单独说明。
AgentScope 1.x 提供了一套声明式 Pipeline 用于编排:SequentialPipeline(顺序)、 ForLoopPipeline(循环)、WhileLoopPipeline(条件循环)、IfElsePipeline(分支)、 SwitchPipeline(多路分支)。开发者用这些 Pipeline 类把多个 Agent 串成固定拓扑。
AgentScope 2.x 已经移除了整套 Pipeline API。它不再提供"用配置拼流程图"的编排层,而是把编排 能力拆给三个更底层的 building block:
| 2.x 能力 | 替代了 1.x 的什么 |
|---|---|
| Plan(本页) | 把"多步任务怎么拆"从硬编码流程图变成 Agent 自管理 |
| Middleware | 把"每步前后做什么"(追踪、RAG、预算)变成洋葱中间件 |
| Agent Team | 把"多 Agent 怎么协作"变成 Leader-Worker 服务层 |
换句话说,1.x 的 Pipeline 是"应用代码决定拓扑",2.x 的 Plan 是"Agent 自主决定下一步"。两者 不是 API 换名,而是控制权从应用代码下放到 Agent。完整的 API 对照与迁移步骤见 1.x 到 2.x 迁移。
常见错误
- 双向边未同步:只写了
blocked_by没写blocks,应用代码读blocks时会拿到空列表。如果 只靠 Agent 调度,单向blocked_by通常够用;一旦有外部代码做依赖分析,就必须双向同步。 - Task 工具未装进 toolkit:
TaskCreate等是FunctionTool子类,必须显式装进Toolkit才能被 Agent 调用。漏装后 Agent 不会报错,但永远不会创建任务,行为退化为自由推理。 - 期望 Pipeline API:从 1.x 迁移时下意识找
SequentialPipeline,在 2.x 里不存在。需要把 "固定流程"重写为"Agent + Plan + Middleware"的组合,或用 Agent Team 做多 Agent 协作。 - 误以为 Task 工具受权限拦截:Task 工具的
check_permissions被硬编码为ALLOW,不会触发ask/deny。这是设计选择--任务列表是 Agent 的内部状态,不属于需要沙箱拦截的外部副作用。 如果你的场景需要对"创建任务"本身审批,需要自己包一层中间件,不能依赖Permission。 - 把 tasks_context 当长期存储:
tasks_context随AgentState序列化,但跨会话的长期持久化 取决于你配置的会话后端。单进程内存后端重启即丢,生产场景需要持久后端(详见 状态与会话)。
先修
- 结构化输出 - 理解
FunctionTool与 Pydantic 模型 的关系,Task 工具就是FunctionTool子类
下一步
- 上下文管理 - 任务列表变长后如何压缩与卸载