《互联网产品的 AI 架构·多能力编排与任务闭环》1.2 串行、并行与条件分支

// file: src/orch/shape-variants.ts
// tier: L1
// verify: node --test test/orch/workflow-shape.test.ts

import { AiError } from "../core/errors.ts";
import {
  CHAIN_A,
  ECHO_PRICE,
  evaluateChain,
  type ChainRun,
  type ChainSpec,
  type FlowNode,
} from "./workflow-shape.ts";

const ASSIST: FlowNode = {
  id: "retrieve",
  capability: "support.assist",
  kind: "ai",
  inputTokens: 1_200,
  outputTokens: 260,
  latencyMs: 300,
  failureRate: 0.025,
  retries: 1,
};

const RERANK: FlowNode = {
  id: "rerank",
  capability: "search.rerank",
  kind: "ai",
  inputTokens: 700,
  outputTokens: 80,
  latencyMs: 160,
  failureRate: 0.01,
  retries: 1,
};

const INTENT: FlowNode = {
  id: "intent",
  capability: "chat.reply",
  kind: "ai",
  inputTokens: 700,
  outputTokens: 40,
  latencyMs: 180,
  failureRate: 0.02,
  retries: 1,
};

const ANSWER: FlowNode = {
  id: "answer",
  capability: "chat.reply",
  kind: "ai",
  inputTokens: 1_100,
  outputTokens: 240,
  latencyMs: 420,
  failureRate: 0.02,
  retries: 1,
};

/** 两段检索串行：知识库检索完成后再查社区内容。 */
export const B_SERIAL: ChainSpec = {
  id: "B",
  name: "私信客服链（两段检索串行）",
  sharedContextTokens: 3_200,
  handoffTokens: 820,
  handoffLatencyMs: 35,
  stages: [
    { id: "intent", mode: "serial", executionRate: 1, nodes: [INTENT] },
    { id: "assist", mode: "serial", executionRate: 1, nodes: [{ ...ASSIST, id: "assist" }] },
    { id: "rerank", mode: "serial", executionRate: 1, nodes: [RERANK] },
    { id: "answer", mode: "serial", executionRate: 1, nodes: [ANSWER] },
  ],
};

/** 两段检索并行：共享一次交接，汇合后一起进入答复。 */
export const B_PARALLEL: ChainSpec = {
  id: "B",
  name: "私信客服链（两段检索并行）",
  sharedContextTokens: 3_200,
  handoffTokens: 820,
  handoffLatencyMs: 35,
  stages: [
    { id: "intent", mode: "serial", executionRate: 1, nodes: [INTENT] },
    {
      id: "retrieve",
      mode: "parallel",
      executionRate: 1,
      nodes: [{ ...ASSIST, id: "assist" }, { ...RERANK, id: "rerank" }],
    },
    { id: "answer", mode: "serial", executionRate: 1, nodes: [ANSWER] },
  ],
};

/** 人工复核每次都跑：把条件分支关掉，作为对照。 */
export const A_ALWAYS_REVIEW: ChainSpec = {
  ...CHAIN_A,
  name: "内容发布安全链（每次人工复核）",
  stages: CHAIN_A.stages.map((stage) =>
    stage.id === "human-review" ? { ...stage, executionRate: 1 } : stage
  ),
};

export interface ShapeDelta {
  costMicros: number;
  costPct: number;
  latencyMs: number;
  latencyPct: number;
  /** 成功率差值，单位是百分点。 */
  successPoints: number;
}

function pct(delta: number, base: number): number {
  if (base === 0) fail("基准为 0，无法计算百分比");
  return Number(((delta / base) * 100).toFixed(2));
}

function fail(message: string): never {
  throw new AiError("invalid-request", `shape-variants: ${message}`);
}

/** 正数代表候选方案更贵、更慢或成功率更高。 */
export function compareRuns(base: ChainRun, candidate: ChainRun): ShapeDelta {
  return {
    costMicros: candidate.expectedCostMicros - base.expectedCostMicros,
    costPct: pct(candidate.expectedCostMicros - base.expectedCostMicros, base.expectedCostMicros),
    latencyMs: candidate.expectedLatencyMs - base.expectedLatencyMs,
    latencyPct: pct(candidate.expectedLatencyMs - base.expectedLatencyMs, base.expectedLatencyMs),
    successPoints: Number(((candidate.successProbability - base.successProbability) * 100).toFixed(2)),
  };
}

export interface PatternComparison {
  serial: ChainRun;
  parallel: ChainRun;
  serialToParallel: ShapeDelta;
  /** 并行相对串行省下的交接次数。 */
  handoffsSaved: number;
}

/** 同一组节点、只改阶段结构：成本应当相同，差异只来自延迟与交接。 */
export function compareSerialParallel(): PatternComparison {
  const serial = evaluateChain(B_SERIAL, ECHO_PRICE);
  const parallel = evaluateChain(B_PARALLEL, ECHO_PRICE);
  return {
    serial,
    parallel,
    serialToParallel: compareRuns(serial, parallel),
    handoffsSaved: serial.handoffCount - parallel.handoffCount,
  };
}

export interface ConditionalComparison {
  always: ChainRun;
  conditional: ChainRun;
  alwaysToConditional: ShapeDelta;
}

/** 同一组节点、只改人工阶段的执行率，观察延迟与成功率的交换。 */
export function compareConditionalReview(): ConditionalComparison {
  const always = evaluateChain(A_ALWAYS_REVIEW, ECHO_PRICE);
  const conditional = evaluateChain(CHAIN_A, ECHO_PRICE);
  return {
    always,
    conditional,
    alwaysToConditional: compareRuns(always, conditional),
  };
}