使用Claude实现异步多智能体编排
技术背景
随着大语言模型能力的增强，单一模型在处理复杂任务时面临上下文长度限制、任务分解困难等问题。多智能体编排通过将任务分配给多个专业化的智能体，实现并行处理、分工协作，从而提升效率和效果。Anthropic的Claude模型支持工具调用，结合异步编程框架asyncio，可以构建灵活的多智能体系统。本教程旨在展示两种基础编排模式，为开发者提供可复用的架构模板，解决实际应用中任务分解、消息传递和生命周期管理的问题。
核心原理
核心原理是基于消息传递的异步协作。每个智能体拥有独立的收件箱（inbox）和事件（event），通过Hub进行消息路由。智能体通过工具调用send_message发送消息，通过wait_for_message阻塞等待消息。基础代理循环（run_agent）处理模型响应，将工具结果和收件箱消息附加到下一次请求中，实现消息内联到达，避免轮询。异步子智能体模式通过create_subagents工具动态生成子任务，主代理可查询状态、收集结果并终止子代理，实现动态编排。
架构分析
系统由Hub（消息中心）、run_agent（代理循环）、工具集（BASE_TOOLS、SLEEP、SUBAGENT_TOOLS）和调度器（extra_dispatch）组成。Hub管理智能体注册、消息投递、状态跟踪和名称生成。run_agent是通用执行器，处理模型调用、工具分发和消息注入。固定团队模式中，所有智能体预先注册，并行运行，通过消息协作。异步子代理模式中，主代理动态创建子代理，子代理独立运行，通过工具与主代理交互。架构设计强调解耦：消息传递与领域逻辑分离，工具可插拔，便于扩展。
完整实战教程
本教程展示了Claude Opus 4.8系统卡中多智能体结果背后的两种多智能体编排模式的“形态”：固定N智能体团队和异步子智能体。这里没有领域任务——仅包含消息传递和子智能体机制，以便您能清楚地看到哪些工具被触发以及触发顺序，然后可以自行添加工具和任务。
所有内容均基于公开的Anthropic Python SDK和asyncio运行，使用任意API密钥，通常在三十秒内完成。
设置
%pip install -qU anthropic
import asyncio
import itertools
from collections import Counter, defaultdict
import anthropic
MODEL = "claude-opus-4-8" # swap for the newest Claude model available to you
client = anthropic.AsyncAnthropic() # reads ANTHROPIC_API_KEY from env
共享构建模块
消息中心
每个智能体都有一个收件箱列表和一个用于阻塞等待的 asyncio.Event。drain 清空收件箱并返回结构化消息；render 将它们格式化为文本，以便附加到工具结果中。
class Hub:
def __init__(self):
self.inbox: dict[str, list[dict]] = defaultdict(list)
self.event: dict[str, asyncio.Event] = defaultdict(asyncio.Event)
self.status: dict[str, str] = {}
self._ids = itertools.count(1)
def register(self, name: str):
self.status[name] = "active"
_ = self.inbox[name], self.event[name]
def new_name(self, prefix="helper") -> str:
n = f"{prefix}{next(self._ids)}"
self.register(n)
return n
def post(self, sender: str, recipients: list[str], content: str) -> list[str]:
delivered = []
for rid in recipients:
if rid in self.status:
self.inbox[rid].append({"from": sender, "content": content})
self.event[rid].set()
delivered.append(rid)
return delivered
def drain(self, name: str) -> list[dict]:
msgs, self.inbox[name] = self.inbox[name], []
self.event[name] = asyncio.Event()
return msgs
@staticmethod
def render(msgs: list[dict]) -> str:
if not msgs:
return ""
body = "\n".join(
f'<agent-message from="{m["from"]}">\n{m["content"]}\n</agent-message>' for m in msgs
)
return f"\n\n[Messages received while you were working:]\n{body}"
消息工具
两种模式中的每个智能体都会获得这两个工具。
SEND_MESSAGE = {
"name": "send_message",
"description": (
"Send a message to one or more other agents. It will appear appended to their next tool "
"result. This is the ONLY way to reach other agents — plain text in your turn goes nowhere."
),
"input_schema": {
"type": "object",
"properties": {
"recipient_ids": {"type": "array", "items": {"type": "string"}, "minItems": 1},
"content": {"type": "string"},
},
"required": ["recipient_ids", "content"],
},
}
WAIT_FOR_MESSAGE = {
"name": "wait_for_message",
"description": (
"Block until another agent messages you. Note: messages also arrive automatically appended "
"to the result of ANY other tool call, so only use this when you have nothing else to do."
),
"input_schema": {"type": "object", "properties": {}},
}
BASE_TOOLS = [SEND_MESSAGE, WAIT_FOR_MESSAGE]
基础代理循环
一个函数运行任何代理：一个标准的工具使用循环，通过 hub 调度 send_message / wait_for_message，通过 extra_dispatch 路由任何额外的工具，并将排空的收件箱附加到最后一个工具结果——因此代理从不轮询；消息内联到达。每次工具调用和收件箱投递都会实时打印，以便您观察编排过程；最后会打印摘要跟踪。
TRACE: dict[str, list[str]] = defaultdict(list)
def _snip(s, n=60):
s = str(s).replace("\n", " ")
return s if len(s) <= n else s[:n] + "…"
async def run_agent(
hub: Hub,
name: str,
system: str,
first_user_turn: str,
tools: list = None,
extra_dispatch=None,
max_turns: int = 20,
) -> str:
tools = tools or BASE_TOOLS
extra_dispatch = extra_dispatch or {}
messages = [{"role": "user", "content": first_user_turn}]
try:
for _ in range(max_turns):
resp = await client.messages.create(
model=MODEL,
max_tokens=2048,
system=system,
tools=tools,
messages=messages,
)
messages.append({"role": "assistant", "content": resp.content})
if resp.stop_reason == "end_turn":
hub.status[name] = "done"
return "".join(getattr(b, "text", "") for b in resp.content)
if resp.stop_reason != "tool_use":
raise RuntimeError(f"unexpected stop_reason: {resp.stop_reason}")
results = []
for block in resp.content:
if block.type != "tool_use":
continue
TRACE[name].append(block.name)
if block.name == "send_message":
rids = block.input["recipient_ids"]
delivered = hub.post(name, rids, block.input["content"])
unknown = [r for r in rids if r not in delivered]
out = f"delivered to {delivered}" + (f"; unknown: {unknown}" if unknown else "")
elif block.name == "wait_for_message":
hub.status[name] = "idling"
try:
await asyncio.wait_for(hub.event[name].wait(), timeout=60)
out = "woke: new messages"
except TimeoutError:
out = "woke: 60s timeout"
hub.status[name] = "active"
elif block.name in extra_dispatch:
out = await extra_dispatch[block.name](block)
else:
out = f"error: no dispatch for {block.name}"
print(f" [{name}] {block.name}({_snip(block.input)}) → {_snip(out)}")
results.append({"type": "tool_result", "tool_use_id": block.id, "content": out})
inbox = hub.drain(name)
for m in inbox:
print(f" [{name}] ← received from {m['from']}: {_snip(m['content'])}")
if results:
results[-1]["content"] += hub.render(inbox) # ← the key line
messages.append({"role": "user", "content": results})
hub.status[name] = "done"
return f"[{name} hit max_turns={max_turns}]"
except Exception:
hub.status[name] = "crashed"
raise
def print_trace():
for agent in sorted(TRACE):
counts = Counter(TRACE[agent])
print(f" {agent}: " + ", ".join(f"{n}×{t}" for t, n in counts.most_common()))
TRACE.clear()
第一部分 · 固定N智能体团队
三个智能体——一个主导和两个辅助——每个都被赋予一行角色描述。它们所做的只是通过send_message互相介绍自己；主导写一句总结并结束运行。没有领域工具：这纯粹是消息传递路径。
TEAM_SYSTEM = "You are {name}, one of 3 agents working together (peers: {peers})."
TASKS = {
"lead": "You are the lead. Introduce yourself to the others. Once everyone has introduced "
"themselves, finish with a one-sentence summary of the team.",
"helper1": "You are a backend engineer named Ada. Introduce yourself to the others, then wait "
"for their replies.",
"helper2": "You are a designer named Bo. Introduce yourself to the others, then wait for their "
"replies.",
}
async def run_team() -> str:
hub = Hub()
names = list(TASKS)
for n in names:
hub.register(n)
helper_tasks = [
asyncio.create_task(
run_agent(
hub,
n,
system=TEAM_SYSTEM.format(name=n, peers=[p for p in names if p != n]),
first_user_turn=TASKS[n],
)
)
for n in names[1:]
]
try:
return await run_agent(
hub,
"lead",
system=TEAM_SYSTEM.format(name="lead", peers=names[1:]),
first_user_turn=TASKS["lead"],
)
finally:
for t in helper_tasks:
t.cancel()
await asyncio.gather(*helper_tasks, return_exceptions=True)
answer = await run_team()
print(f"\n[lead final answer]\n{answer}\n\nTool-call summary:")
print_trace()
- --
第二部分 · 异步子代理（动态生成）
现在给主子代理工具和一个简单的客户端 sleep(seconds) 工具。主代理生成三个辅助代理；每个辅助代理休眠 *N* 秒，通过 send_message 向主代理发送“完成”消息，然后 wait_for_message 等待进一步指令。生成调用立即返回；主代理在它们运行时检查 get_status，通过 wait_for_message 收集它们的报告，然后 kill_subagents 终止所有三个以解散它们。不涉及领域逻辑——仅完整的生成/状态/收集/终止路径。
SLEEP = {
"name": "sleep",
"description": "Sleep for the given number of seconds, then return.",
"input_schema": {
"type": "object",
"properties": {"seconds": {"type": "integer", "minimum": 0, "maximum": 10}},
"required": ["seconds"],
},
}
SUBAGENT_TOOLS = [
{
"name": "create_subagents",
"description": "Spawn helper subagents. Returns immediately — helpers run concurrently in the "
"background in parallel with you. Each gets base_instruction, optionally + "
"per_subagent_instructions[i].",
"input_schema": {
"type": "object",
"properties": {
"base_instruction": {"type": "string"},
"per_subagent_instructions": {
"type": "array",
"items": {"type": "string"},
"maxItems": 10,
},
},
"required": ["base_instruction"],
},
},
{
"name": "get_status",
"description": "Status of every helper (active / idling / done / crashed).",
"input_schema": {"type": "object", "properties": {}},
},
{
"name": "kill_subagents",
"description": "Cancel running helpers you no longer need.",
"input_schema": {
"type": "object",
"properties": {
"subagent_ids": {"type": "array", "items": {"type": "string"}, "minItems": 1}
},
"required": ["subagent_ids"],
},
},
]
async def dispatch_sleep(block) -> str:
s = int(block.input["seconds"])
await asyncio.sleep(s)
return f"slept {s}s"
async def run_spawn_lead() -> str:
hub = Hub()
hub.register("lead")
helpers: dict[str, asyncio.Task] = {}
async def _create(block):
base = block.input["base_instruction"]
per = block.input.get("per_subagent_instructions") or [""]
spawned = []
for suffix in per:
h = hub.new_name()
helpers[h] = asyncio.create_task(
run_agent(
hub,
h,
system=f"You are {h}, a helper.",
first_user_turn=f"{base}\n\n{suffix}".strip(),
tools=[SLEEP, *BASE_TOOLS],
extra_dispatch={"sleep": dispatch_sleep},
)
)
spawned.append(h)
return f"spawned: {', '.join(spawned)}"
async def _status(block):
return "\n".join(f"{n}: {s}" for n, s in hub.status.items() if n != "lead") or "(none)"
async def _kill(block):
killed_ids, to_await = [], []
for sid in block.input["subagent_ids"]:
if sid in helpers and not helpers[sid].done():
helpers[sid].cancel()
hub.status[sid] = "done"
to_await.append(helpers.pop(sid))
killed_ids.append(sid)
await asyncio.gather(*to_await, return_exceptions=True)
return f"cancelled: {', '.join(killed_ids)}" if killed_ids else "no matching active helpers"
try:
return await run_agent(
hub,
"lead",
system="You are the lead.",
first_user_turn=(
"Spawn three helper agents. Instruct each to sleep a different number of seconds "
"(1, 2, 3), report back to you 'done, slept Ns', then wait for your further "
"instructions. After that, check that they're running, collect all three reports, "
"then dismiss all three helpers. Finish with a one-line summary."
),
tools=[*SUBAGENT_TOOLS, *BASE_TOOLS],
extra_dispatch={
"create_subagents": _create,
"get_status": _status,
"kill_subagents": _kill,
},
)
finally:
for t in helpers.values():
t.cancel()
await asyncio.gather(*helpers.values(), return_exceptions=True)
answer = await run_spawn_lead()
print(f"\n[lead final answer]\n{answer}\n\nTool-call summary:")
print_trace()
后续步骤
替换为你自己的领域工具——将它们添加到 tools 中，并在 extra_dispatch 中添加一个处理器。上面的 Hub、run_agent 以及两个团队运行器就是你所需要的全部；其余部分是你的任务和你的工具。
有关完整的工具定义参考，请参阅工具使用指南。
常见问题
常见错误包括：1) 忘记设置ANTHROPIC_API_KEY环境变量，导致认证失败；2) 模型名称错误，应使用可用的最新模型；3) 工具调用未在extra_dispatch中注册，导致'no dispatch'错误；4) 子代理未正确取消，导致资源泄漏；5) 消息未送达，可能因为收件人未注册或名称拼写错误。排查方法：检查环境变量、模型名称、工具定义和调度器映射；使用print_trace查看工具调用序列；确保所有智能体已注册。生产环境优化：使用连接池、超时控制、错误重试、日志记录和监控；考虑使用消息队列（如Redis）替代内存Hub以支持分布式部署。
企业应用扩展
开发者可以替换领域工具，将自定义工具添加到tools列表，并在extra_dispatch中提供处理函数。例如，添加网络搜索、数据库查询或文件操作工具。可以扩展Hub支持持久化消息、优先级队列或广播。也可以修改run_agent以支持更复杂的停止条件、多轮对话或人类审批。此外，可以结合LangChain或CrewAI等框架，但本教程提供了底层实现，便于深入理解。
总结
本教程基于Anthropic Python SDK和asyncio，深入剖析Claude Opus 4.8多智能体编排的两种核心模式：固定N智能体团队和异步子智能体。通过构建消息中心（Hub）和基础代理循环，实现智能体间的异步消息传递与动态生成。教程提供完整可运行的代码，展示工具调用顺序、消息流和状态管理，并指导开发者替换领域工具以扩展应用。适合有编程基础、希望掌握AI工程实践的开发者，帮助理解多智能体系统的架构设计与实现细节。
- --
内容来源与验证
- 内容等级：PREMIUM_A
- 质量等级：A
- 难度：中级
- 验证等级：V1
- 原文作者：paulchen-go
- 原文日期：2026-06-08
- 来源许可：MIT
- 来源版本：
f36ea8f8347adbb095b698c95a9f2a95c8e980a1 - 查看原文
站内还有更多完整实战教程
加入会员，解锁全部教程、完整代码与持续更新内容。
加入会员