《互联网产品的 AI 架构·多能力编排与任务闭环》3.1 把编排写成图

// file: src/orch/graph.ts
// tier: L2
// verify: node --test test/orch/graph.test.ts

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

export type JoinPolicy = "all" | "any" | "quorum";
export type NodeKind = "capability" | "human" | "action";
export type NodeStatus = "succeeded" | "failed" | "skipped";

/** 节点是可调度的最小单位：调用什么、花多少、最多试几次，不在这里写提示词。 */
export interface GraphNode {
  id: string;
  capability: string;
  kind: NodeKind;
  /** 示例成本，整数微元；包含该节点自己的上下文与输出。 */
  costMicros: number;
  /** 示例耗时，毫秒。 */
  latencyMs: number;
  /** 最多尝试次数，默认 1；重试会重复计费。 */
  maxAttempts?: number;
  /** 确定性失败注入：第几次尝试失败（从 1 起）。 */
  failAttempts?: readonly number[];
  /** 多前驱汇合策略，默认 all。 */
  join?: JoinPolicy;
  /** join 为 quorum 时需要的成功票数。 */
  quorum?: number;
}

/** 边表示数据依赖：to 需要 from 的结果，不表示“上一步做完做下一步”。 */
export interface GraphEdge {
  from: string;
  to: string;
}

/** 终止条件除了唯一的终点，还需要步数与成本两道硬闸。 */
export interface GraphLimits {
  maxNodeRuns: number;
  maxCostMicros: number;
}

export interface ExecutionGraph {
  id: string;
  entry: string;
  terminal: string;
  nodes: readonly GraphNode[];
  edges: readonly GraphEdge[];
  limits: GraphLimits;
  /** 每次跨层交接的固定延迟，示例值 35ms。 */
  handoffLatencyMs: number;
}

export interface GraphIssue {
  code: string;
  nodeId?: string;
  message: string;
}

/** 检查点里的单节点记录；恢复时只信任 succeeded。 */
export interface NodeRecord {
  nodeId: string;
  status: NodeStatus;
  attempts: number;
  output: string;
  costMicros: number;
  latencyMs: number;
  idempotencyKey: string;
  level: number;
}

export interface Checkpoint {
  graphId: string;
  rootIdempotencyKey: string;
  records: NodeRecord[];
  /** 跨多次恢复累计的成本，整数微元。 */
  spentCostMicros: number;
  /** 跨多次恢复累计的节点尝试次数。 */
  nodeRuns: number;
  levelsDone: number;
}

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

function issue(code: string, message: string, nodeId?: string): GraphIssue {
  return nodeId === undefined ? { code, message } : { code, message, nodeId };
}

/**
 * 子节点的幂等键由根键与节点 id 派生，重试和恢复都不会改变它。
 * 下游拿到的是同一个根键，而不是每个节点各自生成一个 UUID。
 */
export function deriveIdempotencyKey(rootKey: string, nodeId: string): string {
  let hash = 0x811c9dc5;
  for (const char of `${rootKey}::${nodeId}`) {
    hash ^= char.codePointAt(0) ?? 0;
    hash = Math.imul(hash, 0x01000193) >>> 0;
  }
  return `${rootKey}.${hash.toString(16).padStart(8, "0")}`;
}

export function nodeMap(graph: ExecutionGraph): Map<string, GraphNode> {
  return new Map(graph.nodes.map((node) => [node.id, node]));
}

export function incomingEdges(graph: ExecutionGraph, nodeId: string): GraphEdge[] {
  return graph.edges.filter((edge) => edge.to === nodeId);
}

function reachable(start: string, edges: readonly GraphEdge[], direction: "forward" | "backward"): Set<string> {
  const next = new Map<string, string[]>();
  for (const edge of edges) {
    const from = direction === "forward" ? edge.from : edge.to;
    const to = direction === "forward" ? edge.to : edge.from;
    next.set(from, [...(next.get(from) ?? []), to]);
  }
  const seen = new Set<string>([start]);
  const queue = [start];
  while (queue.length > 0) {
    for (const neighbor of next.get(queue.shift() as string) ?? []) {
      if (!seen.has(neighbor)) {
        seen.add(neighbor);
        queue.push(neighbor);
      }
    }
  }
  return seen;
}

/** Kahn 排序；返回的 cycle 非空说明图有环，order 不完整。 */
function kahn(ids: readonly string[], edges: readonly GraphEdge[]): { order: string[]; cycle: string[] } {
  const indegree = new Map(ids.map((id) => [id, 0]));
  const outgoing = new Map<string, string[]>(ids.map((id) => [id, []]));
  for (const edge of edges) {
    indegree.set(edge.to, (indegree.get(edge.to) ?? 0) + 1);
    outgoing.get(edge.from)?.push(edge.to);
  }
  const ready = ids.filter((id) => (indegree.get(id) ?? 0) === 0).sort();
  const order: string[] = [];
  while (ready.length > 0) {
    const id = ready.shift() as string;
    order.push(id);
    for (const neighbor of [...(outgoing.get(id) ?? [])].sort()) {
      const left = (indegree.get(neighbor) ?? 0) - 1;
      indegree.set(neighbor, left);
      if (left === 0) {
        ready.push(neighbor);
        ready.sort();
      }
    }
  }
  return { order, cycle: ids.filter((id) => (indegree.get(id) ?? 0) > 0).sort() };
}

