夜雨聆风
从ClaudeCode源码中衍生出的多Agent开发框架
这个框架解决什么问题
一句话：让多个 LLM Agent 能像一个团队一样协作完成任务。
单 Agent 的场景大家都熟了——给 LLM 挂几个工具，让它一步步推理完成任务。但一旦问题变复杂（”帮我设计一个 REST API，写代码，跑测试，做 Code Review”），单 Agent 的 prompt 就变得又臭又长，效果也不好。
更自然的思路是拆成多个角色：architect 负责设计、developer 负责写码、reviewer 负责审查，各司其职。open-multi-agent 就是干这事的——提供一套协调机制，让多个 Agent 按照依赖关系协作，共享上下文，自动调度。
运行时依赖只有三个：@anthropic-ai/sdk、openai、zod，没有额外的 runtime 负担。
三种运行模式
框架的入口类叫 OpenMultiAgent，暴露了三个核心方法：
#1. runAgent —— 单 Agent 一次性执行
最简单的用法，一个 Agent 执行一个任务：
const orchestrator = newOpenMultiAgent({
defaultModel: 'claude-sonnet-4-6',
})
const result = await orchestrator.runAgent(
{
name: 'coder',
model: 'claude-sonnet-4-6',
systemPrompt: '你是一个 TypeScript 开发者。',
tools: ['bash', 'file_read', 'file_write'],
maxTurns: 8,
},
'写一个 greet 函数并运行它'
)
console.log(result.output)
console.log(`消耗 token：${result.tokenUsage.input_tokens} 输入 / ${result.tokenUsage.output_tokens} 输出`)
这层其实就是对 Agent 做了一层封装，帮你管好工具注册、LLM 适配器创建这些琐事。
#2. runTeam —— 自动编排（核心功能）
这是整个框架最有意思的部分。你只需要给一个目标和一组 Agent，框架会自动拆解任务：
const architect = {
name: 'architect',
model: 'claude-sonnet-4-6',
systemPrompt: '你是一个软件架构师，负责设计 API 契约和目录结构。',
tools: ['bash', 'file_write'],
}
const developer = {
name: 'developer',
model: 'claude-sonnet-4-6',
systemPrompt: '你是一个 TypeScript 开发者，根据架构师的设计进行实现。',
tools: ['bash', 'file_read', 'file_write', 'file_edit'],
}
const reviewer = {
name: 'reviewer',
model: 'claude-sonnet-4-6',
systemPrompt: '你是一个高级代码审查员。',
tools: ['bash', 'file_read', 'grep'],
}
const team = orchestrator.createTeam('api-team', {
name: 'api-team',
agents: [architect, developer, reviewer],
sharedMemory: true,
})
const result = await orchestrator.runTeam(team, '创建一个 Express.js REST API')
调用 runTeam() 之后发生的事情其实不少，后面会展开。
#3. runTasks —— 手动定义任务流水线
如果你不想让 LLM 来拆任务（有时候你确实知道该怎么拆），可以自己定义：
const tasks = [
{
title: '设计数据模型',
description: '设计一个 URL 缩短服务的数据模型',
assignee: 'designer',
},
{
title: '实现核心逻辑',
description: '根据设计文档实现 URL 缩短服务',
assignee: 'implementer',
dependsOn: ['设计数据模型'], // 依赖第一个任务
},
{
title: '编写测试',
description: '运行并测试实现的服务',
assignee: 'tester',
dependsOn: ['实现核心逻辑'],
},
]
const result = await orchestrator.runTasks(team, tasks)
dependsOn 用任务标题做引用。框架内部会做拓扑排序（Kahn 算法），自动解析出执行顺序。没有依赖关系的任务默认并行跑。
Coordinator 模式：runTeam 的内部流程
runTeam 的实现是整个框架最核心的逻辑，值得详细拆：
第一步：创建 Coordinator Agent
框架临时创建一个 “coordinator” 角色，把你的目标和所有可用 Agent 的信息都塞给它，让它生成一个 JSON 任务数组：
[
{"title":"设计 API","description":"...","assignee":"architect","dependsOn":[]},
{"title":"实现功能","description":"...","assignee":"developer","dependsOn":["设计 API"]},
{"title":"代码审查","description":"...","assignee":"reviewer","dependsOn":["实现功能"]}
]
第二步：任务入队 + 调度
解析出来的任务进入 TaskQueue，Scheduler 负责给未指定 assignee 的任务分配 Agent。
第三步：按轮次执行
执行是分轮次进行的：
- 取出所有 pending 状态的任务
- 每个任务组装 prompt（包含任务描述 + 共享内存中其他 Agent 的产出）
- 通过 AgentPool 并发执行
- 完成的任务写入 SharedMemory，自动解锁下游依赖
- 重新调度，循环直到没有 pending 任务
第四步：综合产出
所有任务跑完后，coordinator 再来一次 LLM 调用，把所有任务结果综合成最终答案。
整个过程有两个关键设计：
- 失败隔离：一个任务失败只会阻塞它的直接下游，不影响其他无关任务。
- 共享内存注入：每个 Agent 跑之前，都能看到之前其他 Agent 产出的摘要，这就是多 Agent 协作中的”上下文传递”。
Agent 的执行循环
单看一个 Agent 的运行过程（在 AgentRunner 里实现），其实就是一个经典的 tool-use 循环：
while (true) {
1. 发消息给 LLM
2. 检查响应中有没有 tool_use block
3. 如果没有 → 结束
4. 如果有 → 并行执行所有工具调用
5. 把工具结果追加到对话历史
6. 回到 1
}
有 maxTurns 兜底，防止死循环。每一轮的 token 消耗、工具调用记录都会累积。
工具的并行执行用了一个自己实现的 Semaphore（信号量），默认最多 4 个工具同时跑。并发控制在 Node.js 单线程模型里用 Promise 队列就够了，不需要 mutex 那些。
工具系统的设计
定义一个工具只需要名字、描述、Zod schema、执行函数：
import { z } from'zod'
import { defineTool } from'open-multi-agent'
const fetchTool = defineTool({
name: 'fetch_data',
description: '从 URL 获取 JSON 数据',
inputSchema: z.object({
url: z.string().url(),
}),
execute: async ({ url }) => {
const res = awaitfetch(url)
return { data: await res.text() }
},
})
框架内部维护一个 ToolRegistry，所有注册的工具在发给 LLM 之前，会把 Zod schema 转成 JSON Schema（框架自己写了一个 zodToJsonSchema 转换器，递归处理对象、数组、联合类型等）。
工具执行时，先用 Zod 校验输入（LLM 生成的参数不一定靠谱），再执行。执行过程中的任何异常都被捕获，返回 { data, isError: true }，不会让整个流程崩掉。
框架内置了 5 个常用工具：bash（跑命令）、file_read、file_write、file_edit（行级别编辑）、grep（正则搜索文件）。
LLM 适配器：多模型支持
适配器层的接口非常简洁：
interfaceLLMAdapter {
chat(messages, options): Promise<LLMResponse>
stream(messages, options): AsyncIterable<StreamEvent>
}
目前实现了四个：Anthropic、OpenAI、Grok、GitHub Copilot。
Anthropic 和 OpenAI 的消息格式差异不小（Anthropic 用 content block 数组，OpenAI 用 tool_calls 字段），但适配器层把这些都抹平了。上层代码完全不用关心底层用的是哪个模型。
用工厂函数按需加载，不用 Anthropic 就不会加载它的 SDK：
const adapter = createAdapter('anthropic', { model: 'claude-sonnet-4-6' })
// 或者
const adapter = createAdapter('openai', {
model: 'gpt-4o',
baseURL: 'http://localhost:11434/v1', // 兼容 Ollama
})
调度策略
Scheduler 支持四种调度策略，通过 autoAssign() 给未分配的任务指定 Agent：
- dependency-first（默认）：优先调度阻塞下游最多的任务，用 BFS 统计反向依赖数量。关键路径上的任务先执行，减少整体等时。
- round-robin：轮询分配。
- least-busy：分给当前任务最少的 Agent。
- capability-match：关键词匹配——把任务描述和 Agent 的 systemPrompt 做分词，算重叠度。
实际使用中 dependency-first 基本够用了。
共享内存：Agent 之间怎么传递上下文
这是多 Agent 协作绕不开的问题：下游 Agent 怎么知道上游 Agent 都做了什么？
框架的方案是 命名空间化的 Key-Value 存储：
researcher/findings → "TypeScript 5.5 引入了 const type parameters…"
coder/plan → "基于 const type parameters 实现功能 X…"
写入时用 agentName/key 作为键，天然隔离。读取时可以按 Agent 名过滤，也可以用 getSummary() 生成一份 Markdown 格式的摘要：
## Shared Team Memory
### researcher
- findings: TypeScript 5.5 引入了 const type parameters…
### coder
- plan: 基于 const type parameters 实现功能 X…
这份摘要会被注入到后续 Agent 的 prompt 里，让它们能”看到”队友的成果。
底层用的是一个普通的 Map，预留了 MemoryStore 接口，想换 Redis 或 SQLite 也行。
结构化输出
有时候需要 Agent 返回特定格式的 JSON（比如做数据分析），框架支持通过 Zod schema 进行校验：
constReviewAnalysis = z.object({
summary: z.string(),
sentiment: z.enum(['positive', 'negative', 'neutral']),
confidence: z.number().min(0).max(1),
keyTopics: z.array(z.string()),
})
const analyst = {
name: 'analyst',
model: 'claude-sonnet-4-6',
systemPrompt: '你是一个产品评价分析师。',
outputSchema: ReviewAnalysis,
}
const result = await orchestrator.runAgent(analyst, '分析这条评价：...')
if (result.structured) {
// result.structured 是类型安全的
console.log(result.structured.sentiment) // 'positive' | 'negative' | 'neutral'
}
内部实现：把 schema 描述注入 system prompt，让 LLM 返回 JSON；收到响应后用 Zod parse；如果校验失败，自动重试一次并把错误信息反馈给 LLM。这个”一次重试”的策略比较实用——大多数情况下一次重试就能修正格式问题，而不会陷入无限循环。
任务重试
网络波动、LLM 偶发错误这些都是常事。框架支持任务级别的自动重试：
{
title: '获取数据',
assignee: 'fetcher',
maxRetries: 2,
retryDelayMs: 500, // 基础延迟 500ms
retryBackoff: 2, // 指数退避因子
}
计算公式：baseDelay × backoff^(attempt - 1)，封顶 30 秒。每次重试的 token 消耗会累加到总用量里。
可观测性
框架提供了一个 onTrace 回调，能拿到四种结构化事件：
const orchestrator = newOpenMultiAgent({
defaultModel: 'claude-sonnet-4-6',
onTrace: (event) => {
switch (event.type) {
case'llm_call':
// 每次 LLM API 调用：模型、turn 数、token 消耗、耗时
break
case'tool_call':
// 每次工具执行：工具名、是否出错、耗时
break
case'task':
// 任务生命周期：成功/失败、重试次数、耗时
break
case'agent':
// Agent 整体：总 turn 数、工具调用数、总 token、耗时
break
}
},
})
所有事件都带 runId，同一次运行的事件可以关联起来。你可以把这些事件接到自己的日志系统、OpenTelemetry 或者 Dashboard 里。
并发控制
整个框架有两层独立的并发控制，都用自己实现的 Semaphore：
- AgentPool 层：控制同时跑多少个 Agent，默认 5。
- ToolExecutor 层：控制单个 Agent 内同时执行多少个工具调用，默认 4。
Semaphore 的实现很标准——一个计数器加一个 Promise 队列，acquire() 拿不到就排队，release() 唤醒队首。Node.js 单线程不需要真正的锁，用事件循环的微任务调度就够了。
审批门控
runTeam 和 runTasks 都支持轮次间审批——每批任务完成后暂停，等人确认再继续：
const result = await orchestrator.runTeam(team, goal, {
onApproval: async (completedTasks, nextPending) => {
console.log('已完成:', completedTasks.map(t => t.title))
console.log('待执行:', nextPending.map(t => t.title))
returnconfirm('继续执行？') // 返回 false 则跳过所有剩余任务
},
})
返回 false 时，所有未执行的任务都被标记为 skipped，不会继续跑。
架构总览
把各层关系画出来大概长这样：
┌─────────────────────────────────────────────┐
│ OpenMultiAgent（对外 API） │
│ runAgent() / runTeam() / runTasks() │
└─────────────┬───────────────────────────────┘
│
┌─────────────▼───────────────────────────────┐
│ 编排层：Scheduler + TaskQueue + AgentPool │
│ 任务拆解 → 拓扑排序 → 调度分配 → 并发执行 │
└─────────────┬───────────────────────────────┘
│
┌─────────────▼───────────────────────────────┐
│ Agent 层：AgentRunner 对话循环 │
│ LLM 调用 → 工具提取 → 并行执行 → 追加结果 │
└──────┬──────────────┬───────────┬───────────┘
│ │ │
┌────▼────┐ ┌─────▼────┐ ┌──▼──────────┐
│ LLM 适配 │ │ 工具系统 │ │ 共享内存 │
│ Anthropic│ │ 注册/校验 │ │ 命名空间 KV │
│ OpenAI │ │ 并发执行 │ │ 摘要注入 │
│ Grok │ │ 错误兜底 │ │ │
└──────────┘ └──────────┘ └─────────────┘
值得借鉴的设计
读完这个项目，有几个设计决策觉得不错：
1. 所有类型定义集中在 types.ts
整个框架的接口、类型全部放在一个文件里(types.ts)，显式避免循环依赖。在中小规模项目里这比分散在各模块里要清晰。
2. 工具错误永远不抛异常
工具执行出错不 throw，而是返回 { isError: true }，让 LLM 自己判断怎么处理。这个设计很重要——如果工具一报错就崩掉，整个 Agent 循环就断了。LLM 其实有能力基于错误信息调整策略。
3. 适配器懒加载
createAdapter() 内部用动态 import，不用 Anthropic 就不加载它的 SDK。对于只用一个 provider 的场景，能减少启动时间和包体积。
4. 两层独立的信号量
Agent 并发和工具并发分开控制，互不干扰。你可以允许 5 个 Agent 同时跑，但每个 Agent 内部最多 4 个工具并发。粒度控制得比较合理。
5. Coordinator 本身也是一个 Agent
它不是硬编码的逻辑，而是用 LLM 来拆解任务。好处是灵活性高，坏处是你对拆解结果的控制力弱了（这也是为什么 runTasks 存在的原因）。
一些局限
客观说几个不足：
- 内存存储不持久化。当前的 SharedMemory 和 MessageBus 都是内存里的 Map，进程一结束就没了。虽然预留了 MemoryStore 接口，但还没有现成的持久化实现。
- 没有 streaming 级的团队编排。
runTeam的返回是等所有任务跑完之后一次性给结果，中间过程只能通过onProgress回调观察。如果想做实时的 streaming 体验，需要自己在 Agent 层用stream()方法组装。 - Coordinator 的任务拆解依赖 LLM 能力。模型能力弱的时候，拆出来的任务质量可能不理想。实际生产中可能更倾向
runTasks手动编排。
怎么跑起来
git clone https://github.com/JackChen-me/open-multi-agent.git
cd open-multi-agent
npm install
npm run build
# 跑单 Agent 示例
export ANTHROPIC_API_KEY=your-key
npx tsx examples/01-single-agent.ts
# 跑团队协作示例
npx tsx examples/02-team-collaboration.ts
写在最后
从工程实践的角度看，open-multi-agent 是一个干净的 TypeScript 多 Agent 框架。代码量不大，没有过度抽象，每个模块职责清晰。如果你正在考虑在 Node.js/TypeScript 技术栈里搞多 Agent 系统，不论是直接用还是参考它的设计，都值得花时间读一读源码。
比起直接用一些大而全的框架，理解底层怎么转的，对写出可靠的 Agent 系统更有帮助。