• 简体中文
  • v6.5.2
  • 编排

    编排是模型驱动委派的可编程姊妹能力。在 TasksTeams 中,由 LLM 在运行时决定是否调用 task / parallel_task——扇出的形状取决于模型的选择。编排把这个决策搬进你的代码:你 用一套语法表达扇出、流水线与验证面板,因此形状是可复现、可测试、受预算约束 且可恢复的——与模型的临场选择无关。

    当工作的结构是宿主提前已知的(并行跑这三个 reviewer;让每个候选项依次经过 explore → verify → review;崩溃后恢复这一批),就用编排。当你希望由模型决定是否 以及如何委派时,就用 Tasks/Teams。

    框架 / 宿主边界(接缝)

    本层的一切都围绕单一接缝 AgentExecutor 编写:"运行这个 step,把结果给我。"这 条接缝把职责切得很干净:

    • 框架拥有语法——有哪些 step、如何组合、并发提示,以及可序列化契约 AgentStepSpec / StepOutcome
    • 宿主拥有放置——传输、调度,以及 step 实际在哪里运行。

    内置的默认 executor(TaskExecutor)在本地、进程内、基于 tokio 运行每个 step。 宿主可以替换为自己的 AgentExecutor,把 step 放置到集群各处;组合子从不观察 step 在哪里运行,因此同一套编排无需改动即可从单进程扩展到集群。

    concurrency_hint()建议性的,而非硬性本地上限——正是它让编排能扩展到单进 程之外(由调度器支撑的宿主会返回其集群级目标,而不是本地上限)。

    SDK 已为你接好这一切:AgentSession::agent_executor() 返回由 session 支撑的 executor(它把每个 step 作为子 agent 在本节点运行,继承 session 的 agent registry、 LLM client、workspace 与 MCP 工具),session_store() 返回 session 的 store。 parallel / pipeline / parallelResumable 方法会替你调用它们。

    步骤契约

    一个 step 由 AgentStepSpec 描述,并解析为 StepOutcome。两者都刻意保持可序列 化:宿主可以把 spec 发送到另一个节点,可恢复组合子也会把 outcome 持久化进 checkpoint。

    AgentStepSpec 字段:

    • task_id——step 的稳定 id(由你指定);会流入生命周期事件与 checkpoint。
    • agent——要运行的 agent 的 registry key(例如 explorereview)。
    • description——用于展示/追踪的简短人类标签。
    • prompt——交给子 agent 的指令。
    • max_steps(可选)——每个 step 的 tool-round 上限。
    • parent_session_id(可选)——用于事件关联的父 session id。
    • output_schema(可选)——设置后,步骤必须返回符合此 JSON Schema 的值(见 强制模式约束的步骤输出)。

    StepOutcome 字段:

    • task_id——产生该结果的 step id。
    • session_id——子运行的 session id(失败的 step 仍可寻址)。
    • agent——运行的 agent。
    • output——step 的文本输出。
    • success——失败或 panic 的 step 为 false(绝不会丢弃兄弟 step)。
    • structured(可选)——经模式校验的对象,仅当任务规格携带 output_schema 时存在。

    key 的大小写风格因 SDK 而异:

    概念Node(camelCase)Python(snake_case)
    step idtaskIdtask_id
    tool-round 上限maxStepsmax_steps
    父 sessionparentSessionIdparent_session_id
    强制模式约束outputSchemaoutput_schema

    agentdescriptionpromptoutputsuccessstructured 在两个 SDK 中拼写相同。

    parallel:带屏障的扇出 [#parallel--barrier-fan-out]

    session.parallel(specs) 把每个任务规格作为扇出分支运行,并按输入顺序解析出每 个 spec 对应的一个 StepOutcome。它映射到核心组合子 execute_steps_parallel。它 是一个屏障:在返回前会等待每个 step。

    每个分支彼此隔离——失败或 panic 的 step 会变成 success: false,绝不会丢弃兄弟 step。并发受 executor 的并发提示约束(默认即 session 配置的并行度)。

    TypeScript
    const outcomes = await session.parallel([
    {
    taskId: 'explore',
    agent: 'explore',
    description: 'Risky changes',
    prompt: 'Find risky changed files in this diff.',
    },
    {
    taskId: 'verify',
    agent: 'verification',
    description: 'Test gaps',
    prompt: 'Identify missing or weak verification.',
    },
    {
    taskId: 'review',
    agent: 'review',
    description: 'Correctness',
    prompt: 'Review the diff for correctness risks.',
    },
    ]);
    for (const outcome of outcomes) {
    if (outcome.success) {
    console.log(outcome.taskId, outcome.output);
    } else {
    console.warn('failed:', outcome.taskId, outcome.output);
    }
    }
    Python
    outcomes = session.parallel([
    {"task_id": "explore", "agent": "explore", "description": "Risky changes",
    "prompt": "Find risky changed files in this diff."},
    {"task_id": "verify", "agent": "verification", "description": "Test gaps",
    "prompt": "Identify missing or weak verification."},
    {"task_id": "review", "agent": "review", "description": "Correctness",
    "prompt": "Review the diff for correctness risks."},
    ])
    for outcome in outcomes:
    if outcome["success"]:
    print(outcome["task_id"], outcome["output"])
    else:
    print("failed:", outcome["task_id"], outcome["output"])

    pipeline:阶段之间无屏障

    session.pipeline(items, stages) 让每个 item 独立地流经一连串阶段——阶段之间 没有屏障,因此 item A 可以处于 stage 3,而 item B 还在 stage 1。墙钟时间是最慢的 单条链,而不是逐阶段屏障会带来的"每阶段最慢之和"。

    阶段是 spec 构造器,而非 spec:每个阶段接收上一个 outcome 和原始 item,返回要运行 的下一个 step,或返回 null / None 提前停止该 item 的链。失败的 step 同样会停止 链(后续阶段只会基于失败结果继续)。回调形状:

    • Node:(ctx) => spec | null,其中 ctx = { previous: StepOutcome | null, item }
    • Python:stage(ctx) -> spec | None,其中 ctx = {"previous": <dict|None>, "item": <item>}

    阶段可以基于上一个 outcome 分支——例如"验证 review 阶段产出的发现"。

    约束(来自源码):

    • 流水线阶段不支持逐阶段 output_schema——需要模式校验的步骤请用 parallel
    • Node: 阶段回调绝不能 throw——一次 throw 会中止进程(与 setBudgetGuard 相同的约束)。请把逻辑包进 try/catch,出错时 return null
    • Node: 超过 timeoutMs(第 3 个参数,默认 30000)仍挂起的阶段会 fail closed——被当作 null 处理,仅停止该条链。
    • Python: 抛异常的阶段 callable 会被捕获并当作 None(仅停止该条链)。

    Node pipeline 阶段回调绝不能 throw。 在当前 napi 版本中,返回值转换时的 JS throw 会中止进程(与 setBudgetGuard 相同的 fail-closed 约束)。务必把阶段逻辑 包进 try/catch 并在出错时 return null

    TypeScript
    const outcomes = await session.pipeline(
    ['src/auth.ts', 'src/payments.ts'],
    [
    (ctx) => ({
    taskId: `explore-${ctx.item}`,
    agent: 'explore',
    description: 'Inspect file',
    prompt: `Summarize the responsibilities and risks of ${ctx.item}.`,
    }),
    (ctx) => {
    try {
    if (!ctx.previous) return null;
    return {
    taskId: `review-${ctx.item}`,
    agent: 'review',
    description: 'Review of prior finding',
    prompt: `Review this summary for correctness risks:\n${ctx.previous.output}`,
    };
    } catch {
    return null; // stages must not throw
    }
    },
    ],
    );
    Python
    def explore_stage(ctx):
    item = ctx["item"]
    return {
    "task_id": f"explore-{item}",
    "agent": "explore",
    "description": "Inspect file",
    "prompt": f"Summarize the responsibilities and risks of {item}.",
    }
    def review_stage(ctx):
    prev = ctx["previous"]
    if prev is None:
    return None
    item = ctx["item"]
    return {
    "task_id": f"review-{item}",
    "agent": "review",
    "description": "Review of prior finding",
    "prompt": f"Review this summary for correctness risks:\n{prev['output']}",
    }
    outcomes = session.pipeline(
    ["src/auth.ts", "src/payments.ts"],
    [explore_stage, review_stage],
    )

    可恢复 / 可迁移工作流

    session.parallelResumable(specs, workflowId)(Node)/ session.parallel_resumable(specs, workflow_id)(Python)是 parallel 加上一份 日志。它映射到 execute_steps_parallel_resumable

    它在每个 step 边界把一份 WorkflowCheckpoint 写入 session store。恢复时它跳过已完 成的 step(复用其缓存的 outcome),只重新派发其余 step。它只记录成功的 step—— 失败的 step 不入日志,因此恢复时会重试。完全成功后 checkpoint 会被删除;只有崩溃才 会留下一份供恢复。

    由于 checkpoint 可序列化、executor 是一个参数,宿主可以通过传入另一个节点的 executor,在另一个节点上恢复被中断的工作流(迁移)。

    该组合子要求已配置 session store——两个 SDK 方法在缺少 store 时都会 reject/ raise(Node 的错误信息是 parallelResumable requires a sessionStore)。

    WorkflowCheckpoint 的模式字段为 schema_version / workflow_id / steps / checkpoint_ms。由未来的、不兼容的 schema_version 写入的检查点在加载时 会被拒绝(ensure_loadable)。该失败是 fail-safe 而非致命的:不可读的 checkpoint 会记录一条 warning,工作流从头重跑,而不是从它无法解释的状态恢复。

    store 相关见 Persistence,迁移路径见 Multi-Machine

    TypeScript
    import { Agent, FileSessionStore } from '@a3s-lab/code';
    const agent = await Agent.create('agent.acl');
    const session = agent.session('/repo', {
    sessionStore: new FileSessionStore('./.a3s/sessions'),
    });
    // 第一次尝试可能会在中途被打断。
    let outcomes = await session.parallelResumable(specs, 'release-batch-42');
    // 崩溃或重启后:使用同一个 workflowId 恢复,并跳过已完成的步骤。
    outcomes = await session.parallelResumable(specs, 'release-batch-42');
    Python
    from a3s_code import Agent, FileSessionStore, SessionOptions
    agent = Agent.create("agent.acl")
    opts = SessionOptions()
    opts.session_store = FileSessionStore("./.a3s/sessions")
    session = agent.session("/repo", opts)
    # 第一次尝试可能会在中途被打断。
    outcomes = session.parallel_resumable(specs, "release-batch-42")
    # 崩溃或重启后:使用同一个 workflow_id 恢复,并跳过已完成的步骤。
    outcomes = session.parallel_resumable(specs, "release-batch-42")

    跨扇出任务的共享预算 [#shared-budget]

    默认情况下,每个子代理各自统计自己的 LLM 成本。给 parallel 传入一个 token 预算后,所有子智能体改为汇入同一个账本——为整个扇出设一个统一上限。它映射 到核心的 WorkflowBudget:一个安装到每个子运行上的、聚合型 BudgetGuard

    预算是一个可选参数,因此向后兼容:

    • 不传预算时,parallel(specs) 返回原来的结果数组,与之前完全一致。
    • 传入预算时,parallel(specs, budgetTokens) 解析为 { outcomes, budget }, 其中 budget 是账本快照(consumedTokens / limitTokens)。

    一旦达到上限,之后启动的 step 会被拒绝——其结果为 success: false,并带有预算 耗尽信息。它是一个软上限:由于用量是在每次 LLM 调用之后记账的,宽扇出 可能在账本追上之前冲过上限几个在飞回合。框架绝不强杀进行中的扇出;预算耗尽 只是拒绝下一次 LLM 调用。

    TypeScript
    // 不传预算 → 原样返回结果数组(行为不变)。
    const outcomes = await session.parallel(specs);
    // 传入预算 → { outcomes, budget }。所有子代理共享一个账本。
    const { outcomes: out, budget } = await session.parallel(specs, 500_000);
    console.log(budget.consumedTokens, budget.limitTokens); // 例如 48213, 500000
    // TS 返回类型是联合类型:Array<StepOutcome> | { outcomes, budget }。
    Python
    # 不传预算 → 原样返回列表(行为不变)。
    outcomes = session.parallel(specs)
    # 传入预算 → {"outcomes": [...], "budget": {"consumed_tokens", "limit_tokens"}}。
    res = session.parallel(specs, budget_tokens=500_000)
    print(res["budget"]["consumed_tokens"], res["budget"]["limit_tokens"])

    循环直到完成(execute_loop)[#looping]

    对于长度未知、需要迭代至收敛的工作(循环直至没有剩余项、反复打磨直到满意),核心语法 提供了 execute_loop。每一轮都是一个屏障(execute_steps_parallel);宿主提供的 谓词看到本轮的结果,返回 LoopDecision::Continue(next_specs)LoopDecision::Stop。必填的 max_iterations 是一个硬上限——一旦达到,即使谓词 还想继续也会停止,从而让 LLM 驱动的循环永远不会失控。

    Rust
    use a3s_code_core::orchestration::{execute_loop, AgentStepSpec, LoopDecision};
    let outcomes = execute_loop(executor, initial_specs, /* max_iterations */ 5, None, |round| {
    // 本轮没有新发现就停止;否则扇出后续 step。
    let follow_ups = derive_follow_ups(round);
    if follow_ups.is_empty() {
    LoopDecision::Stop
    } else {
    LoopDecision::Continue(follow_ups)
    }
    })
    .await;

    从宿主 SDK 你并不需要专门的 loop 动词——直接用你自己语言里的 while/for 围绕 parallel 写循环,根据结果决定下一轮即可。execute_loop 是为 Rust 语法 而存在,并为循环提供一个单一、强制的终止守卫。

    工作流外观接口(Rust / 嵌入)[#workflow-facade]

    session.workflow() 返回一个可廉价克隆的 Workflow,它预先接好了会话的 executor、持久化 store、逐 step 事件流,以及一个稳定的、由会话派生的 root id。它是 把以上能力打包起来的可编程句柄;控制流就是普通 Rust——await 一个动词、查看结果、 决定下一步运行什么。

    • 动词——agent(单步)、parallel(屏障式扇出)、phase命名的、 可恢复的屏障,并发出里程碑)、pipeline(按条目的链),以及不会失败的 log。 每个动词都只委派给一个 combinator。
    • Phase 与事件——phase(name, specs) 派生确定性 checkpoint id ({root}/{index}:{name}),在配置了 store 时走可恢复屏障,并在一个广播上发出 WorkflowEvent::PhaseStart / PhaseEnd,你可用 subscribe() 读取。log() 发出 WorkflowEvent::Log
    • 预算——session.workflow_with_token_budget(Some(limit)) 安装一个共享的 WorkflowBudgetbudget_snapshot() 读取账本,达到上限时会触发 WorkflowEvent::BudgetExhausted
    Rust
    let wf = session.workflow(); // 或 session.workflow_with_token_budget(Some(500_000))
    let mut events = wf.subscribe();
    // 先跑一步,再根据其结果计算出一个*可变*数量的扇出——这正是“动态”所在:
    // 形状在运行时决定,而非提前声明。
    let plan = wf.agent(AgentStepSpec::new("plan", "plan", "plan", goal)).await;
    let specs = derive_specs(&plan); // 你的代码
    let done = wf.phase("implement", specs).await; // 可恢复屏障 + 里程碑
    let reviews = wf.phase("review", to_review(&done)).await; // 预算在各 phase 间共享
    if let Some(b) = wf.budget_snapshot() {
    println!("spent {} / {:?} tokens", b.consumed_tokens, b.limit_tokens);
    }

    SDK 暴露的是扁平的 parallel / pipeline / parallelResumable 动词(以及上面 parallel 的预算重载);完整的 Workflow 句柄——phases、事件订阅、loop combinator——属于 Rust / 嵌入层 API。

    受模式约束的步骤输出 [#schema-forced-step-output]

    携带 output_schema(Node 中为 outputSchema)的任务规格会强制步骤返回符合该 JSON Schema 的值;经校验的对象落在 StepOutcome.structured 中。这复用了与 A3S Code 其余 部分相同的结构化输出强转 + 修复机制。强转失败会把该 step 降级为不成功success: false),因此调用方绝不会把未经校验的文本当作承诺的对象。

    强制模式约束仅适用于 parallel / parallelResumable 的任务规格——适用于 pipeline 阶段。

    TypeScript
    const [outcome] = await session.parallel([
    {
    taskId: 'triage',
    agent: 'review',
    description: 'Structured triage',
    prompt: 'Triage this diff.',
    outputSchema: {
    type: 'object',
    properties: {
    severity: { type: 'string', enum: ['low', 'medium', 'high'] },
    summary: { type: 'string' },
    },
    required: ['severity', 'summary'],
    },
    },
    ]);
    if (outcome.success) {
    console.log(outcome.structured.severity, outcome.structured.summary);
    }
    Python
    outcomes = session.parallel([
    {
    "task_id": "triage",
    "agent": "review",
    "description": "Structured triage",
    "prompt": "Triage this diff.",
    "output_schema": {
    "type": "object",
    "properties": {
    "severity": {"type": "string", "enum": ["low", "medium", "high"]},
    "summary": {"type": "string"},
    },
    "required": ["severity", "summary"],
    },
    },
    ])
    outcome = outcomes[0]
    if outcome["success"]:
    print(outcome["structured"]["severity"], outcome["structured"]["summary"])

    成本治理与生命周期

    编排 step 走的是同一个 session,因此 session 的各项控制对它们直接生效。 setBudgetGuard(Node)/ budget_guard(Python)约束每个 step 的 LLM 成本; close() 会连同 session 的其余工作一起取消进行中的 step;宿主提供的身份标签 (tenant_idprincipalagent_template_idcorrelation_id)会贯穿每个 step, 供宿主侧聚合与计费使用。这些控制的细节见 SessionsLimits