/** 先收结构问题，再谈调度；有环或不可达的图不应该进入执行器。 */
export function validateGraph(graph: ExecutionGraph): GraphIssue[] {
  if (graph.nodes.length === 0) return [issue("empty-graph", "图至少需要一个节点")];
  const issues: GraphIssue[] = [];
  const byId = new Map<string, GraphNode>();
  for (const node of graph.nodes) {
    if (node.id.trim() === "") {
      issues.push(issue("empty-node-id", "节点 id 不能为空"));
      continue;
    }
    if (byId.has(node.id)) issues.push(issue("duplicate-node", `重复的节点 ${node.id}`, node.id));
    else byId.set(node.id, node);
    if (!Number.isInteger(node.costMicros) || node.costMicros < 0) {
      issues.push(issue("bad-cost", "costMicros 必须是非负整数", node.id));
    }
    if (!Number.isFinite(node.latencyMs) || node.latencyMs < 0) {
      issues.push(issue("bad-latency", "latencyMs 必须是非负数", node.id));
    }
    if (node.maxAttempts !== undefined && (!Number.isInteger(node.maxAttempts) || node.maxAttempts < 1)) {
      issues.push(issue("bad-attempts", "maxAttempts 必须是 >= 1 的整数", node.id));
    }
    if (node.join !== undefined && !["all", "any", "quorum"].includes(node.join)) {
      issues.push(issue("bad-join", `未知的汇合策略 ${String(node.join)}`, node.id));
    }
  }

  if (!byId.has(graph.entry)) issues.push(issue("bad-entry", `入口 ${graph.entry} 不在节点表中`));
  if (!byId.has(graph.terminal)) issues.push(issue("bad-terminal", `终点 ${graph.terminal} 不在节点表中`));
  if (!Number.isInteger(graph.limits.maxNodeRuns) || graph.limits.maxNodeRuns < 1) {
    issues.push(issue("bad-run-limit", "maxNodeRuns 必须是 >= 1 的整数"));
  }
  if (!Number.isInteger(graph.limits.maxCostMicros) || graph.limits.maxCostMicros < 0) {
    issues.push(issue("bad-cost-limit", "maxCostMicros 必须是非负整数"));
  }
  if (!Number.isFinite(graph.handoffLatencyMs) || graph.handoffLatencyMs < 0) {
    issues.push(issue("bad-handoff", "handoffLatencyMs 必须是非负数"));
  }

  const edges: GraphEdge[] = [];
  const seen = new Set<string>();
  for (const edge of graph.edges) {
    if (!byId.has(edge.from) || !byId.has(edge.to)) {
      issues.push(issue("unknown-node", `边 ${edge.from}->${edge.to} 引用了不存在的节点`));
    } else if (edge.from === edge.to) {
      issues.push(issue("self-loop", `节点 ${edge.from} 指向自己`, edge.from));
    } else if (seen.has(`${edge.from}->${edge.to}`)) {
      issues.push(issue("duplicate-edge", `重复的边 ${edge.from}->${edge.to}`));
    } else {
      seen.add(`${edge.from}->${edge.to}`);
      edges.push(edge);
    }
  }

  const ids = [...byId.keys()];
  const { cycle } = kahn(ids, edges);
  if (cycle.length > 0) issues.push(issue("cycle", `图存在环，涉及 ${cycle.join(", ")}`));

  if (byId.has(graph.entry) && byId.has(graph.terminal)) {
    const forward = reachable(graph.entry, edges, "forward");
    const backward = reachable(graph.terminal, edges, "backward");
    for (const id of [...ids].sort()) {
      if (!forward.has(id)) issues.push(issue("unreachable", `节点 ${id} 从入口不可达`, id));
      if (!backward.has(id)) issues.push(issue("no-path-to-terminal", `节点 ${id} 到不了终点 ${graph.terminal}`, id));
    }
  }

  const touched = new Set(edges.flatMap((edge) => [edge.from, edge.to]));
  for (const id of [...ids].sort()) {
    if (!touched.has(id) && id !== graph.entry && id !== graph.terminal) {
      issues.push(issue("isolated-node", `节点 ${id} 没有任何边`, id));
    }
  }

  for (const node of byId.values()) {
    if (node.join !== "quorum") continue;
    const deps = edges.filter((edge) => edge.to === node.id).length;
    const quorum = node.quorum ?? 0;
    if (!Number.isInteger(quorum) || quorum < 1 || quorum > deps) {
      issues.push(issue("bad-quorum", `quorum 必须落在 1..${deps}`, node.id));
    }
  }
  return issues;
}

export function assertValidGraph(graph: ExecutionGraph): void {
  const issues = validateGraph(graph);
  if (issues.length > 0) {
    fail(issues.map((entry) => `${entry.code}: ${entry.message}`).join("; "));
  }
}

/** 同一张图每次得到同一个拓扑序，测试与回放才能比较。 */
export function topologicalOrder(graph: ExecutionGraph): string[] {
  assertValidGraph(graph);
  return kahn(graph.nodes.map((node) => node.id), graph.edges).order;
}

/** 层号取“到入口的最长路径”，同一层的节点之间没有依赖，可以并行。 */
export function levelize(graph: ExecutionGraph): string[][] {
  const order = topologicalOrder(graph);
  const levelOf = new Map<string, number>();
  for (const id of order) {
    const parents = incomingEdges(graph, id).map((edge) => levelOf.get(edge.from) ?? 0);
    levelOf.set(id, parents.length === 0 ? 0 : Math.max(...parents) + 1);
  }
  const levels: string[][] = [];
  for (const id of order) {
    const level = levelOf.get(id) as number;
    (levels[level] ??= []).push(id);
  }
  return levels;
}