Multi-Agent

问题：单 Agent 上下文窗口有限 + 角色混乱 + 无法并行

单 Agent：一个 Agent 处理所有任务，prompt 越来越长，行为不可控

解决：专业化分工 + 并行执行 + 错误隔离

Multi-Agent：Supervisor 分配任务 → 各 Agent 专注执行 → 结果汇总

┌───────────────────────────────────────┐
│         Pipeline（流水线）              │  ← A → B → C 串行
└─────────────────┬─────────────────────┘
                  │ 适用：有先后依赖的任务
                  ▼
┌───────────────────────────────────────┐
│       Fan-out/Fan-in（扇出/扇入）       │  ← A → [B, C, D] → E 并行
└─────────────────┬─────────────────────┘
                  │ 适用：独立任务并行执行
                  ▼
┌───────────────────────────────────────┐
│         Supervisor（主管模式）          │  ← Router → [Agent A, B, C] 动态分配
└─────────────────┬─────────────────────┘
                  │ 适用：任务分配、智能路由
                  ▼
┌───────────────────────────────────────┐
│           Debate（辩论模式）            │  ← 多 Agent 对同一问题给出不同观点
└───────────────────────────────────────┘
                  │ 适用：代码审查、技术决策

// 1. Pipeline（流水线）：A → B → C 串行
const pipeline = new StateGraph()
  .addNode("parse", parseNode)
  .addNode("code", codeNode)
  .addNode("test", testNode)
  .addEdge(START, "parse")
  .addEdge("parse", "code")
  .addEdge("code", "test")
  .compile();

// 2. Fan-out/Fan-in：A → [B, C, D] → E 并行
const fanout = new StateGraph()
  .addNode("analyze", analyzeNode)
  .addNode("chunk1", chunk1Node)
  .addNode("chunk2", chunk2Node)
  .addNode("merge", mergeNode)
  .addConditionalEdges("analyze", (state) => [
    new Send("chunk1", state),
    new Send("chunk2", state),
  ])
  .compile();

// 3. Supervisor（主管模式）：Router → [Agent A, B, C]
const supervisor = new StateGraph()
  .addNode("router", routerNode)
  .addNode("coder", coderAgent)
  .addNode("reviewer", reviewerAgent)
  .addConditionalEdges("router", (state) => {
    if (state.task === "code") return "coder";
    if (state.task === "review") return "reviewer";
    return END;
  })
  .compile();

// 4. Debate（辩论模式）：多 Agent 对同一问题给出不同观点
const debate = new StateGraph()
  .addNode("agent1", agent1Node)
  .addNode("agent2", agent2Node)
  .addNode("judge", judgeNode)
  .addEdge("agent1", "judge")
  .addEdge("agent2", "judge")
  .compile();

任务特征分析：
├── 有明确的先后依赖？ → Pipeline
├── 子任务可以并行？ → Fan-out/Fan-in
├── 需要动态分配？ → Supervisor
├── 需要多视角输入？ → Debate / Collaborative
├── 需要分层管理？ → Hierarchical
└── 混合场景？ → 组合使用（顶层 Supervisor + 中层 Collaborative + 底层 Pipeline）

// 方式 1：共享 State（LangGraph 的方式，最常用）
const GraphState = Annotation.Root({
  input: Annotation,
  results: Annotation({
    reducer: (a, b) => a.concat(b),
    default: () => [],
  }),
});

// Agent A 写入 state
function agentA(state) {
  return { results: [resultA] };
}

// Agent B 读取 state 中 Agent A 的结果
function agentB(state) {
  const resultA = state.results[0];
  return { results: [processB(resultA)] };
}

// 方式 2：消息传递（独立进程的 Agent）
class Agent {
  constructor(name) {
    this.name = name;
    this.inbox = [];
  }
  send(target, message) {
    target.inbox.push({ from: this.name, content: message });
  }
  receive() {
    return this.inbox.shift();
  }
}

// 方式 3：事件总线（松耦合系统）
const eventBus = {
  listeners: {},
  on(event, callback) {
    if (!this.listeners[event]) this.listeners[event] = [];
    this.listeners[event].push(callback);
  },
  emit(event, data) {
    (this.listeners[event] || []).forEach((cb) => cb(data));
  },
};

