《互联网产品的 AI 架构·多能力编排与任务闭环》1.1 一个请求要几个能力：回声的三条真实链路

期望成本 = Σ 到达率 × 执行率 × 单次成本 × 期望尝试次数
成功率   = Π (执行率 × 单步成功率 + (1 − 执行率))
每次成功成本 = 期望成本 ÷ 成功率

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

import { AiError } from "../core/errors.ts";

export type NodeKind = "ai" | "human" | "external";
export type StageMode = "serial" | "parallel";
export type ChainId = "A" | "B" | "C";

/** 一次可计量动作：模型调用、人工处理或外部副作用。 */
export interface FlowNode {
  id: string; capability: string; kind: NodeKind;
  inputTokens: number; outputTokens: number; latencyMs: number;
  failureRate: number; retries: number;
}

/** 阶段内串行或并行；executionRate 让条件分支直接进入模型。 */
export interface FlowStage {
  id: string; mode: StageMode; executionRate: number; nodes: readonly FlowNode[];
}

export interface ChainSpec {
  id: ChainId; name: string;
  sharedContextTokens: number; handoffTokens: number; handoffLatencyMs: number;
  stages: readonly FlowStage[];
}

export interface PriceSheet {
  inputMicrosPerKTok: number; outputMicrosPerKTok: number;
  prefixReuseRatio: number; cachedInputPriceRatio: number;
}

export interface NodeBreakdown {
  id: string; capability: string; kind: NodeKind; contextTokens: number;
  attemptCostMicros: number; expectedCostMicros: number;
  effectiveFailureRate: number; latencyMs: number;
}

export interface StageBreakdown {
  id: string; mode: StageMode; executionRate: number;
  costMicros: number; retryCostMicros: number;
  successGivenRun: number; expectedSuccess: number; latencyMs: number;
  nodes: NodeBreakdown[];
}

export interface ChainRun {
  id: ChainId; name: string; nodeCount: number; stageCount: number;
  successProbability: number; expectedCostMicros: number; costPerSuccessMicros: number;
  /** 所有阶段都执行且不重试时的延迟，即最坏路径。 */
  fullPathLatencyMs: number;
  /** 按到达概率与条件执行率加权的延迟，即典型请求。 */
  expectedLatencyMs: number;
  retryCostMicros: number; wastedCostMicros: number; handoffCount: number;
  stages: StageBreakdown[];
}

export const ECHO_PRICE: PriceSheet = {
  inputMicrosPerKTok: 250, outputMicrosPerKTok: 1_000,
  prefixReuseRatio: 0.6, cachedInputPriceRatio: 0.1,
};

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

function integer(value: number, field: string, min = 0): number {
  if (!Number.isInteger(value) || value < min) fail(`${field} 必须是 >= ${min} 的整数`);
  return value;
}

function finite(value: number, field: string, min = 0): number {
  if (!Number.isFinite(value) || value < min) fail(`${field} 必须是 >= ${min} 的有限数`);
  return value;
}

function ratio(value: number, field: string): number {
  if (!Number.isFinite(value) || value <= 0 || value > 1) fail(`${field} 必须落在 (0, 1]`);
  return value;
}

function tokenCost(tokens: number, microsPerKTok: number): number {
  return Math.ceil((tokens * microsPerKTok) / 1_000);
}

/** 首跳全价；后续跳按前缀复用率与缓存折扣计价。 */
export function contextCost(tokenCount: number, price: PriceSheet, reused: boolean): number {
  const factor = reused ? 1 - price.prefixReuseRatio * (1 - price.cachedInputPriceRatio) : 1;
  return Math.ceil((tokenCount * price.inputMicrosPerKTok * factor) / 1_000);
}

/** 单次尝试 = 上下文 + 输入 + 输出；人工与外部动作不计 token。 */
export function attemptCost(
  node: FlowNode, price: PriceSheet, contextTokens: number, reused: boolean,
): number {
  const context = node.kind === "ai" ? contextCost(contextTokens, price, reused) : 0;
  return context +
    tokenCost(node.inputTokens, price.inputMicrosPerKTok) +
    tokenCost(node.outputTokens, price.outputMicrosPerKTok);
}

