从零手写一个轻量级 Multi-Agent 框架:关键层面与实现指南

更新: 9/19/2026 字数: 0 字 时长: 0 分钟

开篇:你真的需要一个"重"框架吗?

做 Agent 开发的前端同学,大概都经历过这样的纠结:想做一个"多角色协作"的功能——比如让一个 Agent 负责查资料、一个负责写代码、一个负责审查——于是打开 LangGraph、AutoGPT、CrewAI 的文档,结果发现:概念一大堆(State Graph、Node、Edge、Crew、Flow……),依赖装一堆,想改个调度逻辑要翻半天源码,最后发现自己 80% 的功能都用不上。

其实,Multi-Agent 框架的内核并不神秘。剥开那些花哨的封装,它本质上就是几个前端开发者天天在用的东西的组合:事件驱动(EventEmitter)、异步任务编排(Promise/async)、状态管理(类似 Redux 的 store)、消息传递(类似 postMessage)。

自己手写一个轻量框架的价值在于:

• 完全可控:调度策略、通信协议想怎么改怎么改,不用跟框架的抽象层较劲;
• 吃透原理:理解了自己写的这套,再看任何大框架都是"换皮";
• 项目轻量:一个几百行的核心模块,零重依赖,打包体积友好,尤其适合前端/边缘环境。

这篇文章,我们就用 TypeScript,从通信、调度、状态、协作、容错五个关键层面,一步步搭出一个能跑的 Multi-Agent 框架。

为什么要多个 Agent? 一个"全能 Agent"要同时扮演搜索、编码、审查多种角色,prompt 会越写越臃肿,上下文互相干扰,效果反而差。拆成多个职责单一的 Agent,各自 prompt 精简、工具聚焦,再由一个协调机制串起来——这就是分而治之在 Agent 领域的体现。

一、整体架构:五个核心层面

先看全局。我们要搭的框架,由五个部分组成:

• Agent:最小工作单元,封装一个角色、一套工具、一段 prompt;
• MessageBus(消息总线):Agent 之间通信的管道,解耦收发双方;
• Orchestrator(调度器):任务分配与流程编排的大脑;
• SharedState(共享状态):全局黑板,存放任务进展与中间产物;
• 容错层:重试、降级、监控,贯穿以上所有模块。

下面逐层拆解。

二、层面一:Agent 定义——最小工作单元

Agent 的本质,是"输入消息 → 调用 LLM/工具 → 输出结果"的一个封装。我们先定义接口:

interface AgentContext {
  bus: MessageBus;
  state: SharedState;
}

interface Message {
  id: string;
  from: string;          // 发送方 agent id
  to: string | 'broadcast'; // 接收方,或广播
  type: string;          // 消息类型:'task' | 'result' | 'error' ...
  payload: any;
  timestamp: number;
}

abstract class BaseAgent {
  constructor(
    public readonly id: string,
    public readonly role: string,          // 角色描述,用于 prompt
    protected capabilities: string[] = [], // 能力标签,用于调度匹配
  ) {}

  // 每个 Agent 必须实现自己的处理逻辑
  abstract handle(msg: Message, ctx: AgentContext): Promise<any>;

  canHandle(taskType: string): boolean {
    return this.capabilities.includes(taskType);
  }
}

一个具体的 Agent(比如调用 LLM 的研究员)长这样:

class ResearchAgent extends BaseAgent {
  constructor() {
    super('researcher', '资料研究员', ['search', 'summarize']);
  }

  async handle(msg: Message, ctx: AgentContext): Promise<any> {
    const { query } = msg.payload;
    // 伪代码:调用 LLM + 搜索工具
    const result = await llm.invoke(
      `你是${this.role}。请检索并总结:${query}`
    );
    // 把结果写入共享状态,供其他 Agent 读取
    ctx.state.set(`research:${query}`, result);
    return result;
  }
}

设计要点:capabilities 能力标签是调度的关键——调度器靠它把任务派给对的 Agent(后面详述)。Agent 之间不直接互相调用方法,而是通过 ctx.bus 和 ctx.state 交互,保持解耦。

三、层面二:通信机制——消息总线