// Token 开销分析
// 单 Agent：1 次 LLM 调用 = 1000 tokens 输入 + 500 tokens 输出 = 1500 tokens
// Multi-Agent（3 个 Agent 串行）：
//   Agent A：1000 + 500 = 1500 tokens
//   Agent B：1000 + 500(A 的输出) + 500 = 2000 tokens
//   Agent C：1000 + 500(B 的输出) + 500 = 2000 tokens
//   总计：5500 tokens（是单 Agent 的 3.7 倍）

// 优化：用摘要替代全文传递
async function summarizeAndPass(output) {
  const summary = await smallLLM.invoke(`用 50 字总结：${output}`);
  return summary; // 从 500 tokens 压缩到 50 tokens
}

// 1. 通信开销
// Agent 之间传递信息需要 tokens，多 Agent 系统的 token 消耗是单 Agent 的 N 倍
// 解决：用摘要替代全文传递、只传递必要信息

// 2. 错误传播
// A 的错误输出传给 B，B 基于错误信息做出错误决策
// 解决：每个 Agent 加输出验证 + Reviewer Agent

// 3. 死循环
// Agent A 认为应该让 B 处理，B 认为应该让 A 处理
// 解决：设置最大循环次数、Supervisor 决策

// 4. 一致性
// 多个 Agent 并行修改同一份数据，产生冲突
// 解决：用不可变 State + Reducer 合并

// 5. 成本控制
// 10 个 Agent 各调一次 LLM，成本是单 Agent 的 10 倍
// 解决：用弱模型做简单任务、缓存结果、按需调用

// 限制每个 Agent 的 token 预算
const agent = createReactAgent({
  llm: openai,
  tools: [searchTool],
  maxTokens: 4000, // 限制单次调用的 token 数
});

// 用弱模型处理简单任务
const simpleAgent = createReactAgent({
  llm: new ChatOpenAI({ model: "gpt-4o-mini" }), // 便宜的模型
  tools: [basicTool],
});

// 缓存重复查询（Prompt Cache）
// Anthropic 的 Prompt Cache 可以节省 90% 的输入 token 费用
// 优化方法：把稳定的 System Prompt 放在前面，变化的用户输入放在后面

问题：单层 Supervisor 无法处理大规模复杂任务

单层 Supervisor：一个主管管所有 Agent，prompt 越来越长，决策质量下降

解决：Hierarchical 分层管理，Collaborative 对等协作

Hierarchical：顶层 Supervisor → 中层 Supervisor → 底层 Agent
Collaborative：Agent A ↔ Agent B ↔ Agent C（无主管，对等协作）

// Hierarchical（层级模式）：多层 Supervisor
const hierarchicalGraph = new StateGraph()
  // 顶层 Supervisor：宏观决策
  .addNode("topSupervisor", topSupervisorNode)
  // 中层 Supervisor：具体分配
  .addNode("devSupervisor", devSupervisorNode)
  .addNode("qaSupervisor", qaSupervisorNode)
  // 底层 Agent：执行任务
  .addNode("coder", coderAgent)
  .addNode("reviewer", reviewerAgent)
  .addNode("tester", testerAgent)
  // 顶层分配到中层
  .addConditionalEdges("topSupervisor", (state) => {
    if (state.domain === "dev") return "devSupervisor";
    if (state.domain === "qa") return "qaSupervisor";
  })
  // 中层分配到底层
  .addConditionalEdges("devSupervisor", (state) => {
    if (state.task === "code") return "coder";
    if (state.task === "review") return "reviewer";
  })
  .compile();

// Collaborative（协作模式）：无主管，对等协作
const collaborativeGraph = new StateGraph()
  .addNode("agent1", agent1Node)
  .addNode("agent2", agent2Node)
  .addNode("agent3", agent3Node)
  .addNode("consensus", consensusNode)
  // 所有 Agent 并行执行
  .addConditionalEdges(START, () => [
    new Send("agent1", state),
    new Send("agent2", state),
    new Send("agent3", state),
  ])
  // 汇总所有 Agent 的输出，达成共识
  .addEdge("agent1", "consensus")
  .addEdge("agent2", "consensus")
  .addEdge("agent3", "consensus")
  .compile();

