AgentScope 2.0 源码深度解析：把 Agent 做成可暂停、可恢复的生产运行时

源码基线：本文基于本地 AgentScope 2.0.5、提交 c613a860 分析。不同提交的实现可能变化；文中的文件链接固定到该快照。

许多 Agent 框架的最小形态都可以写成：把历史消息和工具 Schema 交给模型，模型返回工具调用就执行，再把结果塞回消息列表。

这段循环很容易写，难的是把它变成一个可在产品中运行的系统：工具请求需要确认时如何暂停？用户返回确认后如何从原位置继续？流式 token、工具进度和最终回答如何统一给前端？多个进程如何保证同一会话不被并发写坏？子 Agent 的消息怎样唤醒另一个会话？

AgentScope 2.0 的回答不是把这些能力散落在应用代码中，而是围绕可持久化的 AgentState、Msg 与可序列化的 AgentEvent 协议，建立一个可暂停、可恢复、可观察的 Agent 运行时。Agent 是控制内核；Toolkit、PermissionEngine 和 Workspace 管住能力边界；ChatService、Storage、MessageBus 与 SSE 把同一套内核推到生产环境。

本文的阅读路线：

一、先给结论：AgentScope 2.0 的核心不是“多模型适配”

pyproject.toml 给出的定位是 A Flexible yet Robust Multi-Agent Platform，但从源码看，2.0 最重要的工程选择有四个：

1. 把一次回复建模为状态机，而不是单次函数调用：工具进入 ASKING 或 SUBMITTED 后，回复可以安全挂起；确认或外部结果会作为下一次输入恢复同一状态。
2. 以细粒度事件作为运行时协议：文本、思考、工具参数、工具结果、确认请求和结束状态都进入统一 AgentEvent 流；前端不必理解每个模型厂商的原始流式格式。
3. 把副作用置于权限与工作区边界之后：模型产生 ToolCallBlock 只代表“提议”；Schema 校验、权限决策、人工确认、外部执行和结果回写才决定“发生什么”。
4. 服务层复用 SDK 内核，而不是另写一套编排器：ChatService 每次按 session 组装 Agent，事件进入消息总线；HTTP 负责触发，SSE 负责订阅，团队协作也复用 inbox + wakeup。

因此，它更接近一个 Agent execution runtime：模型只是其中一个可替换的推理部件。

二、源码地图：从包结构判断设计重心

项目要求 Python >=3.11，使用 Pydantic、asyncio、MCP、OpenTelemetry 等依赖；服务、向量库、工作区后端、长记忆被拆为 optional extras。核心代码位于 src/agentscope。

这里有一个容易误判的地方：2.0 源码中没有一个通用 workflow/ 或静态 DAG 包作为主心骨。它的“工作流”主要由 ReAct 状态机、任务工具、后台任务、定时调度与 Team 消息唤醒组成。换言之，它优先解决动态 Agent 执行，而不是先定义一张固定图再运行。

2.1 两个入口：SDK 直接跑，或装入 FastAPI

SDK 使用者直接构造 Agent，调用 reply_stream() 消费事件；只关心最终结果则调用 reply()。

agent = Agent(
    name="assistant",
    system_prompt="You are a helpful assistant.",
    model=model,
    toolkit=toolkit,
)

async for event in agent.reply_stream(UserMsg("user", "分析这个仓库")):
    render(event)

服务模式则从 create_app() 开始。调用者必须提供 StorageBase、MessageBus 与 WorkspaceManagerBase；知识库、额外工具、中间件和子 Agent 模板按需注入。这个参数设计传达了一个事实：生产化 Agent 至少需要 持久化、传输和受控执行空间，它们不是可有可无的后缀。

三、核心执行链：reply_stream() 如何成为可恢复的 ReAct 状态机

reply_stream() 本身非常薄：它调用私有 _reply()，默认过滤最终 Msg，把过程中的 AgentEvent 交给消费者。reply() 只是把同一条流消费到底，取回最后的 Msg。

