Table of contents

Harness：Loop 外围的运行系统

Agent Loop 只定义一次推理与行动如何闭合。Harness 负责让这个循环在真实环境中持续运行：准备输入、装配上下文、暴露能力、约束执行、保存状态，并在失败后恢复。

用户 / Cron / Inbox / 外部 Channel
  → 输入规范化与运行时指令
  → Context / Memory / Skill 装配
  → Agent Loop
      → Model
      → Tool Call
      → Permission / Sandbox
      → Tool Result
  → Event / Artifact / Message
  → State / Task / Schedule 持久化

模型决定下一步做什么；Harness 决定模型能看到什么、能调用什么、调用在哪里执行，以及失败、暂停或重启后从哪里继续。

通用 Agent 架构地图将这些能力分成核心循环、工具、安全、上下文、韧性、规划协作、触发调度和外部接入。后面的章节按这条依赖链逐层展开。

本文以 learn-claude-code 的递进代码为主线。AgentScope 和 DeerFlow 只在对应机制后补充框架接口、失败路径与运行约束，不构成另一条并列教程。

Agent Loop：最小行动闭环

先看最小实现。learn-claude-code 的 S01只有一个 Bash 工具和一个循环。

最小实现

源码：s01_agent_loop/code.py

# ── The core pattern: a while loop that calls tools until the model stops ──
def agent_loop(messages: list):
    while True:
        response = client.messages.create(
            model=MODEL, system=SYSTEM, messages=messages,
            tools=TOOLS, max_tokens=8000,
        )

        # Append assistant turn
        messages.append({"role": "assistant", "content": response.content})

        # If the model didn't call a tool, we're done
        if response.stop_reason != "tool_use":
            return

        # Execute each tool call, collect results
        results = []
        for block in response.content:
            if block.type == "tool_use":
                print(f"\033[33m$ {block.input['command']}\033[0m")
                output = run_bash(block.input["command"])
                print(output[:200])
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                })

        # Feed tool results back, loop continues
        messages.append({"role": "user", "content": results})

messages 保存 Loop 状态。tool_use_id 将结果与调用配对；裁剪上下文时，tool_use 与 tool_result 必须一起保留。

工程化 Loop：状态、并发与暂停

AgentScope 的循环仍然是“推理 → 工具 → 结果 → 再推理”，但下一步不只由 stop_reason 决定，而是由 AgentState 计算。

源码：agent/_agent.py

await self._handle_incoming_messages(msgs)
self.state.reply_id = _generate_id()
self.state.cur_iter = 0

yield ReplyStartEvent(
    session_id=self.state.session_id,
    reply_id=self.state.reply_id,
    name=self.name,
)

while self.state.cur_iter < self.react_config.max_iters:
    action, data = self._check_next_action()

    if action == "exit" and isinstance(data, Msg):
        yield data
        return

    if action == "reasoning":
        await self.compress_context()

        async for evt in self._reasoning():
            if isinstance(evt, Msg):
                yield evt
                return
            yield evt

    for batch in await self._batch_tool_calls():
        if batch.type == "sequential":
            evt_generator = (
                self._execute_sequential_tool_calls(
                    batch.tool_calls
                )
            )
        elif batch.type == "concurrent":
            evt_generator = (
                self._execute_concurrent_tool_calls(
                    batch.tool_calls
                )
            )

_check_next_action() 决定继续 reasoning、执行尚未完成的工具，还是退出。工具按并发安全性分批；执行过程持续产生 Event。

收到 RequireUserConfirmEvent 或 RequireExternalExecutionEvent 时，Loop 会保存等待状态并暂停。恢复时传入 confirmation/external result，继续同一个 session，而不是重新创建一轮无状态调用。

S01 展示 Loop 的最小协议；AgentScope 展示同一个协议如何加入状态机、流式事件、并发、压缩与 HITL。

当 Loop 需要被终端 UI、Web 前端或服务端观察时，内部控制流就要变成 Message 与 Event。

Message、ContentBlock、OpenAI Chat messages 与 Event

S01 直接把供应商消息写进 messages。AgentScope 先定义自己的消息模型：Msg 是消息外壳，content 是一组有序的 ContentBlock。调用模型前，Formatter 再把它们转换成 OpenAI Chat messages；模型运行时产生的 Event 则用于增量构造 Msg。

Message 与 ContentBlock

Msg 在内容之外保存角色、发送者、时间、metadata 和 token usage。源码：_base.py

class Msg(BaseModel):
    name: str
    content: list[ContentBlock]
    role: Literal["user", "assistant", "system"]
    id: str = Field(default_factory=_generate_id)
    metadata: dict = Field(default_factory=dict)
    created_at: str = Field(
        default_factory=lambda: datetime.now().isoformat()
    )
    finished_at: str | None = Field(default=None)
    usage: Usage | None = Field(default=None)

SystemMsg 只允许 TextBlock；UserMsg 允许 TextBlock 和 DataBlock；AssistantMsg 可以包含全部 ContentBlock。ContentBlock 的类型定义如下。源码：_block.py

ContentBlock: TypeAlias = (
    TextBlock
    | ThinkingBlock
    | HintBlock
    | ToolCallBlock
    | ToolResultBlock
    | DataBlock
)

TextBlock 保存文本，ThinkingBlock 保存模型思考，DataBlock 保存图片和音频等数据，HintBlock 承载 Inbox、后台结果和团队消息，ToolCallBlock 与 ToolResultBlock 保存工具调用及其结果。

因此，ContentBlock 不是与 Message 并列的另一种消息。它是 Msg.content 的元素：

Msg
├── role / name / id / metadata / usage
└── content: list[ContentBlock]
    ├── TextBlock
    ├── ThinkingBlock
    ├── DataBlock
    ├── HintBlock
    ├── ToolCallBlock
    └── ToolResultBlock

转换为 OpenAI Chat messages

OpenAI Chat message 主要有以下五种格式：

System message
{"role": "system", "content": "..."}

User message
{"role": "user", "content": "..."}

Assistant text message
{"role": "assistant", "content": "..."}

Assistant tool-call message
{
  "role": "assistant",
  "content": null,
  "tool_calls": [{
    "id": "call-1",
    "type": "function",
    "function": {
      "name": "get_weather",
      "arguments": "{\"city\":\"上海\"}"
    }
  }]
}

Tool result message
{"role": "tool", "tool_call_id": "call-1", "name": "get_weather", "content": "..."}

tool_calls[].id 与 tool_call_id 使用同一个值，将调用和结果配对。

上面用简化 JSON 展示五种消息形态。AgentScope 实际调用时，OpenAIChatFormatter 根据 Msg.content 生成对应格式。

源码：_openai_formatter.py

for block in msg.get_content_blocks():
    if isinstance(block, TextBlock):
        content_blocks.append({
            "type": "text",
            "text": block.text,
        })

    elif isinstance(block, ToolCallBlock):
        tool_calls.append({
            "id": block.id,
            "type": "function",
            "function": {
                "name": block.name,
                "arguments": block.input,
            },
        })

    elif isinstance(block, ToolResultBlock):
        messages.append({
            "role": "tool",
            "tool_call_id": block.id,
            "content": textual_output,
            "name": block.name,
        })

msg_openai = {
    "role": msg.role,
    "name": msg.name,
    "content": content_blocks or None,
}
if tool_calls:
    msg_openai["tool_calls"] = tool_calls

TextBlock 转成文本 content，DataBlock 转成图片或音频输入，HintBlock 转成 user message，ToolCallBlock 转成 assistant.tool_calls，ToolResultBlock 转成 tool message；ThinkingBlock 不写入 OpenAI Chat 历史。

这不是一一对应的拆解。普通 Msg 通常生成一条 OpenAI message；包含 HintBlock 或 ToolResultBlock 的 AssistantMsg 可能展开成额外的 user 或 tool message。Context 保存的是 Msg + ContentBlock，OpenAI Chat messages 只存在于模型调用边界。

Event

消息是完成后的状态，事件是生成状态的增量。AgentScope 的事件类型按生命周期和内容块组织。下面只列出一次 ReAct reply 主路径中的事件。