案例：企业级代码审查系统

顶层：Supervisor 模式
├── 根据 PR 的类型分配到不同团队（前端/后端/基础设施）

中层：Collaborative 模式
├── 前端团队内部：安全专家、性能专家、可读性专家并行审查
├── 后端团队内部：安全专家、性能专家、可读性专家并行审查

底层：Pipeline 模式
├── 每个专家的审查流程：读代码 → 分析问题 → 生成建议 → 格式化输出

汇总：Debate 模式
├── 多个专家的意见可能冲突，由仲裁 Agent 综合判断

问题：一个 Agent 失败可能影响整个工作流

没有容错：A 失败 → B 等待 → 整个流程卡住

解决：三种容错策略 + 重试降级 + Human-in-the-Loop

容错继续：A 失败 → B 和 C 继续 → Supervisor 汇总处理
快速失败：A 失败 → 立即终止 → 返回错误
条件传播：A 失败 → B 不执行 → 返回 A 的错误

// 策略 1：容错继续（Promise.allSettled）
async function faultTolerant(state) {
  const results = await Promise.allSettled([
    agentA(state),
    agentB(state),
    agentC(state),
  ]);
  // 区分成功和失败的结果
  const succeeded = results
    .filter((r) => r.status === "fulfilled")
    .map((r) => r.value);
  const failed = results
    .filter((r) => r.status === "rejected")
    .map((r) => r.reason.message);
  return { results: succeeded, errors: failed };
}

// 策略 2：快速失败（Promise.all）
async function failFast(state) {
  // 任何一个失败，立即抛出异常
  const results = await Promise.all([
    agentA(state),
    agentB(state),
    agentC(state),
  ]);
  return { results };
}

// 策略 3：重试 + 降级
import { RetryPolicy } from "@langchain/langgraph";

const retryPolicy = new RetryPolicy({
  maxAttempts: 3, // 最多重试 3 次
  backoffFactor: 2, // 退避倍数：1s → 2s → 4s
  retryOn: (error) => {
    // 只重试临时性错误（网络超时、API 限流）
    // 不重试永久性错误（参数错误、权限不足）
    return error.code === "TIMEOUT" || error.code === "RATE_LIMIT";
  },
});

// 策略 4：Human-in-the-Loop
async function humanApproval(state) {
  const decision = await interrupt({
    question: "Agent A 建议删除以下文件，是否批准？",
    files: state.filesToDelete,
  });
  if (decision === "approve") {
    return { approved: true };
  }
  return { approved: false, reason: "人类拒绝" };
}

// 指数退避（Exponential Backoff）
// 第 1 次重试：等 1 秒
// 第 2 次重试：等 2 秒
// 第 3 次重试：等 4 秒
// 加随机抖动（Jitter）防止多个 Agent 同时重试

function getRetryDelay(attempt) {
  const base = Math.pow(2, attempt) * 1000; // 指数退避
  const jitter = Math.random() * 1000; // 随机抖动
  return base + jitter;
}

// 重试的错误分类
const retryableErrors = [
  "TIMEOUT", // 网络超时
  "RATE_LIMIT", // API 限流
  "SERVER_ERROR", // 服务端 500 错误
];

const nonRetryableErrors = [
  "INVALID_PARAMS", // 参数错误（重试也不会成功）
  "AUTH_FAILED", // 认证失败
  "NOT_FOUND", // 资源不存在
];

// 幂等性设计：用唯一 ID 去重
const processedIds = new Set();

async function idempotentAgentCall(agentId, input) {
  const callId = `${agentId}-${hash(input)}`;

  // 检查是否已经处理过
  if (processedIds.has(callId)) {
    return getCachedResult(callId); // 返回缓存结果
  }

  const result = await agent.invoke(input);
  processedIds.add(callId);
  cacheResult(callId, result);
  return result;
}