真正的内核在 _reply_impl()。它把输入统一为四类：

关键不在于类型很多，而在于 “恢复”不是重放 prompt。工具调用及其状态已经写入 AgentState.context；恢复事件只修改对应调用的状态，接着继续原有循环。这避免了模型重新规划、重复调用工具或丢失前序结果。

3.1 _next_action()：把控制流从执行逻辑中抽离

每轮循环先调用 _next_action()，结果只可能是三个代数状态：

• Reasoning：需要调用模型；
• Acting(tool_calls)：已有允许执行的工具调用；
• Exit：最终消息可返回、结构化输出已经满足，或抵达最大迭代数。

这比“模型调用后立即执行工具”的直接写法多了一层，但换来两个收益：

• 决策可读：等待确认、继续推理、执行工具、超限退出的优先级集中在一个函数；
• 暂停自然：发现 ASKING 或 SUBMITTED 调用时，返回 Exit，但不将回复标记为完成。后续事件回来后再推进。

其核心顺序可以概括为：先找可执行工具；再看是否有等待中的工具；然后检查结构化输出；最后才决定用文本最终消息退出或继续推理。工具调用生命周期不再隐含在调用栈里，而显式存在于消息块状态中。

3.2 推理前做三件事：压缩、注入、准备模型输入

在 Reasoning 分支中，框架按顺序执行：

1. compress_context()：超阈值时压缩历史；
2. _inject_runtime_state()：把时间、未完成任务、接近上下文上限等变化信息写入 HintBlock；
3. _reasoning()：准备系统提示、摘要、对话上下文和工具 Schema，调用模型并把模型流转成事件。

一个很细的设计是：运行时状态不直接拼进 system prompt，而是持久化为 HintBlock。源码明确说明这是为了让固定 system prompt 保持稳定，保留 prompt cache 的机会，同时让 Agent 感知会变化的时间和任务状态。

3.3 工具调用为什么有“批次”而不是全部并发

模型可能一次给出多个工具调用。_batch_tool_calls() 根据工具的 is_concurrency_safe 属性切成顺序或并发批次；并发细节实现在 _execute_concurrent_tool_calls()：

• 并发批次通过 asyncio.gather() 执行，每个 worker 的事件进入共享队列；即便一个工具失败，其余工具仍会完成；
• 顺序批次遇到确认或外部执行请求会立即停下，避免后续副作用跨过人工 gate；
• 取消并发批次时，代码显式取消 worker、冲刷队列中已产生的中断事件，避免 orphan task。

这说明 AgentScope 不是简单地以“只读=并发”推导调度，而是把并发安全性作为工具作者必须声明的契约。

四、Msg 与 AgentEvent：一份状态，两种时间尺度

AgentScope 最值得借鉴的抽象是把“消息”和“事件”明确分开。

4.1 Msg 是可保存的对话中间表示

Msg 包含 name、role、content、metadata、token usage 和最终结束原因。content 不是单一字符串，而是带类型的 block 列表：

这种 block 化有两个直接好处：一是模型适配层能把各家的响应归一；二是工具调用与工具结果可以和回答共存于同一轮 assistant message 中，持久化后仍能完整恢复。

4.2 AgentEvent 是流式观察协议

EventType 覆盖回复开始结束、模型调用边界、文本/思考/数据 delta、工具调用与结果 delta、确认请求、外部执行请求、用户中断等事件。

模型适配器只需要产生统一的 ChatResponse；_convert_chat_response_to_event() 负责把增量 block 拆成 start/delta/end 事件。于是前端可以独立渲染“正在思考”“正在输入参数”“工具返回增量”这些状态，而无需为 OpenAI、Anthropic、DashScope 分别写一套流式协议。