/** 重试 r 次后，整步失败当且仅当每次都失败。 */
export function effectiveFailureRate(node: FlowNode): number {
  return Math.pow(node.failureRate, node.retries + 1);
}

/** 固定单次成本下的期望尝试次数：1 + f + … + f^r，r=1 时等于 1+f。 */
export function expectedAttempts(node: FlowNode): number {
  if (node.failureRate === 0) return 1;
  return (1 - Math.pow(node.failureRate, node.retries + 1)) / (1 - node.failureRate);
}

export function validateChain(chain: ChainSpec): void {
  if (chain.stages.length === 0) fail("链路至少需要一个阶段");
  integer(chain.sharedContextTokens, "sharedContextTokens");
  integer(chain.handoffTokens, "handoffTokens");
  finite(chain.handoffLatencyMs, "handoffLatencyMs");
  const stageIds = new Set<string>();
  const nodeIds = new Set<string>();
  for (const stage of chain.stages) {
    if (stage.id.trim() === "" || stageIds.has(stage.id)) fail(`stage.id 非法或重复: ${stage.id}`);
    stageIds.add(stage.id);
    ratio(stage.executionRate, `${stage.id}.executionRate`);
    if (stage.nodes.length === 0) fail(`${stage.id} 没有节点`);
    for (const node of stage.nodes) {
      if (node.id.trim() === "" || nodeIds.has(node.id)) fail(`node.id 非法或重复: ${node.id}`);
      nodeIds.add(node.id);
      integer(node.inputTokens, `${node.id}.inputTokens`);
      integer(node.outputTokens, `${node.id}.outputTokens`);
      finite(node.latencyMs, `${node.id}.latencyMs`);
      if (!(node.failureRate >= 0 && node.failureRate < 1)) fail(`${node.id}.failureRate 必须落在 [0, 1)`);
      integer(node.retries, `${node.id}.retries`);
    }
  }
}

/** 把一条链路展开成节点级成本、阶段级成功率与整链期望值。 */
export function evaluateChain(chain: ChainSpec, price: PriceSheet = ECHO_PRICE): ChainRun {
  validateChain(chain);
  integer(price.inputMicrosPerKTok, "inputMicrosPerKTok", 1);
  integer(price.outputMicrosPerKTok, "outputMicrosPerKTok", 1);
  ratio(price.prefixReuseRatio, "prefixReuseRatio");
  ratio(price.cachedInputPriceRatio, "cachedInputPriceRatio");

  let reach = 1, success = 1, expectedCost = 0, retryCost = 0;
  let fullPathLatency = 0, expectedLatency = 0, nodeCount = 0, handoffCount = 0;
  const stages: StageBreakdown[] = [];

  chain.stages.forEach((stage, index) => {
    const contextTokens = index === 0 ? chain.sharedContextTokens : chain.handoffTokens;
    const handoffMs = index === 0 ? 0 : chain.handoffLatencyMs;
    if (index > 0) handoffCount += 1;
    let stageCost = 0, stageRetryCost = 0, stageSuccess = 1;
    let serialLatency = 0, parallelLatency = 0;
    const nodes: NodeBreakdown[] = [];

    for (const node of stage.nodes) {
      nodeCount += 1;
      const single = attemptCost(node, price, contextTokens, index > 0);
      const attempts = expectedAttempts(node);
      stageCost += single * attempts;
      stageRetryCost += single * (attempts - 1);
      stageSuccess *= 1 - effectiveFailureRate(node);
      serialLatency += node.latencyMs;
      parallelLatency = Math.max(parallelLatency, node.latencyMs);
      nodes.push({
        id: node.id, capability: node.capability, kind: node.kind, contextTokens,
        attemptCostMicros: single,
        expectedCostMicros: Math.round(stage.executionRate * single * attempts),
        effectiveFailureRate: Number(effectiveFailureRate(node).toFixed(6)),
        latencyMs: node.latencyMs,
      });
    }

    const stageLatency = stage.mode === "parallel" ? parallelLatency : serialLatency;
    const expectedSuccess = stage.executionRate * stageSuccess + (1 - stage.executionRate);
    expectedCost += reach * stage.executionRate * stageCost;
    retryCost += reach * stage.executionRate * stageRetryCost;
    expectedLatency += reach * stage.executionRate * (stageLatency + handoffMs);
    fullPathLatency += stageLatency + handoffMs;
    stages.push({
      id: stage.id, mode: stage.mode, executionRate: stage.executionRate,
      costMicros: Math.round(stageCost), retryCostMicros: Math.round(stageRetryCost),
      successGivenRun: Number(stageSuccess.toFixed(6)),
      expectedSuccess: Number(expectedSuccess.toFixed(6)),
      latencyMs: stageLatency, nodes,
    });
    success *= expectedSuccess;
    reach *= expectedSuccess;
  });

  return {
    id: chain.id, name: chain.name, nodeCount, stageCount: chain.stages.length,
    successProbability: Number(success.toFixed(6)),
    expectedCostMicros: Math.round(expectedCost),
    costPerSuccessMicros: success === 0 ? Number.POSITIVE_INFINITY : Math.round(expectedCost / success),
    fullPathLatencyMs: Math.round(fullPathLatency),
    expectedLatencyMs: Math.round(expectedLatency),
    retryCostMicros: Math.round(retryCost),
    wastedCostMicros: Math.round(expectedCost * (1 - success)),
    handoffCount, stages,
  };
}

