企业官网建设流程全解析

Conductor 多智能体架构完全指南：九种子智能体编排策略与可运行示例

【免费下载链接】conductorConductor is an event driven agentic workflow engine providing durable and highly resilient execution engine for applications and AI Agents项目地址: https://gitcode.com/GitHub_Trending/co/conductor

导读：本文以 Conductor 官方文档 multi-agent-architecture.md 为骨架，完整讲解 Conductor 中多智能体（Multi-Agent）系统的核心模型——一个父智能体（parent agent）＋一组子智能体（sub-agents）＋一个strategy字段。你将掌握九种编排策略（handoff、router、sequential、parallel、swarm、round_robin、random、plan_execute、manual）的选型依据、代码写法与编译产物，并结合仓库源码与 cookbook 中的可运行示例，理解 Conductor 如何把一次智能体委托编译为一次可持久、可重试、可观测的工作流执行。

一、核心模型：一个父智能体 + 一份策略

在 Conductor 中，多智能体系统的定义极其简洁：一个父智能体持有一组子智能体的列表，外加一个**策略（strategy）**字段，用来决定这些子智能体如何被调度运行。策略只是一个字段，而其余的工程能力——持久化（durability）、重试（retries）、每次委托的可见性（visibility）——全部来自 Conductor 将整个多智能体系统编译成一份普通工作流（workflow）这一事实。

support = Agent( name="support_supervisor", model="openai/gpt-4o-mini", instructions="Route each request to the right specialist.", agents=[billing, technical, sales], strategy=Strategy.HANDOFF, )

这段代码来自官方文档中的示例：support_supervisor是父智能体，billing、technical、sales是子智能体，strategy=Strategy.HANDOFF声明了调度策略。在 agent_handoff.py 中可以看到同一模型下三个子智能体的完整定义——每个子智能体拥有独立的instructions（指令）与tools（工具）作用域。

从仓库源码结构看，这套 Agent 抽象同时存在于 Python、Java、TypeScript、C# 四个 SDK 中（本文引用的是 docs/devguide/ai/conductor-agents.md 中描述的 SDK 行为）。所谓"编译成工作流"，指的是：你通过 SDK 的Agent类（或受支持的 agent framework 对象）定义智能体后，Conductor 会把它编译为一份可检查的 workflow graph，进而作为可复用的AGENT任务（task）被工作流引用。

二、选择策略：决定权在模型、图还是你

多智能体编排的核心分水岭问题是：谁来拍板——是模型（Model）、编排图（Graph），还是你（代码中的业务规则）。官方文档给出的九种策略对照表如下，这是选型的第一依据：

两条实战要点需要特别记住：

• router比handoff更便宜：router只做分类与分发（classify and dispatch），不把对话交给子智能体。当分类之后无需继续对话时，应优先使用router以节省一轮对话的模型开销。
• plan_execute是唯一会"重新规划"（replan）的策略：其余策略一旦做出分发决定就提交（commit）了；只有plan_execute在执行过程中会根据返回结果修订后续计划。

三、四种策略形状：模型挑选 / 图运行全部 / 智能体互相交接 / 规划-执行-再规划

九种策略在形态上可归为四类，官方文档给出了每类的代码骨架：

3.1 "模型挑选一个"：handoff与router

子智能体以**可调用工具（callable tools）**的形式暴露给父智能体的模型，模型按名字挑选一个。