反向路径同样重要：Msg.append_event() 可以把事件重新折叠为完整消息。服务端在单次 ChatService.run() 期间把事件写入短期 replay log，供该次运行中晚到的 SSE 客户端补齐；run 返回时持久化的是当前累积的 reply Msg 与 AgentState，并清理这份事件缓冲。工具因 HITL 挂起时也是如此，后续恢复依赖持久化状态而非旧 replay log。这就是“事件用于过程，消息用于状态”的分工。

五、模型、工具与权限：模型能提议，运行时才有权执行

5.1 ChatModelBase 将厂商差异压到 _call_api()

ChatModelBase 定义统一输入：messages、tools、tool_choice；统一输出：完整 ChatResponse 或异步 ChatResponse 流。

基类已经处理了可重试异常、取消时转换为 FinishedReason.INTERRUPTED、流式 chunk 的累计。具体 OpenAI、Anthropic、DashScope、Gemini、Ollama 等实现只需要承担 API 请求和格式化差异。Agent._call_model() 还在模型之上加入 Agent 级重试和 fallback model，因此“单厂商短暂故障”不会与 Agent 的 ReAct 逻辑耦合。

结构化输出的实现也很务实：当底层 API 不提供原生 JSON Schema 输出时，generate_structured_output() 会构造临时 generate_structured_output 工具并强制 tool_choice，随后解析和校验参数。它复用已有工具调用通道，而非为每个供应商另建一套 schema 协议。

5.2 Toolkit 统一 Python 工具、MCP 与 Skill

Toolkit 是工具的唯一门面：注册本地 ToolBase、MCP client、Skill loader 和工具组。

工具组不是 UI 分类，而是运行时能力边界：默认 basic 组始终存在，其余组必须先激活；未激活工具即使已注册，也会被返回为可供模型理解的错误提示。Skill 也不是一个“可直接调用”的黑盒，它先以名称、描述和目录注入系统提示，再通过 SkillViewer 读取详情。

call_tool() 还统一了 coroutine、同步/异步 generator 三种实现，让工具作者只返回 ToolChunk，框架负责累计为最终 ToolResponse。这使模型侧、工具侧、前端侧都围绕同一流式接口工作。

5.3 一次工具调用经过五道关口

_execute_tool_call() 的顺序非常值得逐项看：

1. 从 Toolkit 确认工具当前可用；
2. 用 JSON repair 解析模型输入，再用 jsonschema 验证；
3. 通过中间件洋葱层和 PermissionEngine 取得决策；
4. 对 ASK 发出 RequireUserConfirmEvent，对外部工具发出 RequireExternalExecutionEvent；
5. 只有 ALLOW 才进入 _acting()，获得 ToolChunk 流并写回 ToolResultBlock。

默认权限模式的优先级在 PermissionEngine._check_default() 中写得很清楚：Deny 规则 → Ask 规则 → 只读快速放行 → 工具自身安全检查 → Allow 规则 → 询问用户。其中安全性 ASK 可以是 bypass-immune，不能被普通 allow rule 覆盖。

这套顺序的重要性在于：默认模式不会因为“模型看起来很确定”而跳过副作用确认；EXPLORE 模式则将非只读操作直接拒绝；ACCEPT_EDITS 才允许工作目录内的编辑由工具自身策略自动放行。

5.4 中间件有七个明确 hook，而不是万能回调

MiddlewareBase 提供六个洋葱式 hook：on_reply、on_reasoning、on_check_permission、on_acting、on_model_call、on_compress_context；另有顺序变换式的 on_system_prompt。

这种切法让 RAG、长记忆、预算、Tracing、TTS、工具卸载等能力能选择合适的介入点。例如 on_acting 只包裹原始工具 I/O，权限和状态写入在其外部完成，因此后台化工具不会偷偷修改 Agent 上下文。代价是扩展者必须理解 hook 边界：在权限 hook 中返回而不调用 next_handler，等价于接管该次权限决策。

六、上下文管理：压缩不是把旧消息删掉

长对话的难点是既要控制 token，又不能破坏工具调用关系。AgentState 把持久化状态拆成：