const node = (
  id: string, capability: string, kind: NodeKind, inputTokens: number,
  outputTokens: number, latencyMs: number, failureRate: number, retries: number,
): FlowNode => ({ id, capability, kind, inputTokens, outputTokens, latencyMs, failureRate, retries });

const stage = (
  id: string, nodes: FlowNode[], executionRate = 1, mode: StageMode = "serial",
): FlowStage => ({ id, mode, executionRate, nodes });

const chain = (id: ChainId, name: string, stages: FlowStage[]): ChainSpec =>
  ({ id, name, sharedContextTokens: 3_200, handoffTokens: 820, handoffLatencyMs: 35, stages });

/** 链路 A：人工复核只在 18% 的灰度区间触发。 */
export const CHAIN_A = chain("A", "内容发布安全链", [
  stage("summarize", [node("summarize", "content.summarize", "ai", 2_600, 420, 620, 0.03, 1)]),
  stage("moderate", [node("moderate", "moderation.classify", "ai", 900, 60, 180, 0.02, 1)]),
  stage("human-review", [node("human-review", "human.review", "human", 0, 0, 90_000, 0.005, 0)], 0.18),
  stage("publish", [node("publish", "action.publish", "external", 0, 0, 40, 0.002, 1)]),
]);

/** 链路 B：判意图、检索、答复三步串行。 */
export const CHAIN_B = chain("B", "私信客服链", [
  stage("intent", [node("intent", "chat.reply", "ai", 700, 40, 180, 0.02, 1)]),
  stage("retrieve", [node("retrieve", "support.assist", "ai", 1_200, 260, 300, 0.025, 1)]),
  stage("answer", [node("answer", "chat.reply", "ai", 1_100, 240, 420, 0.02, 1)]),
]);

/** 链路 C：顺序固定，最后一步是不可回滚的外部动作。 */
export const CHAIN_C = chain("C", "站内下单咨询链", [
  stage("rerank", [node("rerank", "search.rerank", "ai", 700, 80, 160, 0.01, 1)]),
  stage("explain", [node("explain", "chat.reply", "ai", 1_100, 240, 420, 0.02, 1)]),
  stage("create-order", [node("create-order", "action.createOrder", "external", 0, 0, 250, 0.005, 0)]),
]);

export const ECHO_CHAINS: readonly ChainSpec[] = [CHAIN_A, CHAIN_B, CHAIN_C];

0.2 × 2,313 + 0.5 × 2,328 + 0.3 × 1,688 = 2,133 微元