源码：_event.py

class EventType(StrEnum):
    REPLY_START = "REPLY_START"
    REPLY_END = "REPLY_END"

    MODEL_CALL_START = "MODEL_CALL_START"
    MODEL_CALL_END = "MODEL_CALL_END"

    TEXT_BLOCK_START = "TEXT_BLOCK_START"
    TEXT_BLOCK_DELTA = "TEXT_BLOCK_DELTA"
    TEXT_BLOCK_END = "TEXT_BLOCK_END"

    TOOL_CALL_START = "TOOL_CALL_START"
    TOOL_CALL_DELTA = "TOOL_CALL_DELTA"
    TOOL_CALL_END = "TOOL_CALL_END"

    TOOL_RESULT_START = "TOOL_RESULT_START"
    TOOL_RESULT_TEXT_DELTA = "TOOL_RESULT_TEXT_DELTA"
    TOOL_RESULT_END = "TOOL_RESULT_END"

三个 ID 负责关联事件：

• reply_id：同一次 reply。
• block_id：同一个文本、思考或数据块。
• tool_call_id：同一次工具调用及其结果。

从 Event 重建 Message

reply_stream() 是外部消费事件的入口。收到 ReplyStartEvent 时创建空消息，其余事件交给 append_event()：

from agentscope.event import ReplyStartEvent
from agentscope.message import AssistantMsg

msg = None

async for event in agent.reply_stream(user_msg):
    if isinstance(event, ReplyStartEvent):
        msg = AssistantMsg(
            name=event.name,
            content=[],
            id=event.reply_id,
        )
    elif msg is not None:
        msg.append_event(event)

append_event() 根据 block_id 或 tool_call_id 找到内容块，再追加 delta 或更新最终状态。

源码：_base.py

一次 ReAct Reply

一次 reply 可能包含多次模型调用：

ReplyStartEvent
  ModelCallStartEvent
    TextBlock*
    ThinkingBlock*
    ToolCallStartEvent
    ToolCallDeltaEvent
    ToolCallEndEvent
  ModelCallEndEvent

  ToolResultStartEvent
  ToolResultTextDeltaEvent*
  ToolResultEndEvent

  ModelCallStartEvent
    TextBlock*
  ModelCallEndEvent
ReplyEndEvent

当同一批工具并发执行时，不同工具的结果事件可能交错，消费者需要依靠 tool_call_id 分流。

Tools、MCP 与 Skills：三种不同的能力扩展

三者都能扩展 Agent，但入口不同：

Tool：Schema 与 Handler

S02 把 S01 中写死的 Bash 调用改成工具表。

源码：s02_tool_use/code.py

TOOLS = [
    {"name": "read_file", "description": "Read file contents.",
     "input_schema": {"type": "object",
                      "properties": {
                          "path": {"type": "string"},
                          "limit": {"type": "integer"}},
                      "required": ["path"]}},
]

TOOL_HANDLERS = {
    "bash": run_bash,
    "read_file": run_read,
    "write_file": run_write,
    "edit_file": run_edit,
    "glob": run_glob,
}

TOOLS 发送给模型，描述有哪些动作以及参数格式。TOOL_HANDLERS 留在 Harness 内，将工具名映射到 Python 函数。

Loop 的变化只有分发这一行：

handler = TOOL_HANDLERS.get(block.name)
output = (
    handler(**block.input)
    if handler
    else f"Unknown: {block.name}"
)

CoreCoder 使用类表达同一个接口：name、description、JSON Schema 和 execute()。写法不同，边界相同：corecoder/tools/base.py。

并发属于运行时

一次模型响应可以包含多个 tool_use，但能否并发不能只看调用数量。读文件通常可以并发，写文件和 Bash 可能产生冲突。

AgentScope 使用 is_concurrency_safe 切分批次：

源码：agent/_agent.py

for tool_call in tool_calls:
    tool = await self.toolkit.get_tool(tool_call.name)

    if tool is None or tool.is_concurrency_safe:
        batch_type = "concurrent"
    else:
        batch_type = "sequential"

同一并发批次的结果事件可能交错，但每个结果仍通过 tool_call_id 回到对应调用。MCPTool 默认不声明并发安全，因此按顺序执行。

Skill：按需加载知识

Skill 不增加新的执行环境。它把操作规范、工作流和领域知识放进 SKILL.md，需要时再加载。

S07 启动时只把名称和描述放进 system prompt，完整内容保存在注册表中：

源码：s07_skill_loading/code.py

def list_skills() -> str:
    return "\n".join(
        f"- **{s['name']}**: {s['description']}"
        for s in SKILL_REGISTRY.values()
    )

def build_system() -> str:
    catalog = list_skills()
    return (
        f"Skills available:\n{catalog}\n"
        "Use load_skill to get full details when needed."
    )

加载 Skill 仍然走 Tool dispatch：

源码：s07_skill_loading/code.py

def load_skill(name: str) -> str:
    skill = SKILL_REGISTRY.get(name)
    if not skill:
        return f"Skill not found: {name}"
    return skill["content"]

TOOL_HANDLERS = {
    # ...
    "load_skill": load_skill,
}

Skill 内容以 tool_result 进入当前 messages。目录负责发现，load_skill 负责按需展开；加载后的内容仍会占用上下文，并可能在后续压缩中被裁剪。

MCP：发现外部工具

MCP 不是知识文件，而是 Agent 与外部服务之间的工具协议。Server 提供工具定义和调用入口，Client 负责发现与调用。

S19 的教学实现用本地 handler 模拟 MCP Server，但工具池组装逻辑已经完整：

源码：s19_mcp_plugin/code.py

def assemble_tool_pool() -> tuple[list[dict], dict]:
    tools = list(BUILTIN_TOOLS)
    handlers = dict(BUILTIN_HANDLERS)

    for server_name, mcp_client in mcp_clients.items():
        safe_server = normalize_mcp_name(server_name)

        for tool_def in mcp_client.tools:
            safe_tool = normalize_mcp_name(tool_def["name"])
            prefixed = f"mcp__{safe_server}__{safe_tool}"

            tools.append({
                "name": prefixed,
                "description": tool_def.get("description", ""),
                "input_schema": tool_def.get("inputSchema", {}),
            })
            handlers[prefixed] = (
                lambda *, c=mcp_client,
                t=tool_def["name"], **kw:
                c.call_tool(t, kw)
            )

    return tools, handlers

mcp__server__tool 既避免重名，也保留工具来源。连接新的 Server 后，工具池和发给模型的 schema 都要重新生成。

真实 MCP 还要处理 stdio、HTTP、鉴权、断线和超时。S19 使用 mock，是为了只展示“发现 → 命名 → 组装 → 调用”这条主路径。

统一工具入口：Toolkit

官方文档：Tool

AgentScope 的 Toolkit 同时接收本地 Tool、MCP Client 和 Skill Loader。三者进入同一个 tool group，但保留各自的生命周期。

源码：tool/_toolkit.py

toolkit = Toolkit(
    tools=[
        Bash(),
        Read(),
        Write(),
        Edit(),
    ],
    mcps=[
        filesystem_mcp,
        search_mcp,
    ],
    skills_or_loaders=[
        "/path/to/skills",
    ],
)

本地 Tool 实现 ToolBase，除 schema 与 call() 外，还声明 is_concurrency_safe、is_read_only、is_state_injected 等运行时属性。

MCP Client 负责连接、发现和关闭。工具统一命名为 mcp__server__tool；Server 的 readOnlyHint 还会进入权限判断。

Skill 不直接变成业务工具。Toolkit 自动注册 Skill viewer，只把名称和描述加入 System Prompt；模型调用 viewer 后，完整 SKILL.md 才以工具结果进入 Context。

Toolkit 还支持 Tool Group。非基础组可由模型按需激活，工具 schema、MCP 和 Skill 指令会一起进入或退出当前上下文。

S02/S07/S19 分三步实现同一套边界；AgentScope 把注册、schema、dispatch、权限属性与按组激活集中在 Toolkit。

