本文整理自 The Multi-Agent Primitive Model。

Multi-Agent 系统设计：原语以及在 nonoka 中的实践

1. Multi Agent 原语

所有的 Multi Agent 系统都需要解决下面的问题：

怎么定义一个 agent？
一个 agent 做完后，怎么把控制权交给另一个？
多个 agent 之间怎么共享上下文和记忆？
谁来决定下一个 agent 该做什么？

这四个问题恰好对应了 Multi Agent 的四个原语：agent、handoff、shared state、orchestrator。

$a.$ Agent：无状态的计算单元

Agent 是一个配置对象，而不是有长期身份的运行时实体。它没有状态、没有决策逻辑（不决定下一个谁运行）、没有生命周期——给同样的输入，应该得到可预期的输出。

Agent = (system_prompt, tools, model, optional_name)

$b.$ Handoff：控制权转移

Handoff 是一个协议，定义了 agent 之间如何转移控制权。目前的主流框架有如下设计：

实现	决策位置	代表
函数返回下一个 agent	当前 agent 内部	OpenAI Swarm / Agents SDK
图边 + 条件	全局图定义	LangGraph
Speaker selector	中心 selector	AutoGen GroupChat

不同设计的简单实现如下：

函数返回下一个 agent：Agent 把返回另一个 agent 的函数注册为 tool，模型调用它时运行时转移控制权：

from swarm import Agent

weather_agent = Agent(name="weather", instructions="回答天气问题。")

def transfer_to_weather():
    return weather_agent

assistant = Agent(
    name="assistant",
    instructions="把天气问题转给 Weather Agent。",
    functions=[transfer_to_weather],
)

这种设计的特点是局部决策：每个 agent 自己决定下一步交给谁，不需要中心协调器。

图边 + 条件：Handoff 被显式建模为有向图中的边，节点是 agent 或工具，边是状态转换条件：

from langgraph.graph import StateGraph

builder = StateGraph(State)
builder.add_node("router", router_agent)
builder.add_node("researcher", researcher_agent)
builder.add_node("coder", coder_agent)

builder.add_edge("router", "researcher", condition=lambda s: s["next"] == "research")
builder.add_edge("router", "coder", condition=lambda s: s["next"] == "code")

这种设计的特点是全局可见：整个系统的控制流都画在一张图里。

Speaker selector（AutoGen GroupChat）。GroupChat 维护一个共享消息池，每次由 selector（LLM 或规则）决定下一个发言的 agent：

from autogen import GroupChat

groupchat = GroupChat(
    agents=[planner, coder, reviewer],
    messages=[],
    speaker_selection_method="auto",  # LLM 选择下一个发言者
)

这种设计的特点是对话驱动：agent 之间通过自然语言消息协作，更像一个会议而不是一个流水线。

$c.$ Shared state：唯一带状态的地方

Shared state 是一个状态容器，所有 agent 共享的状态都存储在这里。它是唯一带状态的原语。一般分为下面两类：

Full pool：所有 agent 看到同样的完整状态。好处是简单，坏处是容易产生上下文污染和隐私泄露；
Projected view：每个 agent 只看到自己需要的那部分状态。例如 supervisor 看计划，worker 只看自己的子任务。

在设计 Multi Agent 系统的状态时，可以考虑下面的问题来帮助确定最终的状态设计：

哪些状态必须共享？ 对话历史、计划、中间结果、长期记忆分别放在哪里？
每个 agent 能看到什么？ 是否允许 worker 看到 supervisor 的内部推理？
状态如何合并？ 多个 worker 并行返回结果时，如何合并到 shared state 中？
冲突怎么解决？ 两个 agent 同时写同一个 key 怎么办？

$e.g.$ 一个 supervisor 系统里，lead 生成 plan 后写入 shared state 的 plan 字段；每个 worker 读取自己的子任务、写入 findings[<worker_id>]；最后 lead 读取 findings 进行综合。这个简单的读写约定比框架选择更能决定系统是否稳定。

$d.$ Orchestrator：决定下一个谁运行

Orchestrator = ({state, last_speaker}) -> next_agent。它不是必须存在的，例如 swarm 里每个 agent 就是自己决定下一步。但只要有中心协调逻辑，就可以抽象为这样一个函数。下面是四种常见的设计：

类型	决策方式	代表框架
Static	编译期定好的图	LangGraph 确定性图、CrewAI Sequential
LLM-selected	由 LLM 选下一个发言者	AutoGen SelectorGroupChat、CrewAI Hierarchical
Handoff-driven	当前 agent 用 tool call 决定	OpenAI Swarm / Agents SDK
Queue-driven	从共享队列取任务	大规模 swarm / worker pool

选择哪种 orchestrator 取决于具体情景，一般而言：

