一次「能省 420ms」的并行拆分
回声的内容发布链上，moderation.classify 要 420ms：政策分类 260ms，敏感词扫描 160ms。有人提出把两步并行，直觉上能省 160ms。
实际跑下来只省 125ms。两个分支结束后要汇合、校验并交给人工复核，汇合点新增 35ms；并行后的耗时是 max(260, 160) + 35 = 295ms，相对 420ms 省 125ms。如果当时只拿到一个并发槽，两步只能顺序执行，耗时变成 260 + 160 + 35 = 455ms，比不拆还慢 35ms。
并行不会减少总工作量，它只改变等待结构。一个拆分是否值得，要同时回答四个问题：临界路径缩短了多少、并发槽够不够、共享上下文多花多少钱、再多加一个分支还有没有边际收益。本节所有数字都来自 ai-lab 的示例样本运行结果，不是生产实测。
临界路径：只能压缩可重叠的部分
把一段串行工作拆成两个分支后，完成时间由最长分支决定，而不是两者之和：
串行段 S = Σ dᵢ
并行段 P = max(dᵢ) + join
节省 Δ = S − P
理论上限 = S ÷ max(dᵢ)
420ms 的段拆成 260ms 与 160ms，理论上限是 1.62 倍；加上 35ms 汇合开销后，实际加速比是 1.42 倍。最长分支决定了并行的天花板：如果 160ms 的分支被拆成 150ms 与 10ms，最长分支仍是 260ms，节省不变。反过来，先压缩最长分支，收益才会重新出现。
汇合开销不是常数。它包含结果校验、字段合并、冲突处理和一次交接，示例样本按 35ms 计。分支越多、结果结构越不一致，汇合越贵；当汇合开销超过并行省下的时间，拆分就是负收益。
共享配额：并发槽是借来的
提供方给的是总并发，不是每个请求的并发。示例里提供方允许 20 个并发调用，同时有 10 个活跃请求，单个请求能拿到的槽是 floor(20 ÷ 10) = 2。如果拆成三个分支，第三个分支只能等前两个之一结束。
| 单请求槽数 | 完成时间 | 相对串行 |
|---|---|---|
| 1 | 455ms | +35ms |
| 2 | 295ms | −125ms |
| 3 | 295ms | −125ms |
这张表说明两件事。第一，槽数少于分支数时，排队延迟会吃掉收益，甚至让并行变慢。第二，槽数超过分支数不会更快，多出来的槽只能闲着。
更麻烦的是配额会随流量变化。低峰时一个请求能拿 2 个槽，高峰时 20 个槽要分给 20 个请求，每个请求只剩 1 个槽。并行方案在低峰看起来省 125ms，在高峰反而多 35ms。把 fairConcurrency 的结果写进容量评估，比在压测环境里独享配额更接近真实。
成本叠加：每个分支都要带上下文
并行分支各自调用模型，每个分支都要重新携带共享上下文。示例的共享上下文是 3,200 token，输入价 250 微元/百万 token，前缀复用率 60%，缓存输入价是全价的 10%。一个分支实际要付的上下文 token 是 3,200 × (1 − 0.6 × 0.9) = 1,472，成本 ceil(1,472 × 250 ÷ 1,000) = 368 微元。
| 复用率 | 每多一个分支的上下文成本 | 相对 3,786 微元 |
|---|---|---|
| 60% | 368 微元 | +9.72% |
| 30% | 584 微元 | +15.43% |
| 0% | 800 微元 | +21.13% |
368 微元的量级看起来不大，但它是每个并行分支、每次请求都要付的。三能力编排的主场景成本是 3,786 微元，多一个分支就多 9.72%；如果一条链路有三个并行分支，增量会接近三成。并行是用钱买时间，不是免费的加速。
并行度与收益的四个判据
把上面的量合成四个可判定的条件：
savedMs > 0：已经扣掉汇合开销和排队延迟，而不是只算Σdᵢ − max(dᵢ)。microsPerSavedMs ≤ 4：每省 1ms 最多多花 4 微元；这是产品给定的交换线，可以按毛利调整。reuseRatio ≥ breakEvenReuseRatio：示例的回本线是 41.67%，低于它时多分支的上下文成本超过交换线。- 新增并发槽的边际收益大于汇合开销，且不低于当前延迟的 5%。
四个条件全部成立才合并分支；任意一个不成立，保持串行或先做别的优化。判据的价值不在于算出一个小数，而在于让「并行更快」变成一句可以被数字反驳的话。
边际收益递减
四个分支耗时 300、260、180、120ms，串行 860ms。逐档增加并发槽：
| 并发槽 | 完成时间 | 相对上一档节省 |
|---|---|---|
| 1 | 860ms | — |
| 2 | 440ms | 420ms |
| 3 | 300ms | 140ms |
| 4 | 300ms | 0ms |
第一档到第二档收益最大，因为两个长分支开始重叠；第三档只压掉了 180ms 分支的尾巴；第四档没有收益，因为 300ms 的最长分支已经主导。并行度应该停在收益归零之前，而不是「能开多少开多少」。
实现
parallel.ts 把调度、上下文成本、边际收益和合并决策放在同一组纯函数里。调度是确定性的：同一份分支输入永远得到同一组开始与完成时间。
// file: src/orch/parallel.ts
// tier: L2
// verify: node --test test/orch/human.test.ts
import { AiError } from "../core/errors.ts";
export interface Branch {
id: string;
durationMs: number;
costMicros: number;
/** 该分支是否需要重新携带共享上下文。 */
carriesSharedContext: boolean;
}
export interface ParallelPlan {
branches: readonly Branch[];
/** 当前请求实际拿到的并发槽，不是提供方总并发。 */
concurrency: number;
/** 汇合点新增的同步、校验与交接延迟。 */
joinOverheadMs: number;
sharedContextTokens: number;
inputMicrosPerKTok: number;
prefixReuseRatio: number;
cachedInputPriceRatio: number;
/** 可选：整条链路的串行基线，用于折算端到端收益。 */
baselineMs?: number;
}
export interface BranchSchedule {
makeSpanMs: number;
starts: number[];
finishes: number[];
queueDelayMs: number;
utilization: number;
idealMs: number;
}
export interface ParallelEvaluation {
branches: number;
serialMs: number;
parallelMs: number;
savedMs: number;
speedup: number;
idealMs: number;
queueDelayMs: number;
extraCostMicros: number;
extraCostPct: number;
microsPerSavedMs: number | null;
breakEvenReuseRatio: number | null;
endToEndMs: number | null;
decision: "merge" | "keep-serial";
reason: string;
}
export interface ConcurrencyGain {
concurrency: number;
makeSpanMs: number;
savedVsPreviousMs: number;
}
function fail(message: string): never {
throw new AiError("invalid-request", `parallel: ${message}`);
}
function finite(value: number, field: string, min = 0): void {
if (!Number.isFinite(value) || value < min) fail(`${field} 必须是 >= ${min} 的有限数`);
}
function integer(value: number, field: string, min = 0): void {
if (!Number.isInteger(value) || value < min) fail(`${field} 必须是 >= ${min} 的整数`);
}
function ratio(value: number, field: string): void {
if (!Number.isFinite(value) || value < 0 || value > 1) fail(`${field} 必须落在 [0, 1]`);
}
function validate(plan: ParallelPlan): void {
if (plan.branches.length === 0) fail("至少需要一个分支");
const seen = new Set<string>();
for (const branch of plan.branches) {
if (branch.id.trim() === "") fail("branch.id 不能为空");
if (seen.has(branch.id)) fail(`重复的 branch.id ${branch.id}`);
seen.add(branch.id);
finite(branch.durationMs, `${branch.id}.durationMs`, 1);
integer(branch.costMicros, `${branch.id}.costMicros`);
}
integer(plan.concurrency, "concurrency", 1);
finite(plan.joinOverheadMs, "joinOverheadMs");
integer(plan.sharedContextTokens, "sharedContextTokens");
integer(plan.inputMicrosPerKTok, "inputMicrosPerKTok");
ratio(plan.prefixReuseRatio, "prefixReuseRatio");
ratio(plan.cachedInputPriceRatio, "cachedInputPriceRatio");
if (plan.baselineMs !== undefined) finite(plan.baselineMs, "baselineMs", 1);
}
/** 固定并发槽的确定性列表调度：按输入顺序放入最早空闲的槽。 */
export function scheduleBranches(plan: ParallelPlan): BranchSchedule {
validate(plan);
const slots = new Array<number>(plan.concurrency).fill(0);
const starts: number[] = [];
const finishes: number[] = [];
for (const branch of plan.branches) {
let slot = 0;
for (let index = 1; index < slots.length; index += 1) {
if (slots[index] < slots[slot]) slot = index;
}
starts.push(slots[slot]);
finishes.push(slots[slot] + branch.durationMs);
slots[slot] = finishes[finishes.length - 1];
}
const busyMs = plan.branches.reduce((sum, branch) => sum + branch.durationMs, 0);
const spanWithoutJoin = Math.max(...finishes);
const idealMs = Math.max(...plan.branches.map((branch) => branch.durationMs)) + plan.joinOverheadMs;
const makeSpanMs = spanWithoutJoin + plan.joinOverheadMs;
return {
makeSpanMs,
starts,
finishes,
queueDelayMs: makeSpanMs - idealMs,
utilization: spanWithoutJoin === 0 ? 0 : busyMs / (spanWithoutJoin * plan.concurrency),
idealMs,
};
}
export function sharedContextTokens(plan: ParallelPlan): number {
ratio(plan.prefixReuseRatio, "prefixReuseRatio");
ratio(plan.cachedInputPriceRatio, "cachedInputPriceRatio");
integer(plan.sharedContextTokens, "sharedContextTokens");
const factor = 1 - plan.prefixReuseRatio * (1 - plan.cachedInputPriceRatio);
return Math.ceil(plan.sharedContextTokens * factor);
}
/** 一个分支重新携带共享上下文的整数微元成本。 */
export function sharedContextCostMicros(plan: ParallelPlan): number {
integer(plan.inputMicrosPerKTok, "inputMicrosPerKTok");
return Math.ceil((sharedContextTokens(plan) * plan.inputMicrosPerKTok) / 1_000);
}
/** 配额共享：单请求可用槽 = 提供方总并发 ÷ 同时活跃请求数。 */
export function fairConcurrency(providerConcurrency: number, activeRequests: number): number {
integer(providerConcurrency, "providerConcurrency", 1);
integer(activeRequests, "activeRequests", 1);
return Math.max(1, Math.floor(providerConcurrency / activeRequests));
}
function syntheticPlan(
durations: readonly number[],
concurrency: number,
joinOverheadMs: number,
): ParallelPlan {
return {
branches: durations.map((duration, index) => ({
id: `b${index}`,
durationMs: duration,
costMicros: 0,
carriesSharedContext: false,
})),
concurrency,
joinOverheadMs,
sharedContextTokens: 0,
inputMicrosPerKTok: 1,
prefixReuseRatio: 0,
cachedInputPriceRatio: 0,
};
}
/** 并行度逐档增加时的完成时间；收益为 0 说明最长分支已主导。 */
export function marginalGains(
durations: readonly number[],
maxConcurrency: number,
joinOverheadMs = 0,
): ConcurrencyGain[] {
if (durations.length === 0) fail("至少需要一个分支耗时");
durations.forEach((duration, index) => finite(duration, `durations[${index}]`, 1));
integer(maxConcurrency, "maxConcurrency", 1);
finite(joinOverheadMs, "joinOverheadMs");
const results: ConcurrencyGain[] = [];
for (let concurrency = 1; concurrency <= Math.min(maxConcurrency, durations.length); concurrency += 1) {
const makeSpanMs = scheduleBranches(
syntheticPlan(durations, concurrency, joinOverheadMs),
).makeSpanMs;
const previous = results[results.length - 1]?.makeSpanMs;
results.push({
concurrency,
makeSpanMs,
savedVsPreviousMs: previous === undefined ? 0 : previous - makeSpanMs,
});
}
return results;
}
/** 并行是否值得：延迟收益、上下文成本、复用回本线与端到端折算。 */
export function evaluateParallel(
plan: ParallelPlan,
options: { maxMicrosPerSavedMs?: number } = {},
): ParallelEvaluation {
validate(plan);
const threshold = options.maxMicrosPerSavedMs ?? 4;
finite(threshold, "maxMicrosPerSavedMs");
const schedule = scheduleBranches(plan);
const serialMs = plan.branches.reduce((sum, branch) => sum + branch.durationMs, 0);
const savedMs = serialMs - schedule.makeSpanMs;
const carrying = plan.branches.filter((branch) => branch.carriesSharedContext).length;
const oneCopy = sharedContextCostMicros(plan);
const extraCostMicros = oneCopy * Math.max(0, carrying - 1);
const branchCost = plan.branches.reduce((sum, branch) => sum + branch.costMicros, 0);
const serialTotal = branchCost + (carrying > 0 ? oneCopy : 0);
const microsPerSavedMs = savedMs > 0 ? extraCostMicros / savedMs : null;
let breakEvenReuseRatio: number | null = null;
if (savedMs > 0 && carrying > 1) {
const fullCopy = ((carrying - 1) * plan.sharedContextTokens * plan.inputMicrosPerKTok) / 1_000;
const budget = threshold * savedMs;
if (budget >= fullCopy) {
breakEvenReuseRatio = 0;
} else if (plan.cachedInputPriceRatio < 1) {
const value = (1 - budget / fullCopy) / (1 - plan.cachedInputPriceRatio);
breakEvenReuseRatio = value > 1 ? null : Number(value.toFixed(4));
}
}
const merge = savedMs > 0 && (extraCostMicros === 0 ||
(microsPerSavedMs !== null && microsPerSavedMs <= threshold));
return {
branches: plan.branches.length,
serialMs,
parallelMs: schedule.makeSpanMs,
savedMs,
speedup: schedule.makeSpanMs === 0 ? 0 : serialMs / schedule.makeSpanMs,
idealMs: schedule.idealMs,
queueDelayMs: schedule.queueDelayMs,
extraCostMicros,
extraCostPct: serialTotal === 0 ? 0 : Number(((extraCostMicros / serialTotal) * 100).toFixed(2)),
microsPerSavedMs: microsPerSavedMs === null ? null : Number(microsPerSavedMs.toFixed(2)),
breakEvenReuseRatio,
endToEndMs: plan.baselineMs === undefined ? null : plan.baselineMs - savedMs,
decision: merge ? "merge" : "keep-serial",
reason: savedMs <= 0
? `并行后反而慢 ${Math.abs(savedMs)}ms，保持串行`
: merge
? `每省 1ms 多花 ${microsPerSavedMs?.toFixed(2) ?? "0"} 微元，未超过 ${threshold}`
: `每省 1ms 多花 ${microsPerSavedMs?.toFixed(2) ?? "0"} 微元，超过 ${threshold}`,
};
}
scheduleBranches 用最早空闲槽做列表调度，因此槽数不足时会把排队延迟显式算出来。evaluateParallel 不把「并行」当成默认正确：它先算节省，再算多花的上下文成本，最后用交换线和回本复用率给出 merge 或 keep-serial。
示例样本上的运行结果
两个分支 260ms 与 160ms、两槽、汇合 35ms、复用率 60% 的样本：
| 指标 | 数值 |
|---|---|
| 串行段耗时 | 420ms |
| 并行段耗时 | 295ms |
| 节省 | 125ms |
| 端到端基线 1,269ms 折算 | 1,144ms |
| 多花上下文成本 | 368 微元 |
| 每省 1ms 的成本 | 2.94 微元 |
| 回本复用率 | 41.67% |
| 决策 | merge |
同一拆分在 30% 复用率下每省 1ms 要花 4.67 微元，超过 4 的交换线，决策翻转为 keep-serial；只有一个并发槽时耗时 455ms，节省 −35ms，同样保持串行。这说明并行方案不是「拆了就赚」，而是依赖复用率与配额两个外部条件。
回声链路上的三个落点
内容发布链 A 的 moderation.classify 拆成政策分类与敏感词扫描，在 60% 复用率与两槽下值得合并；若折算到全书共享的 1,269ms 编排基线，等效端到端从 1,269ms 降到 1,144ms。私信客服链 B 的 support.assist 可以预取订单快照，与回复草稿并行，但只省 90ms、多花 368 微元，合 4.09 微元/毫秒，刚好超过交换线，示例里保持串行。站内下单咨询链 C 的 search.rerank 是 chat.reply 的前置条件，没有可并行的独立分支，强行拆只会制造假依赖。
人工复核不在这张并行图里。它的耗时以分钟计，应该走异步任务与人工节点，而不是塞进 1,269ms 的同步链路；这是 6.2 与 6.3 的主题。
小结
并行分支的收益由临界路径决定，成本由共享上下文复制决定，稳定性由并发配额决定。示例样本里，420ms 的串行段拆成 260ms 与 160ms 后省 125ms，但多花 368 微元，合 2.94 微元/毫秒；复用率低于 41.67% 或只剩一个并发槽时，这个拆分不再划算。
判据可以压缩成一句话：先算扣掉汇合与排队后的节省，再算每个分支的上下文成本，最后看复用率是否越过回本线。
阅读导航：上一节：5.3 编排级重试预算与熔断 · 下一节：6.2 人工介入节点 。
常见错误与边界
- 只算
Σdᵢ − max(dᵢ)，忘掉汇合开销。示例里 160ms 的理论节省会缩水到 125ms。 - 按提供方总并发做规划。单请求只能拿到总并发除以活跃请求数的份额。
- 假设低峰配额等于高峰配额。高峰时槽数下降，原本的收益会变成负值。
- 把共享上下文当成只传一次。每个并行分支都要重新携带，复用率越低越贵。
- 用单次调用成本评价拆分。应看每次成功交付的成本，把多分支的上下文计入。
- 并行度开到最长分支之外。超过最长分支的槽不会缩短完成时间。
- 为了「结构对称」强行拆分没有独立性的步骤。C 链的检索与回复有前置依赖，拆了只是多一次交接。
- 把示例的 41.67% 当成通用常数。token 结构与缓存价一变，回本线就要重算。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。