Schema 只校验参数结构；路径和命令权限由运行时检查。

Workspace、Sandbox 与 Permission

四个概念分别解决不同问题：

路径边界

S02 将当前目录作为 WORKDIR，所有文件路径先经过 safe_path()：

源码：s02_tool_use/code.py

def safe_path(p: str) -> Path:
    path = (WORKDIR / p).resolve()
    if not path.is_relative_to(WORKDIR):
        raise ValueError(f"Path escapes workspace: {p}")
    return path

resolve() 消除 ../ 并解析符号链接，is_relative_to() 再检查最终路径是否仍在工作区内。这个检查限制文件工具，但不会隔离 Bash 进程。

Workspace 写进 system prompt 也不等于边界。CoreCoder 的文件工具会调用 Path.resolve()，但没有检查结果必须位于工作目录内，因此绝对路径和 ../ 仍可能访问外部文件。

Workspace 的隔离粒度

同一个 Workspace 可以属于一次会话、一个 Agent 或一个用户。AgentScope 将隔离粒度映射为稳定的 workspace_id：

源码：workspace_manager/_base.py

class IsolationPolicy(StrEnum):
    PER_SESSION = "per_session"
    PER_AGENT = "per_agent"
    PER_USER = "per_user"

if self._isolation is IsolationPolicy.PER_AGENT:
    return hashlib.blake2b(
        f"{user_id}::{agent_id}".encode("utf-8"),
        digest_size=8,
    ).hexdigest()

if self._isolation is IsolationPolicy.PER_USER:
    return hashlib.blake2b(
        f"user::{user_id}".encode("utf-8"),
        digest_size=8,
    ).hexdigest()

return _generate_id()

IsolationPolicy 决定哪些 session 共享同一个 workspace_id，不提供进程或文件系统安全。实际边界由 Local、Docker、E2B 等 Workspace backend 决定。

Permission：执行前的三道门

S03 在 Tool dispatch 前插入 deny list、规则匹配和用户确认。

源码：s03_permission/code.py

def check_permission(block) -> bool:
    if block.name == "bash":
        reason = check_deny_list(
            block.input.get("command", "")
        )
        if reason:
            return False

    reason = check_rules(block.name, block.input)
    if reason:
        decision = ask_user(
            block.name, block.input, reason
        )
        if decision == "deny":
            return False

    return True

Loop 只多一个检查点。拒绝执行后仍要返回 tool_result，保证调用与结果成对：

if not check_permission(block):
    results.append({
        "type": "tool_result",
        "tool_use_id": block.id,
        "content": "Permission denied.",
    })
    continue

deny list 只能挡住已知字符串，不能作为完整安全边界。编码、变量展开或另一种命令写法都可能绕过它。路径校验、结构化规则、确认流程和执行沙箱需要同时存在。

权限状态、规则与工具自检

官方文档：Permission System

PermissionContext 保存当前 mode、工作目录以及 allow/deny/ask rules：

源码：permission/_context.py

class PermissionContext(BaseModel):
    mode: PermissionMode = PermissionMode.DEFAULT

    working_directories: dict[
        str,
        AdditionalWorkingDirectory,
    ] = Field(default_factory=dict)

    allow_rules: dict[
        str, list[PermissionRule]
    ] = Field(default_factory=dict)

    deny_rules: dict[
        str, list[PermissionRule]
    ] = Field(default_factory=dict)

    ask_rules: dict[
        str, list[PermissionRule]
    ] = Field(default_factory=dict)

Permission Engine 先按 mode 选择策略。DEFAULT 的顺序是 deny rule、ask rule、工具自己的 check_permissions()、allow rule，最后默认 ASK。

EXPLORE 只允许 check_read_only() 为真的调用；ACCEPT_EDITS 还允许工作目录内的修改；DONT_ASK 将 ASK 转成 DENY。

源码：permission/_engine.py

mode = self.context.mode

if mode == PermissionMode.DEFAULT:
    return await self._check_default(
        tool,
        tool_input,
    )
if mode == PermissionMode.EXPLORE:
    return await self._check_explore(
        tool,
        tool_input,
    )
if mode == PermissionMode.ACCEPT_EDITS:
    return await self._check_accept_edits(
        tool,
        tool_input,
    )
if mode == PermissionMode.DONT_ASK:
    return await self._check_dont_ask(
        tool,
        tool_input,
    )

ASK 会产生 RequireUserConfirmEvent 并暂停 reply；UserConfirmResultEvent 使用同一个 reply_id 恢复执行。DENY 则生成配对的 denied tool result。

Sandbox：隔离执行环境

Sandbox 将 Bash、文件和进程操作放进受控环境。DeerFlow 2.0 的完整实现把生命周期与具体后端分开：

源码：sandbox_provider.py

class SandboxProvider(ABC):
    @abstractmethod
    def acquire(
        self,
        thread_id: str | None = None,
        *,
        user_id: str | None = None,
    ) -> str:
        pass

    @abstractmethod
    def get(self, sandbox_id: str) -> Sandbox | None:
        pass

    @abstractmethod
    def release(self, sandbox_id: str) -> None:
        pass

Provider 管理 Sandbox 的创建、复用与释放；Sandbox 提供 execute_command()、read_file()、write_file() 等统一操作。DeerFlow 可以使用本地、AIO 容器或 E2B 后端。

本地 Provider 主要提供线程目录映射，不是安全的 Shell 隔离边界，因此 DeerFlow 默认关闭 host Bash。AIO 容器或 E2B 才适合执行不受信任的命令。

可替换的 Workspace Backend

官方文档：Workspace

AgentScope 的 Workspace 同时提供执行 backend、工具、Skill 目录和 Context offloader。工具从 Workspace 取得 backend，压缩内容也写回同一环境。

workspace = DockerWorkspace()
await workspace.initialize()

agent = Agent(
    name="coder",
    system_prompt="...",
    model=model,
    toolkit=Toolkit(
        tools=await workspace.list_tools(),
    ),
    offloader=workspace,
)

LocalWorkspace 直接使用宿主文件系统与进程，不提供安全沙箱。DockerWorkspace、E2BWorkspace、OpenSandboxWorkspace 和 K8sWorkspace 将工具执行放入独立环境。

IsolationPolicy 控制 Workspace 的共享粒度；Workspace 类型决定执行边界。两者是独立配置。

Worktree：隔离修改，不隔离进程

并行任务即使使用同一个 Sandbox，也可能同时修改同一份代码。S18 为任务创建独立分支与工作树：

源码：s18_worktree_isolation/code.py

def create_worktree(name: str, task_id: str = "") -> str:
    path = WORKTREES_DIR / name
    ok, result = run_git([
        "worktree", "add", str(path),
        "-b", f"wt/{name}", "HEAD",
    ])
    if not ok:
        return f"Git error: {result}"

    if task_id:
        bind_task_to_worktree(task_id, name)

    return f"Worktree '{name}' created at {path}"

Worktree 隔离分支与文件修改，但共享宿主进程、网络和系统资源。它解决协作冲突，不解决恶意命令。

Hooks 与 Middleware

权限、日志、限流和追踪都会穿过 Agent Loop。把这些逻辑写进循环，核心代码会不断膨胀。

Hooks：离散扩展点

S04 用事件名和回调列表建立 Hook 注册表：

源码：s04_hooks/code.py

HOOKS = {
    "UserPromptSubmit": [],
    "PreToolUse": [],
    "PostToolUse": [],
    "Stop": [],
}

def register_hook(event: str, callback):
    HOOKS[event].append(callback)

def trigger_hooks(event: str, *args):
    for callback in HOOKS[event]:
        result = callback(*args)
        if result is not None:
            return result
    return None

S03 的权限判断从 Loop 中移到 PreToolUse：

register_hook("PreToolUse", permission_hook)
register_hook("PreToolUse", log_hook)
register_hook("PostToolUse", large_output_hook)
register_hook("Stop", summary_hook)

Loop 只保留触发点：

源码：s04_hooks/code.py