需要严格可重现、可审计 → Static；
需要动态适应输入 → LLM-selected 或 Handoff-driven；
需要高吞吐并行 → Queue-driven。

2. Nonoka agent 的 multi-agent 设计

$a.$ Sub Agent 接口设计

nonoka agent 以及 nonoka cli 中采用了 AgentTool 的设计方式，与四个原语的对应关系如下：

原语	笔记里的抽象	nonoka 的实现
Agent	无状态配置对象	frozen dataclass
Handoff	(from, to, reason, payload)	Capability 统一协议 + AgentTool 实现
Shared state	message pool / blackboard / kv / vector	Session ↔ SessionState 快照 + SQLite；checkpoint 管控制、memory 管上下文
Orchestrator	`({state, last_speaker})` -> next	ReAct 等 Agent 范式 API 内部负责调用

nonoka 的 handoff 是直接将 sub agent 作为一个 Tool 给 ReAct 等 Agent 范式 API 内部调用的。AgentTool 调用的 sub agent 也会像普通的 Agent 一样完整地进行 _ensure_llm → emit_session_start → paradigm.run → trace → emit_session_end 流程。

$b.$ Lineage：对 Sub Agent 调用的重放设计

但是如果只是一个单纯的 Tool 的话会有下面的问题：父进程如果在这之后崩溃，子 agent 的结果和它做的事就全丢了。普通的 Tool 只需要回到调用前 SessionState 重来一遍就行，但是 AgentTool 的重放成本十分高。

这里简单提及一下 nonoka agent 对 Tool 调用的回复流程：

崩溃后 resume()，扫描对话历史，找出”assistant 消息里声明了要调用某个工具、但没有对应结果”的调用（也即悬挂调用）；

对每个悬挂调用，重新 invoke() 执行一遍；

把结果写回对话，继续运行。

为此 nonoka agent 引入了 lineage 的设计，它包含下面的模块：

怎么认出是同一次调用：用”子 agent 名 + 任务描述 sha256 的部分结果” 得到一个唯一 id 。崩溃前后同一个任务算出的 id 相同，恢复时可以直接用 key 找到；
记录存在哪：存在父 session 的状态里，并且一写入就立刻持久化到 checkpoint——父进程当场崩溃，记录也不会丢；
调用前先查记录：有记录且已完成，直接返回记录里存的结果，不重跑；有记录但还在跑，把子 session 从 checkpoint 恢复出来接着跑，不新开；没有记录，才正常新开一个子 session。

# 伪代码：_run_child 开头（nonoka/core/agent_tool.py）
lineage_key = f"{agent.name}:{sha256(prompt)[:16]}"
record = ctx.session.extension_state["agent_tool_lineage"].get(lineage_key)

if record and record["status"] == "completed":
  return RunResult(success=True, data=record["result_text"])  # 缓存，不重跑
if record and record["status"] == "running":
  session = await _resume_child_session(ctx, runner, record)   # 续跑，不新开
if session is None:
  lineage[key] = {"child_session_id": ..., "status": "running", ...}
  await checkpoint_store.save_session(ctx.session)            # 先落盘再跑

AgentTool 并没有绕过 ReAct 的恢复框架，而是在重新调用 invoke 这个过程中找到 lineage 记录、恢复执行：

找 key，确认是同一次调用。

# invoke 内部
prompt = arguments["task"]                    # task + 可选 context
lineage_key = f"{agent.name}:{sha256(prompt)[:16]}"

去父 session 的 extension_state 里找 lineage 记录。

# invoke → _run_child 开头
lineage = ctx.session.extension_state.setdefault("agent_tool_lineage", {})
record = lineage.get(lineage_key)

用这个记录去找子 session，由于子 Session 也是 Agent 调用，它的 SessionState 也会被持久化、直接从数据库加载即可。

# _resume_child_session
state = await runner.checkpoint_store.load_session(record["child_session_id"])
# → Session.from_state(state, ...) 重建子 session（含 memory、预算）
# → paradigm.resume(session, runner) 接着跑

$c.$ nonoka 的状态投影

nonoka agent 的状态投影如下

# nonoka/core/agent_tool.py
class MemoryStrategy(str, Enum):
  ISOLATE = "isolate"   # 默认：子 Session 全新空记忆
  INHERIT = "inherit"   # 拷贝父 memory 最后 N 条（默认 5），不回写
  SHARE = "share"       # 共享同一 WorkingMemory 对象（有竞态，慎用）

Multi-Agent 系统设计：原语以及在 nonoka 中的实践

1. Multi Agent 原语

a.a.a. Agent：无状态的计算单元

b.b.b. Handoff：控制权转移

c.c.c. Shared state：唯一带状态的地方

d.d.d. Orchestrator：决定下一个谁运行