• summary：已压缩历史的摘要；
• context：未压缩消息；
• reply_context：reply_id、迭代数和结构化输出要求；
• permission_context：模式、规则和工作目录；
• tool_context：激活工具组与文件读取缓存；
• tasks_context：任务状态；
• middle_context：中间件跨轮状态。

源码位置：state/_state.py。

6.1 压缩动作与工作区卸载

当估算 token 超过 trigger_ratio × context_size，_compress_context_impl() 会保留最近一段上下文，把较早消息连同 system prompt、旧 summary 和压缩指令交给模型生成新的结构化摘要。

若配置了 Offloader，被压缩的原文还会写入工作区；摘要追加“内容位于某路径”的提醒。工具结果过大时也采用相同策略：上下文保留截断部分和指针，完整结果卸载。这是一个比“丢掉旧上下文”更实用的折中：模型的活跃工作集变小，但必要时仍可通过工具回看原文。

6.2 文件读取缓存与上下文一致性

ToolContext 内置基于 mtime 的文件读取缓存，并对文件数量和总字节数做 LRU 淘汰。上下文压缩后，clean_file_cache() 还会根据仍保留的 Read 调用清理无关缓存。这种细节说明状态设计并不只关心聊天文本，也考虑了 coding agent 高频文件读取的成本与过期问题。

七、从 SDK 到生产服务：事件总线是连接件

create_app() 注册 agent、chat、session、workspace、model、credential、RAG、schedule 等路由，但真正把一次聊天跑起来的是 ChatService.run()。

其组装过程可以压缩为：读取 Agent/Session/Workspace → 补工作目录权限 → 注入 Inbox、状态变更、工具卸载等中间件 → 按 session 追加 RAG/TTS → 组装 Toolkit → 解析模型与 fallback → 恢复 AgentState 并运行 Agent。

7.1 为什么 POST 不直接返回流

POST /chat/ 只启动任务并立即返回。它不会在 HTTP 响应里承载模型流；持续事件来自 GET /sessions/{session_id}/stream。

看似多了一跳，实则带来生产上的好处：

• 使用 Redis 等共享 MessageBus 后端时，同一 session 的运行由 acquire_lock(MessageBusKeys.session_lock(session_id), ...) 串行化，避免多个进程同时修改一份状态；
• 每个事件同时写入本次运行的短期 replay log 和 live Pub/Sub；客户端在运行期间晚连接时，可以先读回放，再订阅实时通道；
• 前端可以保持一条长期 SSE 连接，连续接收同一 session 后续的多轮执行；
• HITL 的确认结果通过 wakeup 队列恢复，避免刚挂起的旧 run 与新 run 抢占 session。

stream_session_events() 还每 30 秒发送 SSE 心跳，并使用后台 feeder + queue，而不是直接取消异步生成器的 __anext__()，规避了取消后生成器无法关闭的常见异步陷阱。

7.2 资源隔离默认是拒绝跨用户访问

服务层的 create_app() 默认注入 DenyAllResourceAccessPolicy。这意味着跨 owner 访问 credential、agent、knowledge base 必须显式授权；ChatService 在装配时通过 ResourceAccessService 解析这些可见资源。工作区及其派生工具则按 session/owner 由 WorkspaceManager 管理；接入方注入的自定义工具仍须自行实现认证、授权与租户隔离。对多租户 Agent 平台来说，这比只在路由层做一次 owner 判断更可靠。

八、多智能体：团队协作被降解为可观察的会话通信

AgentScope 的团队系统不另起一套“多 Agent 引擎”。成员依旧是独立 Agent 与独立 session；团队能力通过几个工具和消息总线组合出来：

8.1 创建 worker 时复制什么，不复制什么

AgentCreate 先确认调用者是当前团队 Leader，再选择 SubAgentTemplate。模板携带 system prompt、context/ReAct 配置、权限上下文和任务上下文；Leader 已获得的权限规则和工作目录可按模板策略合并给 worker。