blocked = trigger_hooks("PreToolUse", block)
if blocked:
    results.append({
        "type": "tool_result",
        "tool_use_id": block.id,
        "content": str(blocked),
    })
    continue

output = TOOL_HANDLERS[block.name](**block.input)
trigger_hooks("PostToolUse", block, output)

Hook 适合“在某个点做一件事”。返回值可以拦截工具，Stop Hook 也可以注入新消息并让 Loop 继续。

Middleware：包裹一段生命周期

AgentScope 的 Middleware 不只监听点位，还能包住整个异步调用：

源码：middleware/_base.py

class MiddlewareBase:
    async def on_reply(...): ...
    async def on_reasoning(...): ...
    async def on_acting(...): ...
    async def on_model_call(...): ...
    async def on_compress_context(...): ...
    async def on_system_prompt(...): ...

前五个是洋葱链：

middleware_1 before
  middleware_2 before
    original handler
  middleware_2 after
middleware_1 after

on_system_prompt 是串行 transformer：前一个中间件的输出会成为后一个中间件的输入。

on_acting 只包住实际的 toolkit.call_tool。权限检查、参数验证和 context 写入在它外面，因此后台卸载工具时不会顺带并发修改 Agent Context。

工业实现：Middleware 顺序也是协议

DeerFlow 的运行时链只取四个代表机制：输入清洗、工具结果清洗、Sandbox 审计和工具错误处理。它们分别守住输入、回灌内容、执行前检查和异常返回。

源码：tool_error_handling_middleware.py

outer_wrappers: list[AgentMiddleware] = [
    InputSanitizationMiddleware(),
    # ToolOutputBudgetMiddleware is between the two in DeerFlow.
    ToolResultSanitizationMiddleware(),
]

# ...

tail.append(SandboxAuditMiddleware())

# ...

tail.append(
    ToolErrorHandlingMiddleware(app_config=app_config)
)

middlewares = [
    *outer_wrappers,
    *thread_hooks,
    *tail,
]

列表中的前项是外层。清洗必须先于内层处理，审计必须包住实际执行，错误处理必须保证每次工具调用都有可回灌的结果。这里的顺序属于运行协议，不是代码排版。

System Prompt 与运行时指令

System Prompt 不只描述“你是谁”。工具列表、工作目录、记忆索引和当前运行状态都会影响它的内容。

分段组装

S10 将 Prompt 拆成稳定片段和动态片段：

源码：s10_system_prompt/code.py

PROMPT_SECTIONS = {
    "identity": "You are a coding agent. Act, don't explain.",
    "tools": "Available tools: bash, read_file, write_file.",
    "workspace": f"Working directory: {WORKDIR}",
    "memory": "Relevant memories are injected below when available.",
}

def assemble_system_prompt(context: dict) -> str:
    sections = [
        PROMPT_SECTIONS["identity"],
        PROMPT_SECTIONS["tools"],
        PROMPT_SECTIONS["workspace"],
    ]

    memories = context.get("memories", "")
    if memories:
        sections.append(f"Relevant memories:\n{memories}")

    return "\n\n".join(sections)

identity、tools 和 workspace 每轮都需要；memory 只有存在内容时才加入。Skill Catalog、权限模式或团队身份也可以用相同方式成为条件片段。

根据状态重新计算

工具执行后，文件、任务和记忆都可能变化。S10 每轮更新 context，再重新取得 Prompt：

源码：s10_system_prompt/code.py

system = get_system_prompt(context)

while True:
    response = client.messages.create(
        model=MODEL,
        system=system,
        messages=messages,
        tools=TOOLS,
        max_tokens=8000,
    )

    # tool execution ...

    context = update_context(context, messages)
    system = get_system_prompt(context)

get_system_prompt() 使用确定性序列化后的 context 作为缓存键。状态不变时复用字符串，状态变化时重新组装。

运行时 Prompt 装配

AgentScope 每次准备模型输入时重新取得 System Prompt。基础文本之后依次加入已激活 Skill 的指令、Workspace/offloader 指令，再经过 Middleware transformer 链。

源码：agent/_agent.py

async def _get_system_prompt(self) -> str:
    prompt = [self._system_prompt]

    skill_instructions = (
        await self.toolkit.get_skill_instructions(
            self.state.tool_context.activated_groups,
        )
    )
    if skill_instructions:
        prompt.append(skill_instructions)

    if isinstance(self.offloader, WorkspaceBase):
        instructions = (
            await self.offloader.get_instructions()
        )
        if instructions:
            prompt.append(instructions)

    result = "\n".join(prompt)

    for mw in self._system_prompt_middlewares:
        result = await mw.on_system_prompt(self, result)

    return result

模型输入再按固定顺序组装：System Prompt、压缩摘要、当前 Context、工具 schema。

源码：agent/_agent.py

messages = [
    SystemMsg(
        name="system",
        content=await self._get_system_prompt(),
    ),
]

if self.state.summary:
    messages.append(
        UserMsg(name="user", content=self.state.summary)
    )

messages.extend(self.state.context)

tools = await self.toolkit.get_tool_schemas(
    self.state.tool_context.activated_groups,
)

S10 由一个 context dict 和缓存函数完成组装；AgentScope 将动态来源分配给 Toolkit、Workspace 和 Middleware。两者都保持基础 Prompt 稳定，只在运行时附加当前状态。

System Prompt 决定下一轮模型看到哪些规则；messages 保存已经发生的执行历史。

Context Compact：按成本分层缩短历史

上下文满了不应立刻总结整段对话。S08 先做不调用模型的局部处理，仍超限才生成摘要。

四层压缩

源码：s08_context_compact/code.py

while True:
    messages[:] = tool_result_budget(messages)
    messages[:] = snip_compact(messages)
    messages[:] = micro_compact(messages)

    if estimate_size(messages) > CONTEXT_LIMIT:
        messages[:] = compact_history(messages)

    try:
        response = client.messages.create(
            model=MODEL, system=SYSTEM,
            messages=messages, tools=TOOLS,
            max_tokens=8000,
        )
    except Exception as e:
        if "prompt_too_long" in str(e).lower():
            messages[:] = reactive_compact(messages)
            continue
        raise

执行顺序是：

1. tool_result_budget：最新一轮工具结果过大时，将完整内容落盘，只保留路径和预览。
2. snip_compact：消息过多时删除中段，保留开头与最近历史。
3. micro_compact：把较早的长工具结果替换为占位符。
4. compact_history：仍超限时保存 transcript，再用模型把历史压成摘要。

前三层不调用模型，成本低，也保留了重新读取工具输出的路径。第四层会丢失细节，因此最后执行。

保持调用与结果成对

裁剪边界不能落在 tool_use 与 tool_result 之间：

源码：s08_context_compact/code.py

if head_end > 0 and _message_has_tool_use(messages[head_end - 1]):
    while (
        head_end < len(messages)
        and _is_tool_result_message(messages[head_end])
    ):
        head_end += 1

if (
    tail_start > 0
    and _is_tool_result_message(messages[tail_start])
    and _message_has_tool_use(messages[tail_start - 1])
):
    tail_start -= 1

否则模型会看到没有结果的调用，或没有来源的结果。压缩改变的是历史表示，不能破坏消息协议。

CoreCoder 采用另一种分层策略：50% 截断长工具输出，70% 总结旧历史，90% 强制折叠。实现更短，但 token 只是 len(text) // 3 的估算：corecoder/context.py。

结构化摘要与尾部保留

AgentScope 在每次 reasoning 前调用模型的 count_tokens()。默认达到上下文窗口的 80% 时压缩，并保留约 10% 的最近消息。

源码：agent/_agent.py

cfg = context_config or self.context_config

kwargs = await self._prepare_model_input()
estimated_tokens = await self.model.count_tokens(**kwargs)

threshold = cfg.trigger_ratio * self.model.context_size
if estimated_tokens < threshold:
    return

msgs_to_compress, msgs_to_reserve = (
    await self._split_context_for_compression(
        cfg.reserve_ratio * self.model.context_size,
        kwargs.get("tools", []),
    )
)