support = Agent( name="support", model=MODEL, instructions="Route to billing, technical, or sales.", agents=[billing, technical, sales], strategy=Strategy.HANDOFF, # or Strategy.ROUTER )

3.2 "图运行全部"：sequential与parallel

模型不参与顺序决策——顺序由编排图决定。

pipeline = Agent( name="review_pipeline", model=MODEL, agents=[researcher, writer, editor], strategy=Strategy.SEQUENTIAL, # or Strategy.PARALLEL )

sequential按顺序运行全部子智能体，每个子智能体都能看到上一个的输出；parallel同时运行全部并收集每个答案。

3.3 "智能体互相交接"：swarm

控制权在子智能体之间传递，直到其中一个产出最终答案为止。

swarm = Agent( name="triage_swarm", model=MODEL, agents=[intake, diagnosis, resolution], strategy=Strategy.SWARM, )

3.4 "规划、执行、再规划"：plan_execute

模型先生成一份子智能体调用计划，执行它，并在结果返回后修订计划。

planner = Agent( name="incident_planner", model=MODEL, agents=[log_reader, metrics_reader, remediation_drafter], strategy=Strategy.PLAN_EXECUTE, )

适用于步骤无法预先确定（not knowable up front）的场景，例如事件排查类任务。

四、Conductor 在策略之上添加了什么

官方文档明确指出，多智能体框架本身只解决"谁决定、怎么跑"，而 Conductor 把策略编译成工作流之后，额外获得四项关键能力：

1. 每次委托都是一次独立的执行（Each delegation is its own execution）：某个专家（specialist）可以单独重试，而不需要重跑路由决策。这意味着路由是幂等的、可恢复的，一次子智能体失败不会连坐父智能体的分类逻辑。
2. 选择被记录下来（The choice is recorded）：哪个子智能体跑了、为什么跑，都存在于执行记录（execution）中，而不只是一行日志。这直接支持了事后审计与 Agent 评估（Evals）。
3. 子智能体保留自己的工具与护栏（Sub-agents keep their own tools and guardrails）：例如 billing 智能体无法触达 fulfilment 工具——每个子智能体的tools、instructions、护栏是隔离的作用域。这正对应 cookbook agent-handoff.md 生产建议中"分别限定每个专家的工具范围"。
4. 并行是真并行（Parallel means actually parallel）：parallel与扇出（fan-out）编译为FORK_JOIN，而不是一个串行循环。

关于第 4 点，仓库中有两处直接证据：

• 静态并行：ai/examples/33-conductor-agent-multi-agent.json 展示了一个框架无关的多智能体工作流：FORK_JOIN扇出到两个已部署的专家AGENT任务（run_planner、run_researcher），随后JOIN收集两个分支的结果。每条分支拥有独立的 executionId，轮询时不会阻塞 worker 线程：

{ "name": "fork_agents", "taskReferenceName": "fork_agents_ref", "type": "FORK_JOIN", "forkTasks": [ [ { "name": "run_planner", "taskReferenceName": "run_planner_ref", "type": "AGENT", "inputParameters": { "agentType": "conductor", "name": "planner", "prompt": "${workflow.input.prompt}" } } ], [ { "name": "run_researcher", "taskReferenceName": "run_researcher_ref", "type": "AGENT", "inputParameters": { "agentType": "conductor", "name": "researcher", "prompt": "${workflow.input.prompt}" } } ] ] }

• 动态扇出：agent_scatter_gather.py 的 docstring 说明：scatter_gather()构建的协调者通过FORK_JOIN_DYNAMIC将 worker 智能体分发 N 次，其中 N 由模型在运行时决定，而不是硬编码在图里。静态并行用FORK_JOIN，运行时才能确定分支数的扇出用FORK_JOIN_DYNAMIC——两者都是真并行，而非循环。

更深一层，Conductor 的 Agent 运行时模型（见 docs/devguide/ai/index.md）保证"模型的输出是提议（proposal）而非命令"：模型提议的每一步都要经过校验、审批（approvals）、护栏后才调度为可执行任务；每一步的结果在下一轮开始前持久化，因此崩溃、发布或长时间等待都不会丢失智能体的进度。这正是"每次委托都是一次独立执行"的底层机制。

五、实战一：handoff监督者-专家模式（可运行）

官方文档在"Next steps"中指向 agent-handoff.md，该 cookbook 提供了一个开箱即用的监督者（supervisor）+ 三个专家的完整示例，源码在 agent_handoff.py：

from conductor.ai.agents import Agent, AgentRuntime, Strategy, tool MODEL = "openai/gpt-4o-mini" @tool def check_balance(account_id: str) -> dict: """Check the balance of a bank account.""" return {"account_id": account_id, "balance": 5432.10, "currency": "USD"} @tool def lookup_order(order_id: str) -> dict: """Look up the status of an order.""" return {"order_id": order_id, "status": "shipped", "eta": "2 days"} @tool def get_pricing(product: str) -> dict: """Get pricing information for a product.""" return {"product": product, "price": 99.99, "discount": "10% off"} billing = Agent( name="billing", model=MODEL, instructions="You handle billing questions: balances, payments, invoices.", tools=[check_balance], ) technical = Agent( name="technical", model=MODEL, instructions="You handle technical questions: order status, shipping, returns.", tools=[lookup_order], ) sales = Agent( name="sales", model=MODEL, instructions="You handle sales questions: pricing, products, promotions.", tools=[get_pricing], ) support = Agent( name="support_supervisor", model=MODEL, instructions="Route each request to the right specialist: billing, technical, or sales.", agents=[billing, technical, sales], strategy=Strategy.HANDOFF, ) if __name__ == "__main__": with AgentRuntime() as runtime: result = runtime.run(support, "What's the balance on account ACC-123?") result.print_result() print("execution id:", result.execution_id)

运行前提（cookbook 原文）：一个已配置 LLM provider 的 Conductor 服务器，并设置好CONDUCTOR_SERVER_URL。运行方式：

python agent_handoff.py

向support_supervisor询问账户余额，会被路由到billing，后者调用check_balance。打开Executions界面可以看到监督者与被选中的专家是两个独立的执行——这正是第 4 节"每次委托都是独立执行"的直接体现。

cookbook 还给出了生产环境要点，其中与本文策略选型强相关的是：

• 专家指令就是路由信号：描述重叠会导致错误的交接（wrong handoffs）；
• router在不需要对话时比handoff更便宜——按问题形态选策略，而不是为了求新；
• handoff 决策本质是模型输出：要记录哪个专家跑了、为什么跑（对应"选择被记录"能力）；
• 分别限制每个专家的预算，避免单个专家耗尽整体预算。

六、实战二：scatter_gather百路并行（可运行）

针对parallel的极端形态，官方文档在"Next steps"中指向 agent-scatter-gather.md：一个协调者（coordinator）把一次请求分解成 100 个独立子任务，全部并行运行（每个子任务是一个独立的子工作流，拥有自己的重试），最后汇总结果。源码在 agent_scatter_gather.py：

from conductor.ai.agents import Agent, AgentRuntime, scatter_gather, tool MODEL = "openai/gpt-4o-mini" SYNTHESIS_MODEL = "openai/gpt-4o" # larger context, it sees all 100 results @tool def search_knowledge_base(query: str) -> dict: """Look up a topic. Replace with a real search or vector-DB call.""" return { "query": query, "results": [ f"{query}: mid-sized economy with a services-led profile", f"{query}: population growth close to the regional average", ], } researcher = Agent( name="country_researcher", model=MODEL, instructions=( "You profile one country. Call search_knowledge_base exactly once, then " "write 2-3 sentences covering economy, population and one distinctive fact. " "Do not call the tool more than once." ), tools=[search_knowledge_base], max_turns=5, ) COUNTRIES = ['Afghanistan', 'Albania', 'Algeria', 'Andorra', 'Angola', 'Argentina', ...] # 100 个国家 country_list = "\n".join(f"{i + 1}. {c}" for i, c in enumerate(COUNTRIES)) coordinator = scatter_gather( name="country_coordinator", worker=researcher, model=SYNTHESIS_MODEL, instructions=( f"Create EXACTLY {len(COUNTRIES)} country_researcher calls, one per country " f"below, passing just the country name. Issue ALL calls in a SINGLE response.\n\n" f"Countries:\n{country_list}\n\n" f"When all {len(COUNTRIES)} results are back, compile a short report grouped " f"by region." ), retry_count=3, retry_delay_seconds=5, timeout_seconds=900, ) if __name__ == "__main__": with AgentRuntime() as runtime: result = runtime.run( coordinator, f"Profile all {len(COUNTRIES)} countries in the list.", ) result.print_result() print("execution id:", result.execution_id)

该示例的几个关键参数与机制：

• scatter_gather()为你构建协调者：分解（decompose）→ 扇出（fan out）→ 汇总（synthesize）三步被封装成一个函数；
• 扇出宽度由模型在运行时决定（Create EXACTLY ... calls, one per country的指令 +FORK_JOIN_DYNAMIC），不是硬编码在图中；
• 每个子任务都是独立的子工作流，拥有自己的重试；
• 部分结果默认可用：fail_fast=False（该 API 的默认行为）意味着一个 worker 失败不会拖垮整个批次；
• 使用更大的模型做汇总（SYNTHESIS_MODEL），因为它要一次性读取全部 100 份结果；
• max_turns限制 worker 的轮数，防止单个 worker 死循环拖住 JOIN；retry_count、retry_delay_seconds、timeout_seconds分别控制重试次数、重试延迟与总超时。

生产注意（cookbook 原文）：限流先于引擎成为瓶颈——100 个并发调用会先触达 provider 配额；汇总阶段的上下文窗口要控制（worker 输出保持简短）；97/100 的部分成功对你的调用方意味着什么，要在上线前决策；成本随数量线性增长，建议先用 5 个 worker 验证形态再跑 100 个。

七、在 AGENT 任务中编排已部署的多智能体

当多智能体系统被deploy到服务器后，工作流通过AGENT任务按名字调用（详见 conductor-agents.md）：

• agentType: "a2a"（默认）调用远程 A2A 端点；
• agentType: "conductor"按name运行已部署的 Conductor Agent。

AGENT任务输出executionId、agentName、state、text以及运行完成时的结构化output；其state是标准化的 A2A 生命周期值：working、input-required、completed、failed、canceled。maxDurationSeconds（默认 86400 秒）与maxPollFailures（默认 30 次）为全量运行的兜底护栏。这一契约是理解"每次委托是一次独立执行"的服务端形态：一次委托对应一个可轮询、可恢复、可取消的执行。

注意：使用已部署 Agent 需要先在服务器配置中启用 AI 集成（conductor-agents.md）：

conductor.integrations.ai.enabled=true

八、每种策略的可运行示例索引

官方文档声明：以下每种策略都在四个 SDK 中对照main分支验证过。策略对应的 SDK 示例位于各 SDK 仓库的examples/agents（Python 与 TypeScript）与agent-examples（Java、C#）中；在本仓库内可以直接运行/阅读对应 cookbook 与资产文件：

九、延伸阅读

• Multi-agent handoff recipe —— 一个带三个专家的可运行监督者示例
• Massively parallel agents —— 扇出到 100 个子智能体
• Agent Configuration —— 智能体上还能配置哪些内容
• Conductor Agents —— Agent 生命周期（create / plan / deploy / serve / run）与AGENT任务契约
• Agents & AI 总览 —— 智能体的回合循环与三条约稿路径（声明式 AI 工作流 / Conductor Agents / A2A 远程委托）

一句话总结：多智能体编排的复杂度全被收敛进一个strategy字段——模型拍板用handoff/router，图拍板用sequential/parallel/round_robin/random，智能体互相拍板用swarm，边跑边规划用plan_execute，业务规则拍板用manual；而无论选哪种，Conductor 都会把整个系统编译成一份可持久、可重试、可观测、可审计的工作流。

【免费下载链接】conductorConductor is an event driven agentic workflow engine providing durable and highly resilient execution engine for applications and AI Agents项目地址: https://gitcode.com/GitHub_Trending/co/conductor