问题：Agent 连续工作 5 小时+，上下文溢出 + 进程中断 + 状态丢失 + 成本爆炸

没有设计：Agent 运行 3 小时后上下文溢出，忘记之前的决策
           进程 OOM 崩溃，所有进度丢失
           token 费用失控

解决：异步执行 + Checkpoint 恢复 + 长期记忆 + 进度追踪

异步执行：Agent 在后台运行，用户可查看/暂停/恢复
Checkpoint：每个节点执行后自动保存快照，中断后从最近快照恢复
长期记忆：4 层记忆模型定期压缩，100K→5K tokens
进度追踪：子任务拆分 + 状态机 + 依赖解锁

// 异步执行 + 进度追踪
import { Client } from "@langchain/langgraph-sdk";

const client = new Client({ apiUrl: "http://localhost:2024" });

// 1. 创建异步 run（不阻塞）
const run = await client.runs.create(threadId, "agent-graph", {
  input: { task: "重写用户认证模块" },
});

// 2. 轮询进度
const status = await client.runs.get(run.id);
console.log(`状态: ${status.status}`); // pending / running / completed / failed
console.log(`进度: ${status.metadata.progress}%`);

// 3. 中途暂停（Human-in-the-Loop）
await client.runs.interrupt(run.id);

// 4. 恢复执行
await client.runs.resume(run.id, { input: { approved: true } });

// 子任务拆分 + 依赖管理 + 断点续传
const taskPlan = {
  quests: [
    { id: "Q1", name: "分析现有代码", status: "completed", dependsOn: [] },
    { id: "Q2", name: "设计新架构", status: "completed", dependsOn: ["Q1"] },
    { id: "Q3", name: "编写核心模块", status: "running", dependsOn: ["Q2"] },
    { id: "Q4", name: "编写测试", status: "pending", dependsOn: ["Q3"] },
    { id: "Q5", name: "集成测试", status: "pending", dependsOn: ["Q3", "Q4"] },
  ],
};

// Q3 完成后自动 cascade-unblock Q4
function onComplete(questId) {
  const quest = quests.find((q) => q.id === questId);
  quest.status = "completed";
  // 解锁依赖此任务的后续任务
  quests
    .filter((q) => q.dependsOn.includes(questId))
    .forEach((q) => {
      const allDepsCompleted = q.dependsOn.every(
        (depId) => quests.find((d) => d.id === depId).status === "completed",
      );
      if (allDepsCompleted) q.status = "unlocked";
    });
}

问题：玩具级 Agent 无法上生产环境

玩具级：Agent 有全部权限，没有审计日志，出错直接崩溃
        LLM 输出不可控，token 费用无上限

解决：权限收敛 + 审计日志 + 容错降级 + 确定性边界 + 成本控制

权限收敛：Agent 只有最小权限（MCP tool 级别控制）
审计日志：每步操作记录到 LangSmith，可完整溯源
容错降级：LLM 输出验证失败 → 降级到规则引擎
确定性边界：LLM 管理解意图，规则引擎管执行
成本控制：token 预算 + 模型分级 + Prompt Cache

// 企业级 Agent 的五层防护设计

// 1. 权限收敛：MCP tool 级别控制
const readOnlyAgent = createReactAgent({
  llm: openai,
  tools: [readFileTool, searchCodeTool], // 只有读权限，没有 writeFileTool
});

// 2. 审计日志：记录每步操作
const auditLogger = {
  log(agentId, action, input, output, tokenUsage) {
    const entry = {
      timestamp: new Date().toISOString(),
      agentId,
      action,
      input: truncate(input, 500),
      output: truncate(output, 500),
      tokens: tokenUsage,
    };
    // 写入 LangSmith 或自建审计系统
    langsmithClient.createRun(entry);
  },
};

// 3. 输出验证 + 降级
const validatedAgent = {
  async execute(task) {
    const result = await agent.invoke(task);
    const validation = schemaValidator.safeParse(result);
    if (!validation.success) {
      // 降级到规则引擎
      return ruleEngine.fallback(task);
    }
    return result;
  },
};