旧消息不会压成自由文本，而是按 SummarySchema 生成 task overview、current state、discoveries、next steps 和 context to preserve。

压缩完成后，摘要与未压缩尾部分开保存：

源码：agent/_agent.py

new_summary = cfg.summary_template.format(**res.content)

if self.offloader:
    path = await self.offloader.offload_context(
        self.state.session_id,
        msgs=msgs_to_compress,
    )
    new_summary += (
        f"\n<system-reminder>"
        f"Compressed context: {path}"
        f"</system-reminder>"
    )

self.state.summary = new_summary
self.state.context = msgs_to_reserve

AgentScope 还在每次工具调用后执行第二条路径：若结果超过 ContextConfig.tool_result_limit，就按 token 预算拆成保留部分与移出部分。

源码：agent/_agent.py

n_tokens = await self.model.count_tokens(
    [
        AssistantMsg(
            self.name,
            content=tool_result.output,
        ),
    ],
    None,
)

if (
    n_tokens
    <= self.context_config.tool_result_limit
):
    return tool_result, None

# split into reserved and offloaded blocks

配置了 Workspace/offloader 时，压缩掉的消息写入 context.jsonl，被截断的工具结果写入独立文件；Context 中保留路径，Agent 可以用 Read/Grep 重新取回。

官方文档：Context

因此两边并非“局部裁剪”和“只做摘要”的区别：

• learn-claude-code：budget → snip → micro → summary，顺序显式写在 Loop 中。
• AgentScope：tool result truncation 与 structured summary 两条自动路径，并统一接入 offloader。

配置不当导致压缩输入仍超长时，AgentScope 会继续移除最旧消息后重试。

Compact 只能让当前工作继续。用户偏好、项目约定等需要跨压缩保留的事实，应进入独立的 Memory。

Memory：索引常驻，正文按需加载

Memory 与对话历史分开存储：

Compact = 压缩当前会话已经发生的过程
Memory  = 跨压缩、跨会话保留可复用事实

两级存储

S09 把每条记忆写成独立 Markdown，再重建 MEMORY.md：

源码：s09_memory/code.py

def write_memory_file(
    name: str, mem_type: str,
    description: str, body: str,
):
    slug = name.lower().replace(" ", "-").replace("/", "-")
    filepath = MEMORY_DIR / f"{slug}.md"
    filepath.write_text(
        f"---\nname: {name}\n"
        f"description: {description}\n"
        f"type: {mem_type}\n---\n\n{body}\n"
    )
    _rebuild_index()
    return filepath

def read_memory_index() -> str:
    if not MEMORY_INDEX.exists():
        return ""
    return MEMORY_INDEX.read_text().strip()

MEMORY.md 只保存名称、描述和文件链接，适合常驻 system prompt。完整正文只在相关时加载，避免所有记忆长期占用上下文。

检索后临时注入

每个用户回合开始时，系统根据最近消息选择最多五条记忆。选择失败时退回关键词匹配。

源码：s09_memory/code.py

memories_content = load_memories(messages)
memory_turn = (
    len(messages) - 1
    if messages
    and isinstance(messages[-1].get("content"), str)
    else None
)
system = build_system()

# compression ...

request_messages = messages
if memories_content and memory_turn is not None:
    request_messages = messages.copy()
    request_messages[memory_turn] = {
        **messages[memory_turn],
        "content": (
            memories_content
            + "\n\n"
            + messages[memory_turn]["content"]
        ),
    }

这里复制 messages 后再注入，记忆只进入本次模型请求，不污染持久对话历史。

在压缩前提取

S09 在压缩前保存快照，reply 结束后再提取和合并记忆：

pre_compress = [
    m if isinstance(m, dict)
    else {
        "role": m.get("role", ""),
        "content": str(m.get("content", "")),
    }
    for m in messages
]

# compact and run agent ...

if response.stop_reason != "tool_use":
    extract_memories(pre_compress)
    consolidate_memories()
    return

若从压缩后的摘要提取，偏好、路径和决策可能已经丢失。快照保证提取器看到原始内容，而合并步骤负责减少重复记忆。

Memory 不是未经审查的事实库。自动提取可能写入错误或过期信息，完整系统还需要来源、更新时间、冲突处理和删除能力。

Agentic Memory 的渐进披露

官方文档：Long-Term Memory

AgentScope 2.0.5dev 已提供原生 AgenticMemoryMiddleware。它与 S09 采用相同的两级文件结构：

Memory/
├── MEMORY.md
├── user_profile.md
├── feedback_testing.md
└── project_context.md

MEMORY.md 只保存名称、描述和链接，并注入 System Prompt；完整正文保存在主题文件中。

源码：agentic_memory/_middleware.py

memory_md_content = (
    await self._get_memory_md_content() or ""
)
memory_md_truncated = self._truncate_if_needed(
    memory_md_content,
    self._parameters.memory_max_tokens,
)

content = (
    f"{memory_instructions}\n"
    f"## MEMORY.md\n{memory_md_truncated}"
)

return f"{current_prompt}\n\n{content}"

Agent 使用已有的 Read / Write / Edit 工具创建和维护记忆，而不是由 Middleware 在 reply 结束后强制提取。Middleware 负责提供写入规范、注入索引和检索正文。

异步检索主题文件

每次 reply 开始时，Middleware 启动一个 LLM 检索任务，根据用户输入与 frontmatter description 选择相关文件：

源码：agentic_memory/_middleware.py

self._retrieval_task = asyncio.create_task(
    self._retrieve_relevant_files(
        agent,
        self._cached_input,
    )
)

# before each reasoning step
if (
    self._retrieval_task is not None
    and self._retrieval_task.done()
):
    retrieval_result = (
        self._retrieval_task.result()
    )
    agent.state.append_context(
        agent.name,
        [HintBlock(hint=retrieval_result)],
    )

检索与当前 reply 并行，只在后续 reasoning 前检查结果。若第一轮模型没有调用工具并直接结束，异步检索可能来不及注入，这是用低延迟换取及时性的边界。

S09 会在 reply 结束后自动抽取并合并记忆；Agentic Memory 让 Agent 自己决定何时读写文件。二者的存储结构相似，写入控制权不同。

AgentScope 还支持 Mem0Middleware。它适合向量检索或托管后端，可自动检索、回写，也可暴露 search_memory / add_memory 工具；它是替代实现，不是 AgentScope Memory 的唯一实现。

State 与 Recovery：让任务可以继续

Harness 必须区分三类状态：

Turn State
  messages / pending tool calls / current retry

Session State
  transcript / summary / todo / permission / task progress

Durable State
  memory / task graph / cron / workspace binding / artifacts

Turn 失败时只重放安全步骤；进程重启时从 Session State 恢复；跨会话事实进入 Durable State。把所有状态都塞进 messages，无法同时满足恢复、查询和生命周期管理。

三条恢复路径

S11 将故障分成三类：

源码：s11_error_recovery/code.py

class RecoveryState:
    def __init__(self):
        self.has_escalated = False
        self.recovery_count = 0
        self.consecutive_529 = 0
        self.has_attempted_reactive_compact = False
        self.current_model = PRIMARY_MODEL

def retry_delay(attempt, retry_after=None):
    if retry_after:
        return retry_after
    base = min(
        BASE_DELAY_MS * (2 ** attempt),
        32000,
    ) / 1000
    jitter = random.uniform(0, base * 0.25)
    return base + jitter

退避必须有上限和抖动。优先遵守 Retry-After，避免所有并发请求在同一时间再次撞向服务。

恢复动作回到 Loop

模型调用失败后，Harness 修改状态并 continue，而不是另起一条无上下文请求。

源码：s11_error_recovery/code.py