Agent 之间不能硬编码地互相调用(否则耦合爆炸)。前端开发者最熟悉的解耦方案就是发布/订阅(pub/sub)——本质上和 DOM 的 addEventListener、Node 的 EventEmitter、Vue 的 EventBus 是一回事。

实现一个极简消息总线:

type Handler = (msg: Message) => void | Promise<void>;

class MessageBus {
  private subscribers = new Map<string, Set<Handler>>();

  // 订阅某种消息类型
  subscribe(type: string, handler: Handler): () => void {
    if (!this.subscribers.has(type)) this.subscribers.set(type, new Set());
    this.subscribers.get(type)!.add(handler);
    // 返回取消订阅函数(React 开发者会很熟悉这个模式)
    return () => this.subscribers.get(type)?.delete(handler);
  }

  // 发布消息
  async publish(msg: Message): Promise<void> {
    const handlers = this.subscribers.get(msg.type) ?? new Set();
    // 并发触发所有订阅者
    await Promise.allSettled([...handlers].map((h) => h(msg)));
  }
}

为什么用消息总线而不是直接调用? 三个好处:①解耦——新增/删除 Agent 不影响其他人;②可观测——所有消息流经总线,天然是埋点、日志、回放的最佳位置;③易扩展——想加"消息持久化""跨进程通信(WebSocket/Worker)",只需改总线实现,上层无感。

四、层面三:任务调度——把活派给对的人

调度器(Orchestrator)是框架的大脑,决定"什么任务、什么时候、交给谁"。常见有三种调度模式:

模式	说明	适用场景
顺序流水线	A → B → C 按固定顺序	流程明确(研究→编码→审查)
能力路由	按任务类型匹配 Agent 能力	任务类型多样、动态
中心化编排	由一个 LLM "主管"决定下一步	复杂、需动态规划

我们实现一个"能力路由 + 顺序流水线"的混合调度器:

interface Task {
  id: string;
  type: string;      // 任务类型,用于匹配 Agent 能力
  payload: any;
  dependsOn?: string[]; // 依赖的前置任务 id
}

class Orchestrator {
  private agents: BaseAgent[] = [];
  constructor(private bus: MessageBus, private state: SharedState) {}

  register(agent: BaseAgent) {
    this.agents.push(agent);
  }

  // 能力路由:找到能处理该任务的 Agent
  private route(task: Task): BaseAgent | undefined {
    return this.agents.find((a) => a.canHandle(task.type));
  }

  // 执行一批任务,自动处理依赖顺序
  async run(tasks: Task[]): Promise<Map<string, any>> {
    const results = new Map<string, any>();
    const done = new Set<string>();

    // 拓扑排序:依赖满足才执行
    while (done.size < tasks.length) {
      const ready = tasks.filter(
        (t) => !done.has(t.id) &&
               (t.dependsOn ?? []).every((d) => done.has(d))
      );
      if (ready.length === 0) throw new Error('存在循环依赖或无法满足的依赖');

      // 无依赖关系的任务并发执行
      await Promise.all(
        ready.map(async (task) => {
          const agent = this.route(task);
          if (!agent) throw new Error(`无 Agent 可处理任务类型: ${task.type}`);

          const msg: Message = {
            id: crypto.randomUUID(),
            from: 'orchestrator', to: agent.id,
            type: task.type, payload: task.payload,
            timestamp: Date.now(),
          };
          const res = await agent.handle(msg, { bus: this.bus, state: this.state });
          results.set(task.id, res);
          done.add(task.id);
        })
      );
    }
    return results;
  }
}

这里的精髓:用 dependsOn 声明任务依赖,调度器做拓扑排序——没有依赖关系的任务用 Promise.all 并发跑,有依赖的等前置完成。这套逻辑前端同学应该很熟,和构建工具(Webpack/Vite)处理模块依赖图、或任务运行器(gulp)编排任务是同一个思路。

五、层面四:状态管理——共享黑板

多个 Agent 协作,需要一块"公共黑板"来共享中间产物——这就是黑板模式(Blackboard Pattern)。前端开发者可以直接把它类比成 Redux/Zustand 的全局 store:单一数据源,谁都能读,写入可被订阅。