// 4. 确定性边界：LLM 理解 + 规则执行
async function boundedAgent(userInput) {
  // LLM 只负责理解意图（不确定性在这里）
  const intent = await llm.invoke(`提取用户意图：${userInput}`);
  // 规则引擎负责执行（确定性逻辑）
  return deterministicEngine.execute(intent);
}

// 5. 成本控制：token 预算管理
const budgetConfig = {
  perAgent: { maxTokens: 4000 },
  perWorkflow: { maxTokens: 50000 },
  totalDaily: { maxTokens: 1000000 },
  onExceed: "degrade", // 超预算时降级到弱模型
};

企业级 Agent 的分层架构：

┌───────────────────────────────────────┐
│    理解层（LLM，不确定性）              │  ← 自然语言理解、意图提取、上下文推理
├───────────────────────────────────────┤
│    验证层（Schema，半确定性）            │  ← JSON Schema 校验、输出格式约束
├───────────────────────────────────────┤
│    执行层（规则引擎，确定性）            │  ← 工具调用、数据操作、业务逻辑
├───────────────────────────────────────┤
│    监控层（审计 + 告警）                │  ← 操作日志、成本追踪、异常告警
└───────────────────────────────────────┘

关键原则：不确定性只存在于理解层，执行层必须确定性

┌───────────────────────────────────────┐
│    Supervisor 模式                     │  ← 一个主管路由 + N 个 Agent 执行
└─────────────────┬─────────────────────┘
                  │ 适用：动态任务分配、智能路由
                  ▼
┌───────────────────────────────────────┐
│    Hierarchical 模式                   │  ← 多层 Supervisor 逐级分配
└─────────────────┬─────────────────────┘
                  │ 适用：大型复杂任务、有组织结构
                  ▼
┌───────────────────────────────────────┐
│    Network 模式                        │  ← Agent 之间直接通信，无主管
└─────────────────┬─────────────────────┘
                  │ 适用：需要多轮协商的场景
                  ▼
┌───────────────────────────────────────┐
│    自定义拓扑                           │  ← 任意图结构，按需组合
└───────────────────────────────────────┘
                  │ 适用：复杂业务流程

// Supervisor 模式：动态路由 + Agent 执行
import { StateGraph, END } from "@langchain/langgraph";
import { ChatOpenAI } from "@langchain/openai";

const llm = new ChatOpenAI({ model: "gpt-4o" });

// Supervisor 节点：用 LLM 决定下一步
async function supervisorNode(state) {
  const response = await llm.invoke([
    SystemMessage(
      (content = `你是主管，决定下一步调用哪个 Agent。
可用 Agent：coder（写代码）、reviewer（审查代码）、tester（写测试）。
输出 JSON：{"next": "coder"} 或 {"next": "reviewer"} 或 {"next": "FINISH"}`),
    ),
    ...state.messages,
  ]);
  const decision = JSON.parse(response.content);
  return { next: decision.next };
}

// Worker 节点：执行具体任务
async function coderNode(state) {
  const result = await coderAgent.invoke(state);
  return { messages: [result] };
}

// 组装图
const graph = new StateGraph(State)
  .addNode("supervisor", supervisorNode)
  .addNode("coder", coderNode)
  .addNode("reviewer", reviewerNode)
  .addNode("tester", testerNode)
  .addConditionalEdges("supervisor", (state) => state.next, {
    coder: "coder",
    reviewer: "reviewer",
    tester: "tester",
    FINISH: END,
  })
  .addEdge("coder", "supervisor")
  .addEdge("reviewer", "supervisor")
  .addEdge("tester", "supervisor")
  .compile();