这是一种“模板定义角色，Leader 继承有限授权”的模式。Worker 不是 Leader 的内存副本：它有自己的 AgentState 和 session；但它也不是完全裸奔的新 Agent：团队目标、角色描述、模型配置和必要权限会被明确装配。

8.2 TeamSay 如何让另一个 Agent 醒来

TeamSay 先通过存储构造成员名到 (session_id, agent_id) 的目录，再把内容包装成：

<team-message from="leader">
  请汇报调研结论和证据。
</team-message>

这个 HintBlock 被 queue_push 到收件人 inbox，随后 enqueue_run_trigger 唤醒目标 session。InboxMiddleware 在下一轮将收件箱内容带入 Agent 上下文。

因此，多 Agent 通信不是进程内函数调用，也不是把一段文本塞到共享 memory，而是：由 MessageBus 后端承载的 inbox 队列 + 可调度的会话唤醒 + 统一事件流。其默认说明文案鼓励 Leader 分派、成员完成后回报，避免无节制的 peer-to-peer chatter；但底层 TeamSay 仍支持定向发送和广播。

8.3 子 Agent 的 HITL 如何投影给 Leader

团队中更棘手的情况是：worker 触发了用户确认，但用户前端只盯着 Leader session。服务层的 SubagentHitlProjector 将待确认卡片投影到 Leader 会话；提交确认时，/chat/ 会根据 reply_id 解析真正的 worker session，再把恢复事件入队。这样用户看到的是一个团队统一入口，后端恢复的却是正确的子会话状态机。

这是 AgentScope 团队设计最成熟的部分：协作不只是在“能发消息”，还包括确认请求、会话归属和前端可见性的闭环。

九、二次开发：从哪个扩展点切入

有两条实践建议尤其重要：

1. 工具必须把安全语义做成代码。定义 is_concurrency_safe、is_read_only、输入 Schema、权限建议和 match_rule()，不要只依赖 prompt 约束。
2. 不要绕开 AgentState 写临时全局状态。暂停、恢复、服务重启、跨进程调度和团队消息之所以成立，前提是运行事实能落到 session state、存储或消息总线中。

十、设计取舍与局限

AgentScope 的设计并非没有代价。

如果你的需求只是单轮问答，直接用模型 SDK 会更轻。如果你要的是固定、可验证的 DAG，显式工作流引擎可能更直观。但当需求包含工具副作用、人工确认、会话恢复、流式前端、沙箱、RAG、后台任务和团队协作时，AgentScope 的复杂度不是“过度设计”，而是把原本会散落在业务层的工程问题提前收敛到运行时。

十一、总结：用状态和事件驯服 Agent 的不确定性

AgentScope 2.0 的源码主线可以浓缩为：

模型负责提出下一步；AgentState 记录已经发生了什么；AgentEvent 把过程交给外部；Toolkit、权限和工作区决定副作用能否发生；服务层则为多会话、多 Agent，以及共享基础设施下的多进程运行提供一致性机制。

从这一视角再看 Agent.reply_stream()，它不是一个“流式聊天 API”，而是整个系统的最小执行接口：新消息、用户确认、外部结果和中断都被投递到同一状态机；无论 SDK、HTTP、SSE 还是团队工具，最终都在推动这台状态机前进。

这也是构建生产级 Agent 时最值得复用的思路：不要只设计 prompt 和工具列表；先设计 状态怎样保存、事件怎样传播、权限怎样拦截、暂停后怎样恢复。当这四件事成立，模型能力变化才不会让系统的控制面失控。

参考源码

• agent/_agent.py
• state/_state.py
• message/_base.py
• event/_event.py
• tool/_toolkit.py
• permission/_engine.py
• app/_service/_chat.py
• app/_router/_session.py
• app/_tool/

× 爱折腾的工程师

真诚赞赏，手留余香

2元 5元 10元 50元 100元 任意金额

2元

使用微信扫描二维码完成支付

• ← Previous Post
• Next Post →