class SharedState {
  private data = new Map<string, any>();
  private listeners = new Set<(key: string, value: any) => void>();

  get<T = any>(key: string): T | undefined {
    return this.data.get(key);
  }

  set(key: string, value: any): void {
    this.data.set(key, value);
    // 通知订阅者(类似 store.subscribe)
    this.listeners.forEach((fn) => fn(key, value));
  }

  subscribe(fn: (key: string, value: any) => void): () => void {
    this.listeners.add(fn);
    return () => this.listeners.delete(fn);
  }

  snapshot(): Record<string, any> {
    return Object.fromEntries(this.data); // 便于调试/持久化
  }
}

关键设计考量:

• 不可变更新:复杂对象建议写入副本,避免多个 Agent 引用同一对象互相踩踏(和 React state 的不可变原则一致);
• 命名空间:用 research:xxx、code:xxx 这样的 key 前缀隔离不同 Agent 的数据;
• 可快照:snapshot() 让你能随时 dump 出完整状态,做调试、持久化或"时间旅行"回放。

六、层面五:错误处理——一个出错不拖垮全局

LLM 调用天然不稳定:超时、限流、格式错乱都可能发生。多 Agent 系统里,单个 Agent 挂掉不能拖垮整条链路。核心手段是 重试 + 降级 + 隔离:

interface RetryOptions {
  maxRetries: number;
  backoff: number; // 退避基数(ms)
}

async function withResilience<T>(
  fn: () => Promise<T>,
  opts: RetryOptions,
  fallback?: () => T | Promise<T>,
): Promise<T> {
  let lastErr: unknown;
  for (let i = 0; i <= opts.maxRetries; i++) {
    try {
      return await fn();
    } catch (err) {
      lastErr = err;
      if (i < opts.maxRetries) {
        // 指数退避:1s, 2s, 4s...(应对限流很有效)
        await new Promise((r) => setTimeout(r, opts.backoff * 2 ** i));
      }
    }
  }
  // 重试耗尽 → 降级方案兜底
  if (fallback) return await fallback();
  throw lastErr;
}

在调度器里包一层,让每个 Agent 的执行都具备韧性:

const res = await withResilience(
  () => agent.handle(msg, ctx),
  { maxRetries: 2, backoff: 1000 },
  () => ({ error: true, degraded: `${agent.id} 失败,返回降级结果` })
);

工程要点:

• 超时控制:用 Promise.race 给每个 Agent 加超时,防止单点卡死全局;
• 错误隔离:调度器用 Promise.allSettled 而非 Promise.all,让部分 Agent 失败时其他仍能完成;
• 熔断:某 Agent 连续失败 N 次就暂时"下线",避免雪崩;
• 监控:让容错层往 MessageBus 发 error 事件,统一收集告警。

七、完整可运行示例:研究→编码→审查流水线

把五个层面拼起来,做一个"用户提需求 → 研究员查资料 → 编码员写代码 → 审查员审查"的完整 Demo:

// ===== 三个 Agent(此处用 mock 代替真实 LLM 调用)=====
class ResearchAgent extends BaseAgent {
  constructor() { super('researcher', '研究员', ['research']); }
  async handle(msg: Message, ctx: AgentContext) {
    const findings = `关于"${msg.payload.topic}"的调研结论(mock)`;
    ctx.state.set('research:result', findings);
    return findings;
  }
}

class CoderAgent extends BaseAgent {
  constructor() { super('coder', '编码员', ['code']); }
  async handle(msg: Message, ctx: AgentContext) {
    const research = ctx.state.get('research:result'); // 读取上游产物
    const code = `// 基于调研:${research}\nfunction demo() { return 42; }`;
    ctx.state.set('code:result', code);
    return code;
  }
}

class ReviewAgent extends BaseAgent {
  constructor() { super('reviewer', '审查员', ['review']); }
  async handle(msg: Message, ctx: AgentContext) {
    const code = ctx.state.get('code:result');
    return `审查通过 ✓ (代码长度 ${code?.length})`;
  }
}