while True:
    try:
        response = with_retry(
            lambda mt=max_tokens, mdl=state.current_model:
                client.messages.create(
                    model=mdl,
                    system=system,
                    messages=messages,
                    tools=TOOLS,
                    max_tokens=mt,
                ),
            state,
        )
    except Exception as e:
        if is_prompt_too_long_error(e):
            if not state.has_attempted_reactive_compact:
                messages[:] = reactive_compact(messages)
                state.has_attempted_reactive_compact = True
                continue
        return

    if response.stop_reason == "max_tokens":
        if not state.has_escalated:
            max_tokens = ESCALATED_MAX_TOKENS
            state.has_escalated = True
            continue

        messages.append({
            "role": "assistant",
            "content": response.content,
        })
        if state.recovery_count < MAX_RECOVERY_RETRIES:
            messages.append({
                "role": "user",
                "content": CONTINUATION_PROMPT,
            })
            state.recovery_count += 1
            continue
        return

这里重试的是模型调用，不是已经产生副作用的工具。写文件、发消息或创建资源之后能否重试，取决于工具是否幂等，以及 Harness 是否保存了调用 ID 和完成状态。

工程实现：重试只是韧性的一部分

AgentScope 通过 ModelConfig 配置每个模型的重试次数和 Fallback Model：

源码：agent/_config.py

agent = Agent(
    name="coder",
    system_prompt=SYSTEM,
    model=primary_model,
    model_config=ModelConfig(
        max_retries=2,
        fallback_model=fallback_model,
    ),
)

DeerFlow 将更多负路径拆成 Middleware：

• ToolErrorHandlingMiddleware 把工具异常转成配对的 ToolMessage，让模型可见并自纠。
• DanglingToolCallMiddleware 修复缺失结果，保持工具协议闭合。
• LoopDetectionMiddleware 阻止重复工具调用持续消耗轮次。
• TokenBudgetMiddleware 和 SubagentLimitMiddleware 为一次运行设置硬上限。
• SandboxAuditMiddleware 记录文件和命令操作。

源码：agents/middlewares

恢复决策也要进入 Event、日志和指标：至少记录错误分类、尝试次数、等待时间、当前模型、压缩动作与最终退出原因。否则自动恢复只会把故障隐藏得更久。

Todo、Task 与 Background Task

三种机制都出现“任务”，保存的却不是同一种状态：

Todo：覆盖式写入计划

S05 的 todo_write 不执行步骤。模型每次提交完整列表，Harness 校验后整体替换内存状态。

源码：s05_todo_write/code.py

def run_todo_write(todos: list) -> str:
    global CURRENT_TODOS
    todos, error = _normalize_todos(todos)
    if error:
        return error

    CURRENT_TODOS = todos
    return f"Updated {len(CURRENT_TODOS)} tasks"

每项只有 content 和 pending / in_progress / completed。没有 ID、owner、依赖，也不落盘。

连续三个工具轮没有更新 Todo 时，Loop 会在下一次模型调用前插入提醒。计数按工具轮增加，不按单个工具调用增加。

Task System：持久化依赖状态

S12 为工作单元加入 ID、owner 和 blockedBy，每条任务写入 .tasks/{id}.json。

源码：s12_task_system/code.py

@dataclass
class Task:
    id: str
    subject: str
    description: str
    status: str
    owner: str | None
    blockedBy: list[str]

def save_task(task: Task):
    _task_path(task.id).write_text(
        json.dumps(asdict(task), indent=2)
    )

认领任务前必须检查所有依赖：

源码：s12_task_system/code.py

def can_start(task_id: str) -> bool:
    task = load_task(task_id)
    for dep_id in task.blockedBy:
        if not _task_path(dep_id).exists():
            return False
        if load_task(dep_id).status != "completed":
            return False
    return True

def claim_task(task_id: str, owner: str = "agent") -> str:
    task = load_task(task_id)
    if task.status != "pending":
        return f"Task {task_id} is {task.status}, cannot claim"
    if not can_start(task_id):
        return "Blocked"
    task.owner = owner
    task.status = "in_progress"
    save_task(task)

状态路径只有 pending → in_progress → completed。完成后会重新扫描当前可开始的任务，但没有环检测、文件锁或原子认领，因此教学实现不能直接承担多进程调度。

Task 进入运行状态

AgentScope 的 Task 不单独写 JSON 文件，而是保存在 AgentState.tasks_context 中。Task 同时维护 blocks 与 blocked_by 两个方向：

源码：state/_task.py

class Task(BaseModel):
    subject: str
    description: str
    metadata: dict[str, Any]
    state: Literal[
        "pending", "in_progress", "completed"
    ] = "pending"
    id: str = Field(default_factory=_generate_id)
    owner: str | None = None
    blocks: list[str] = Field(default_factory=list)
    blocked_by: list[str] = Field(default_factory=list)

TaskCreate / TaskGet / TaskList / TaskUpdate 都是 state-injected 工具。调用时由运行时注入当前 AgentState，模型不能伪造另一份状态。

TaskUpdate 在增加依赖时会同步更新关系的两端，也支持 owner、metadata 与删除。但它没有强制检查状态只能按顺序迁移，TaskList 也只是展示阻塞关系；是否认领可执行任务主要由工具说明约束。

在 Agent Service 中，任务随 session state 保存。它比 S12 的散落 JSON 更容易跟随一次会话恢复，但仍不是跨 session 的全局调度数据库。

Background Task：先闭合协议，再返回结果

慢命令放到 daemon thread 后，原始 tool_use 不能一直没有结果。S13 立即返回一个占位 tool_result：

源码：s13_background_tasks/code.py

if should_run_background(block.name, block.input):
    bg_id = start_background_task(block)
    results.append({
        "type": "tool_result",
        "tool_use_id": block.id,
        "content": (
            f"[Background task {bg_id} started] "
            "Result will be available when complete."
        ),
    })

worker 完成后把输出写入共享字典。主循环在后续工具轮收集结果，作为普通文本事件注入：

for notif in collect_background_results():
    user_content.append({
        "type": "text",
        "text": notif,
    })

完成通知不再使用原 tool_use_id，因为那次调用已经由占位结果闭合。

S13 没有独立的通知唤醒器。若模型收到占位结果后直接结束，完成通知要等下一次用户输入或后续工具轮才能被收集。

超时卸载与主动唤醒

AgentScope 的 ToolOffloadMiddleware 不靠命令关键词判断。工具先正常执行，超过 timeout_secs 才转为后台任务；已经运行的 asyncio task 不会被重新启动。

源码：app/middleware/_tool_offload_middleware.py

tool = await agent.toolkit.get_tool(tool_call.name)
if tool is not None and (
    tool.is_state_injected
    or tool.is_external_tool
):
    async for item in next_handler(**input_kwargs):
        yield item
    return

drain_task = asyncio.create_task(
    _drain_to_queue()
)
deadline = loop.time() + self._timeout_secs

这两类工具都绕过 ToolOffloadMiddleware：state-injected tool 在当前执行轮次前台运行；external tool 进入外部执行事件协议，暂停并等待结果。

超时后，Middleware 立即返回与原调用配对的占位 ToolResponse。后台执行完成时，再把结果封装成 HintBlock，投递到 session inbox 并发送 wake trigger：

源码：app/middleware/_tool_offload_middleware.py

await self._message_bus.queue_push(
    MessageBusKeys.inbox(session_id),
    hint.model_dump(mode="json"),
)
await enqueue_run_trigger(
    self._message_bus,
    user_id=self._user_id,
    session_id=session_id,
    agent_id=self._agent_id,
)

因此即使当前 reply 已结束，完成结果也能唤醒新一轮 reasoning。取消或 watcher 异常时不会投递完成通知，这是当前实现仍需上层处理的失败分支。

Subagent 与 Agent Team

SubAgent 与 Team 都能分工，关键区别在返回路径和生命周期。

SubAgent：把上下文隔离成一次工具调用

S06 给父 Agent 增加 task 工具。handler 创建全新的 messages，运行另一个 Agent Loop，最后只返回摘要。

源码：s06_subagent/code.py