// Hierarchical 模式：多层 Supervisor
const hierarchicalGraph = new StateGraph(State)
  // 顶层：宏观决策
  .addNode("topSupervisor", topSupervisorNode)
  // 中层：领域分配
  .addNode("devSupervisor", devSupervisorNode)
  .addNode("qaSupervisor", qaSupervisorNode)
  // 底层：执行
  .addNode("coder", coderAgent)
  .addNode("reviewer", reviewerAgent)
  .addNode("tester", testerAgent)
  // 顶层 → 中层
  .addConditionalEdges("topSupervisor", (state) => state.domain, {
    dev: "devSupervisor",
    qa: "qaSupervisor",
  })
  // 中层 → 底层
  .addConditionalEdges("devSupervisor", (state) => state.task, {
    code: "coder",
    review: "reviewer",
  })
  // 底层 → 回到中层
  .addEdge("coder", "devSupervisor")
  .addEdge("reviewer", "devSupervisor")
  .compile();

// Network 模式：Agent 之间直接通信
const networkGraph = new StateGraph(State)
  .addNode("architect", architectAgent)
  .addNode("securityExpert", securityAgent)
  .addNode("perfExpert", perfAgent)
  // 每个 Agent 可以决定调用其他 Agent 或结束
  .addConditionalEdges("architect", (state) => state.next, {
    securityExpert: "securityExpert",
    perfExpert: "perfExpert",
    FINISH: END,
  })
  .addConditionalEdges("securityExpert", (state) => state.next, {
    architect: "architect",
    perfExpert: "perfExpert",
    FINISH: END,
  })
  .addConditionalEdges("perfExpert", (state) => state.next, {
    architect: "architect",
    securityExpert: "securityExpert",
    FINISH: END,
  })
  .compile();

// 自定义拓扑：Supervisor + Collaborative + Pipeline
const customGraph = new StateGraph(State)
  // 顶层 Supervisor 路由
  .addNode("router", routerNode)
  // 并行分析（Collaborative）
  .addNode("securityCheck", securityAgent)
  .addNode("perfCheck", perfAgent)
  .addNode("readabilityCheck", readabilityAgent)
  // 汇总 + Pipeline 后处理
  .addNode("consensus", consensusNode)
  .addNode("formatter", formatterNode)
  // Supervisor → 并行分析
  .addConditionalEdges("router", () => [
    new Send("securityCheck", state),
    new Send("perfCheck", state),
    new Send("readabilityCheck", state),
  ])
  // 并行分析 → 汇总
  .addEdge("securityCheck", "consensus")
  .addEdge("perfCheck", "consensus")
  .addEdge("readabilityCheck", "consensus")
  // 汇总 → Pipeline
  .addEdge("consensus", "formatter")
  .compile();

问自己三个问题：
1. 这类请求的分支能不能列全？        能 → 固定路由
2. 每个分支的处理步骤是否长期稳定？    是 → 固定路由
3. 任务规模是否随输入剧烈变化？        是 → 动态拆解

type Route =
  | { kind: "fixed"; handler: "export-pdf" | "export-csv" }
  | { kind: "dynamic"; goal: string };

// 能枚举的走固定分支，枚举不了的才进动态编排
function route(intent: Intent): Route {
  if (intent.type === "export") {
    return {
      kind: "fixed",
      handler: intent.format === "pdf" ? "export-pdf" : "export-csv",
    };
  }
  return { kind: "dynamic", goal: intent.raw };
}

串行链路：整体成功率 ≈ (1 - p)^N      （p 为单步成功率）
并行扇出：任一失败概率 ≈ 1 - (1 - p)^N

# Supervisor 模式的实现（简化）
from langgraph.prebuilt import create_react_agent

# Router 节点：根据 LLM 输出决定下一步
def supervisor_router(state):
    response = llm.invoke([
        SystemMessage(content="你是主管，决定下一步调用哪个 Agent"),
        *state["messages"],
    ])
    # LLM 输出：{"next": "coder"} 或 {"next": "reviewer"} 或 {"next": "FINISH"}
    return json.loads(response.content)["next"]

# 组装图
graph = StateGraph(State)
graph.add_node("supervisor", supervisor_node)
graph.add_node("coder", coder_agent)
graph.add_node("reviewer", reviewer_agent)
graph.add_conditional_edges("supervisor", supervisor_router, {
    "coder": "coder",
    "reviewer": "reviewer",
    FINISH: END,
})
graph.add_edge("coder", "supervisor")
graph.add_edge("reviewer", "supervisor")