// ===== 组装框架 =====
async function main() {
  const bus = new MessageBus();
  const state = new SharedState();
  const orchestrator = new Orchestrator(bus, state);

  // 可观测:监听所有状态变更
  state.subscribe((k, v) => console.log(`[状态更新] ${k} =`, String(v).slice(0, 40)));

  orchestrator.register(new ResearchAgent());
  orchestrator.register(new CoderAgent());
  orchestrator.register(new ReviewAgent());

  // 定义带依赖的任务流水线
  const tasks: Task[] = [
    { id: 't1', type: 'research', payload: { topic: '如何实现防抖函数' } },
    { id: 't2', type: 'code',     payload: {}, dependsOn: ['t1'] },
    { id: 't3', type: 'review',   payload: {}, dependsOn: ['t2'] },
  ];

  const results = await orchestrator.run(tasks);
  console.log('\n最终结果:', results.get('t3'));
}

main();

运行后会依次输出研究、编码、审查三个阶段的状态更新和最终结论。整个框架核心不到 200 行,却完整覆盖了通信、调度、状态、依赖编排、容错——这就是"轻量"的意义。

八、拓展:对比、优化与演进

8.1 与主流框架的对比

维度	手写轻量框架	LangGraph	CrewAI
学习成本	低(都是熟悉的 JS 概念)	中高(图抽象)	中
可控性	完全可控	受框架约束	受角色抽象约束
生态/工具	需自建	丰富	中等
适用规模	中小型、定制化	复杂有状态流程	角色协作场景

结论:原型验证、学习原理、中小型定制项目,手写完全够用且更灵活;当流程复杂到需要条件分支、循环、检查点回溯时,再考虑迁移到成熟框架。

8.2 性能优化技巧

• 并发而非串行:无依赖任务务必 Promise.all 并发,别一个个 await;
• 流式输出:Agent 的 LLM 结果用流式(SSE / ReadableStream)返回,改善前端体感;
• 结果缓存:相同输入的 Agent 调用结果可缓存(Map / IndexedDB),避免重复烧 token;
• 上下文裁剪:只把 Agent 真正需要的 state 片段传给它,别把整个黑板塞进 prompt。

8.3 前端环境的限制与解法

限制	解法
API Key 不能暴露在前端	Agent 的 LLM 调用走后端代理(BFF),前端只编排
主线程被密集计算阻塞	把 Agent 逻辑放进 Web Worker,消息总线用 postMessage
浏览器无长驻进程	长任务交给后端;前端只做轻量编排与展示
跨标签页/多端协同	消息总线底层换成 WebSocket / BroadcastChannel

其中 Web Worker 尤其值得一提:我们的 MessageBus 抽象天然适配——把 publish/subscribe 的底层从内存 Map 换成 worker.postMessage + onmessage 即可,上层 Agent 代码几乎不用动。这就是前面强调"通信解耦"的回报。

九、收尾:关键决策清单与常见问题

开发前的决策清单

1. 调度模式:流程固定选顺序流水线,类型多变选能力路由,需动态规划选 LLM 主管编排;
2. 通信方式:单进程用内存总线,跨 Worker/多端用 postMessage/WebSocket;
3. 状态粒度:黑板存"共享产物",Agent 私有数据不要往黑板塞;
4. 容错级别:关键路径必须有重试 + 降级,非关键路径可"尽力而为";
5. 可观测性:一开始就把日志/埋点挂在消息总线上,别等出问题才补。

常见问题速查

问题	排查方向
任务卡住不动	检查 dependsOn 是否有循环依赖;是否缺超时控制
Agent 拿不到上游数据	检查 state 的 key 命名是否一致;写入是否在读取之前
部分 Agent 失败拖垮全局	把 Promise.all 换成 allSettled;加错误隔离
LLM 频繁限流报错	加指数退避重试;控制并发数
结果不稳定/难复现	用 state.snapshot() 做回放调试;固定随机种子/temperature

一句话总结:Multi-Agent 框架的核心,不过是把"事件总线 + 异步编排 + 全局 store + 重试降级"这几样前端老熟人组合起来。理解了通信、调度、状态、协作、容错这五个层面,你不仅能手写一个够用的轻量框架,更能看透任何大框架的底层套路——框架是工具,原理才是你的核心竞争力。