def spawn_subagent(description: str) -> str:
    messages = [
        {"role": "user", "content": description}
    ]

    for _ in range(30):
        response = client.messages.create(
            model=MODEL,
            system=SUB_SYSTEM,
            messages=messages,
            tools=SUB_TOOLS,
            max_tokens=8000,
        )
        messages.append({
            "role": "assistant",
            "content": response.content,
        })
        if response.stop_reason != "tool_use":
            break

        # execute subagent tools ...

    return extract_text(messages[-1]["content"])

子 Agent 不拥有 task 工具，因此不能继续递归委派。它仍执行相同的权限 Hook，但完整历史在返回后被丢弃，父 Agent 只接收结论。

CoreCoder 采用相同边界：AgentTool.execute() 同步调用 sub.chat(task)。它隔离消息历史，但共享 LLM、进程和工作目录：corecoder/tools/agent.py。

Agent Team：S15 执行，S16 驻留

S15 将 teammate 放进 daemon thread，但生命周期仍是有限的：最多执行 10 个工具轮，模型停止调用工具后立即发送总结并退出。

关键控制流：s15_agent_teams/code.py

for _ in range(10):
    inbox = BUS.read_inbox(name)
    if inbox:
        messages.append({
            "role": "user",
            "content": (
                f"<inbox>{json.dumps(inbox)}</inbox>"
            ),
        })

    response = client.messages.create(
        model=MODEL,
        system=system,
        messages=messages[-20:],
        tools=sub_tools,
        max_tokens=8000,
    )
    messages.append({
        "role": "assistant",
        "content": response.content,
    })

    if response.stop_reason != "tool_use":
        break

BUS.send(name, "lead", summary, "result")
active_teammates.pop(name, None)

S16 才加入 idle inbox loop。teammate 没有工具调用时不退出，而是等待新消息或 shutdown request：

关键控制流：s16_team_protocols/code.py

while not shutdown_requested:
    inbox = BUS.read_inbox(name)

    # handle protocol and regular messages ...

    response = client.messages.create(
        model=MODEL,
        system=system,
        messages=messages[-20:],
        tools=sub_tools,
        max_tokens=8000,
    )

    if response.stop_reason != "tool_use":
        while not shutdown_requested:
            time.sleep(1)
            inbox = BUS.read_inbox(name)
            if inbox:
                # append messages, then return to LLM
                break

plan approval、shutdown request 和 response 使用 request_id 关联请求。S15 解决后台并发，S16 才建立长期成员协议。

S17：从驻留队友到自主队友

S16 的 teammate 只能等 leader 发消息。S17 在 idle 阶段加入任务板扫描：没有新消息时，主动寻找依赖已完成、尚未被领取的任务。

源码：s17_autonomous_agents/code.py

def scan_unclaimed_tasks() -> list[dict]:
    """Find pending, unowned tasks with all dependencies completed."""
    unclaimed = []
    for f in sorted(TASKS_DIR.glob("task_*.json")):
        task = json.loads(f.read_text())
        if (task.get("status") == "pending"
                and not task.get("owner")
                and can_start(task["id"])):
            unclaimed.append(task)
    return unclaimed

def idle_poll(name: str, messages: list, role: str) -> str:
    """Poll for 60s. Return 'work', 'shutdown', or 'timeout'."""
    for _ in range(IDLE_TIMEOUT // IDLE_POLL_INTERVAL):
        time.sleep(IDLE_POLL_INTERVAL)

        # Check inbox — dispatch protocol messages first
        inbox = BUS.read_inbox(name)
        if inbox:
            # Check for shutdown_request
            for msg in inbox:
                if msg.get("type") == "shutdown_request":
                    req_id = msg.get("metadata", {}).get("request_id", "")
                    BUS.send(name, "lead", "Shutting down gracefully.",
                             "shutdown_response",
                             {"request_id": req_id, "approve": True})
                    return "shutdown"

            # Non-protocol inbox: inject and resume work
            messages.append({"role": "user",
                "content": "<inbox>" + json.dumps(inbox) + "</inbox>"})
            return "work"

        # Scan task board
        unclaimed = scan_unclaimed_tasks()
        if unclaimed:
            task = unclaimed[0]
            result = claim_task(task["id"], name)
            if "Claimed" in result:
                messages.append({"role": "user",
                    "content": f"<auto-claimed>Task {task['id']}: "
                               f"{task['subject']}</auto-claimed>"})
                return "work"

    return "timeout"

can_start() 负责检查依赖，claim_task() 负责占有任务。扫描和领取分开，因此真正的 claim 必须是原子操作；否则多个 teammate 可能同时选中同一任务。

S17 的自主性只发生在任务分配层。teammate 仍受工具权限、Workspace、最大轮数和 shutdown 协议约束，不会因为能够认领任务就绕过 Harness。

工业限制：并发上限不能只写在 Prompt

DeerFlow 的 SubagentLimitMiddleware 在模型返回后检查 task 调用，超过上限时直接截断多余调用：

源码：subagent_limit_middleware.py

task_indices = [
    i for i, tc in enumerate(tool_calls)
    if tc.get("name") == "task"
]
if len(task_indices) <= self.max_concurrent:
    return None

indices_to_drop = set(
    task_indices[self.max_concurrent:]
)
truncated_tool_calls = [
    tc for i, tc in enumerate(tool_calls)
    if i not in indices_to_drop
]
updated_msg = clone_ai_message_with_tool_calls(
    last_msg,
    truncated_tool_calls,
)
return {"messages": [updated_msg]}

Prompt 可以告诉模型“最多创建三个子任务”，Middleware 才能保证限制一定执行。生产系统还要同时限制全局并发、单任务预算和递归深度。

Worker：独立 Session 与状态

AgentScope 将团队放在 Agent Service 层。开发者先注册 SubAgentTemplate，描述 worker 的 system prompt、Context/ReAct 配置、初始 Task 和权限继承策略。

源码：app/_types.py

class SubAgentTemplate(BaseModel):
    type: str
    description: str
    system_prompt_template: str
    context_config: ContextConfig
    react_config: ReActConfig
    permission_context: PermissionContext
    tasks_context: TaskContext

AgentCreate 被调用后，服务创建独立的 AgentRecord、SessionRecord 和 AgentState。worker 继承 leader 的模型配置与 workspace_id，但拥有独立 Context、Task、事件流和权限状态。

源码：app/_tool/_agent_create.py

worker_state = AgentState(
    permission_context=worker_permission_context,
    tasks_context=template.tasks_context.model_copy(
        deep=True,
    ),
)

worker_session = await self._storage.upsert_session(
    user_id=self._user_id,
    agent_id=worker_agent.id,
    config=SessionConfig(
        workspace_id=leader_session.config.workspace_id,
        chat_model_config=(
            leader_session.config.chat_model_config
        ),
    ),
    state=worker_state,
)

首个任务不作为父 Agent 的同步工具结果返回，而是写入 worker session inbox 并触发 wake。后续结果通过 TeamSay 回到 leader。

这与 S06 的一次性 SubAgent 不同：S06 是嵌套函数调用；AgentScope Team 是多个可独立运行和恢复的 session。共享 workspace_id 也意味着它默认不是文件隔离，写代码时仍需独立 Workspace 或 Worktree。

教学实现仍共享文件系统，线程也不等于隔离。若 teammate 会写代码，还需要为每个成员分配独立 Workspace 或 Git Worktree。

Cron Scheduler

Cron 不直接执行命令。它在时间匹配时生产一个 prompt，再由 Agent Loop 读取并决定行动。

注册与持久化

源码：s14_cron_scheduler/code.py

def schedule_job(
    cron: str,
    prompt: str,
    recurring: bool = True,
    durable: bool = True,
):
    err = validate_cron(cron)
    if err:
        return err

    job = CronJob(
        id=f"cron_{random.randint(0, 999999):06d}",
        cron=cron,
        prompt=prompt,
        recurring=recurring,
        durable=durable,
    )
    with cron_lock:
        scheduled_jobs[job.id] = job

    if durable:
        save_durable_jobs()
    return job

durable 只把定义写入 .scheduled_tasks.json。调度线程仍必须保持运行，进程关闭期间错过的时间点不会自动补偿。

Scheduler 只写队列

源码：s14_cron_scheduler/code.py

def cron_scheduler_loop():
    while True:
        time.sleep(1)
        now = datetime.now()
        minute_marker = now.strftime("%Y-%m-%d %H:%M")

        with cron_lock:
            for job in list(scheduled_jobs.values()):
                if cron_matches(job.cron, now):
                    if _last_fired.get(job.id) != minute_marker:
                        cron_queue.append(job)
                        _last_fired[job.id] = minute_marker

                    if not job.recurring:
                        scheduled_jobs.pop(job.id, None)

minute_marker 防止同一 job 在一分钟内重复入队。时间来自本机 datetime.now()，因此运行环境的时区就是调度时区。

Queue Processor 负责唤醒

scheduler 不调用模型。独立 queue processor 发现队列非空后，尝试取得 agent_lock，只在 Agent 空闲时拉起新回合。

源码：s14_cron_scheduler/code.py

def queue_processor_loop():
    while True:
        time.sleep(0.2)
        if not has_cron_queue():
            continue
        if not agent_lock.acquire(blocking=False):
            continue

        try:
            if has_cron_queue():
                run_agent_turn_locked()
        finally:
            agent_lock.release()

Agent Loop 消费队列时，只追加普通 user 消息：

for job in consume_cron_queue():
    messages.append({
        "role": "user",
        "content": f"[Scheduled] {job.prompt}",
    })

因此 Cron 复用了既有入口：User Message → Agent Loop → Tool Use。调度、交付和执行彼此分离。

持久调度与 Session Inbox

AgentScope Agent Service 使用 APScheduler，不手写 cron parser。ScheduleCreate 除五字段表达式外，还保存 IANA timezone、启停时间、permission mode 和 stateful：

源码：app/_manager/_scheduler/_tools/_schedule_create.py

class _ScheduleCreateParams(BaseModel):
    name: str
    description: str = ""
    cron_expression: str
    timezone: str = "UTC"
    enabled: bool = True
    started_at: datetime | None = None
    ended_at: datetime | None = None
    stateful: bool = False
    permission_mode: str = PermissionMode.DONT_ASK.value

stateful=True 时多次触发复用同一个 session；否则每次创建新 session。服务启动时会从 storage 恢复 schedule，再注册到 AsyncIOScheduler。

触发后同样不直接调用 ChatService，而是把 description 包装成 HintBlock，投递到目标 session 并唤醒：

源码：app/_manager/_scheduler/_scheduler_manager.py

hint = HintBlock(
    hint=(
        "<scheduled-task>\n"
        f"{record.data.description}\n"
        "</scheduled-task>"
    ),
    source=json.dumps(
        {
            "label": "schedule",
            "sublabel": record.data.name,
        },
        ensure_ascii=False,
    ),
)

await message_bus.queue_push(
    MessageBusKeys.inbox(session.id),
    hint.model_dump(mode="json"),
)
await enqueue_run_trigger(
    message_bus,
    user_id=record.user_id,
    session_id=session.id,
    agent_id=record.agent_id,
)

TeamSay、后台工具结果与 Cron 因此共用同一条交付路径：HintBlock → session inbox → wakeup → reasoning。差别只在事件来源。

消息协议：Inbox、Wakeup 与外部 IM

Agent 间消息至少需要三件事：按接收者投递、可消费的 inbox、让空闲 Agent 醒来的信号。外部即时通信还要增加平台 Channel 适配层。

最小文件邮箱

S15 为每个 Agent 建立一个 JSONL inbox。发送是追加，读取是消费后删除。

源码：s15_agent_teams/code.py

class MessageBus:
    def send(
        self,
        from_agent: str,
        to_agent: str,
        content: str,
        msg_type: str = "message",
    ):
        msg = {
            "from": from_agent,
            "to": to_agent,
            "content": content,
            "type": msg_type,
            "ts": time.time(),
        }
        inbox = MAILBOX_DIR / f"{to_agent}.jsonl"
        with open(inbox, "a") as f:
            f.write(json.dumps(msg) + "\n")

    def read_inbox(self, agent: str) -> list[dict]:
        inbox = MAILBOX_DIR / f"{agent}.jsonl"
        if not inbox.exists():
            return []
        msgs = [
            json.loads(line)
            for line in inbox.read_text().splitlines()
        ]
        inbox.unlink()
        return msgs

读取后的消息以 <inbox>...</inbox> 注入 user turn，Agent 不需要另一套推理入口。

这个教学实现没有文件锁。追加与删除并发时可能丢消息，崩溃后也没有 ack 或重试。

产品实现：Inbox 与 Wakeup 分离

AgentScope Agent Service 的 TeamSay 先把 HintBlock 放进目标 session 的 inbox，再写入 run trigger。

源码：app/_tool/_team_say.py

for sid, aid in recipients:
    await self._message_bus.queue_push(
        MessageBusKeys.inbox(sid),
        payload,
    )
    await enqueue_run_trigger(
        self._message_bus,
        user_id=self._user_id,
        session_id=sid,
        agent_id=aid,
    )

先投递，再唤醒，保证调度器收到信号时 inbox 中已经有数据。session 正在运行时，当前执行轮次会读取 inbox；空闲时由 wake trigger 启动新一轮。

InboxMiddleware 在每次 reasoning 前 drain 队列，把消息恢复成 HintBlock 并加入 Agent Context：

源码：app/middleware/_inbox_middleware.py

entries = await self._bus.queue_drain(
    MessageBusKeys.inbox(agent.state.session_id),
    max_count=self._max_count,
)

hint_blocks = [
    HintBlock.model_validate(payload)
    for _entry_id, payload in entries
]

agent.state.context.append(
    AssistantMsg(
        id=agent.state.reply_id,
        name=agent.name,
        content=list(hint_blocks),
    )
)

消息通道负责送达，协议层负责语义。普通协作消息可以直接消费；plan approval、shutdown 等请求还要携带 request_id，让 response 回到正确的等待状态。

外部 IM Channel 是适配层

飞书、Slack、Telegram 等外部聊天平台位于 Harness 之外。DeerFlow 将平台消息统一成 InboundMessage，再由调度器映射为 Agent thread；Agent 输出则转成 OutboundMessage 返回原平台。

源码：channels/message_bus.py

@dataclass
class InboundMessage:
    channel_name: str
    chat_id: str
    user_id: str
    text: str
    msg_type: InboundMessageType = InboundMessageType.CHAT
    thread_ts: str | None = None
    topic_id: str | None = None
    files: list[dict[str, Any]] = field(default_factory=list)

外部 IM 的 MessageBus 连接 channel adapter 与 Agent 调度器；AgentScope 的 MessageBus 管理 session inbox、事件和唤醒。名字相同，所在层次不同。

完整 Harness：把组件放回 Loop

S20 没有增加新机制，而是把前面的教学组件合到同一个可运行系统。循环仍负责“模型 → 工具 → 结果”；Harness 负责循环前后的输入、能力、安全、状态与调度。

用户输入
  → UserPromptSubmit / input sanitation
  → Cron / Inbox / background notification
  → Context compact / memory / skills
  → Model call + recovery
  → tool_use?
      否 → Stop / terminal guard → output
      是 → validation / permission / sandbox
          → tool execution
          → result sanitation / audit
          → tool_result 回灌
          → 下一轮

这张图的重点不是组件数量，而是执行位置。位置决定它能读取什么状态、能否阻断操作，以及失败后由谁恢复。

S20 的 agent_loop() 最终仍然只做三件事：调用模型，执行 tool_use，把 tool_result 追加回 messages。调度、安全、恢复和团队协作都挂在这条闭环的前后。

messages → LLM(tools)
             ↓ tool_use
messages ← tool_result
             ↓
          while

去掉 Harness 的扩展机制，核心仍然是 LLM + tools + messages + while。

参考资料

• Learn Claude Code
• 深入 Claude Code 源码
• 通用 Agent 架构地图
• AgentScope 2.0
• shareAI-lab/learn-claude-code
• he-yufeng/CoreCoder
• agentscope-ai/agentscope
• bytedance/deer-flow