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( namesupport_supervisor, modelopenai/gpt-4o-mini, instructionsRoute each request to the right specialist., agents[billing, technical, sales], strategyStrategy.HANDOFF, )这段代码来自官方文档中的示例support_supervisor是父智能体billing、technical、sales是子智能体strategyStrategy.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还是你代码中的业务规则。官方文档给出的九种策略对照表如下这是选型的第一依据StrategyWho decidesRunsReach for it whenhandoffModelOne sub-agent, conversationallyA specialist should take over the conversationrouterModelOne sub-agent, no conversationYou just need classification and dispatchsequentialGraphAll, in orderEach step builds on the previous outputparallelGraphAll, at onceIndependent opinions you want to compareswarmSub-agentsUntil one finishesAgents should pass control between themselvesround_robinGraphNext in rotationSpreading load or alternating reviewersrandomGraphOne at randomA/B comparison between agent versionsplan_executeModel, then graphA planned sequence, replanned as it goesThe steps arent knowable up frontmanualYou, in codeWhatever you selectRouting is a business rule, not a judgement call两条实战要点需要特别记住router比handoff更便宜router只做分类与分发classify and dispatch不把对话交给子智能体。当分类之后无需继续对话时应优先使用router以节省一轮对话的模型开销。plan_execute是唯一会重新规划replan的策略其余策略一旦做出分发决定就提交commit了只有plan_execute在执行过程中会根据返回结果修订后续计划。三、四种策略形状模型挑选 / 图运行全部 / 智能体互相交接 / 规划-执行-再规划九种策略在形态上可归为四类官方文档给出了每类的代码骨架3.1 模型挑选一个handoff与router子智能体以**可调用工具callable tools**的形式暴露给父智能体的模型模型按名字挑选一个。support Agent( namesupport, modelMODEL, instructionsRoute to billing, technical, or sales., agents[billing, technical, sales], strategyStrategy.HANDOFF, # or Strategy.ROUTER )3.2 图运行全部sequential与parallel模型不参与顺序决策——顺序由编排图决定。pipeline Agent( namereview_pipeline, modelMODEL, agents[researcher, writer, editor], strategyStrategy.SEQUENTIAL, # or Strategy.PARALLEL )sequential按顺序运行全部子智能体每个子智能体都能看到上一个的输出parallel同时运行全部并收集每个答案。3.3 智能体互相交接swarm控制权在子智能体之间传递直到其中一个产出最终答案为止。swarm Agent( nametriage_swarm, modelMODEL, agents[intake, diagnosis, resolution], strategyStrategy.SWARM, )3.4 规划、执行、再规划plan_execute模型先生成一份子智能体调用计划执行它并在结果返回后修订计划。planner Agent( nameincident_planner, modelMODEL, agents[log_reader, metrics_reader, remediation_drafter], strategyStrategy.PLAN_EXECUTE, )适用于步骤无法预先确定not knowable up front的场景例如事件排查类任务。四、Conductor 在策略之上添加了什么官方文档明确指出多智能体框架本身只解决谁决定、怎么跑而 Conductor 把策略编译成工作流之后额外获得四项关键能力每次委托都是一次独立的执行Each delegation is its own execution某个专家specialist可以单独重试而不需要重跑路由决策。这意味着路由是幂等的、可恢复的一次子智能体失败不会连坐父智能体的分类逻辑。选择被记录下来The choice is recorded哪个子智能体跑了、为什么跑都存在于执行记录execution中而不只是一行日志。这直接支持了事后审计与 Agent 评估Evals。子智能体保留自己的工具与护栏Sub-agents keep their own tools and guardrails例如 billing 智能体无法触达 fulfilment 工具——每个子智能体的tools、instructions、护栏是隔离的作用域。这正对应 cookbook agent-handoff.md 生产建议中分别限定每个专家的工具范围。并行是真并行Parallel means actually parallelparallel与扇出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.pyfrom 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( namebilling, modelMODEL, instructionsYou handle billing questions: balances, payments, invoices., tools[check_balance], ) technical Agent( nametechnical, modelMODEL, instructionsYou handle technical questions: order status, shipping, returns., tools[lookup_order], ) sales Agent( namesales, modelMODEL, instructionsYou handle sales questions: pricing, products, promotions., tools[get_pricing], ) support Agent( namesupport_supervisor, modelMODEL, instructionsRoute each request to the right specialist: billing, technical, or sales., agents[billing, technical, sales], strategyStrategy.HANDOFF, ) if __name__ __main__: with AgentRuntime() as runtime: result runtime.run(support, Whats 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 handoffsrouter在不需要对话时比handoff更便宜——按问题形态选策略而不是为了求新handoff 决策本质是模型输出要记录哪个专家跑了、为什么跑对应选择被记录能力分别限制每个专家的预算避免单个专家耗尽整体预算。六、实战二scatter_gather百路并行可运行针对parallel的极端形态官方文档在Next steps中指向 agent-scatter-gather.md一个协调者coordinator把一次请求分解成 100 个独立子任务全部并行运行每个子任务是一个独立的子工作流拥有自己的重试最后汇总结果。源码在 agent_scatter_gather.pyfrom 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( namecountry_researcher, modelMODEL, 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_turns5, ) 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( namecountry_coordinator, workerresearcher, modelSYNTHESIS_MODEL, instructions( fCreate EXACTLY {len(COUNTRIES)} country_researcher calls, one per country fbelow, passing just the country name. Issue ALL calls in a SINGLE response.\n\n fCountries:\n{country_list}\n\n fWhen all {len(COUNTRIES)} results are back, compile a short report grouped fby region. ), retry_count3, retry_delay_seconds5, timeout_seconds900, ) if __name__ __main__: with AgentRuntime() as runtime: result runtime.run( coordinator, fProfile 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_fastFalse该 API 的默认行为意味着一个 worker 失败不会拖垮整个批次使用更大的模型做汇总SYNTHESIS_MODEL因为它要一次性读取全部 100 份结果max_turns限制 worker 的轮数防止单个 worker 死循环拖住 JOINretry_count、retry_delay_seconds、timeout_seconds分别控制重试次数、重试延迟与总超时。生产注意cookbook 原文限流先于引擎成为瓶颈——100 个并发调用会先触达 provider 配额汇总阶段的上下文窗口要控制worker 输出保持简短97/100 的部分成功对你的调用方意味着什么要在上线前决策成本随数量线性增长建议先用 5 个 worker 验证形态再跑 100 个。七、在 AGENT 任务中编排已部署的多智能体当多智能体系统被deploy到服务器后工作流通过AGENT任务按名字调用详见 conductor-agents.mdagentType: 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.mdconductor.integrations.ai.enabledtrue八、每种策略的可运行示例索引官方文档声明以下每种策略都在四个 SDK 中对照main分支验证过。策略对应的 SDK 示例位于各 SDK 仓库的examples/agentsPython 与 TypeScript与agent-examplesJava、C#中在本仓库内可以直接运行/阅读对应 cookbook 与资产文件Strategy本仓库可直接查看的资产handoffagent_handoff.py对应 cookbookagent-handoff.mdrouterSDK 示例08_router_agent.py/Example08RouterAgent.java/08-router-agent.ts/08_RouterAgent形状同handoff仅策略值不同sequentialSDK 示例06_sequential_pipeline.py/Example06SequentialPipeline.java/06-sequential-pipeline.ts/06_SequentialPipelineparallelSDK 示例07_parallel_agents.py/Example07ParallelAgents.java/07-parallel-agents.ts/07_ParallelAgentsswarmSDK 示例17_swarm_orchestration.py/Example17SwarmOrchestration.java/17-swarm-orchestration.ts/17_SwarmOrchestrationrandomSDK 示例16_random_strategy.py/Example16RandomStrategy.java/16-random-strategy.ts/16_RandomStrategymanualSDK 示例18_manual_selection.py/Example18ManualSelection.java/18-manual-selection.ts/18_ManualSelectionplan_executeSDK 示例108_plan_execute_refs.py/Example108PlanExecuteRefs.java/108-plan-execute-refs.ts/108_PlanExecuteRefsround_robin尚无专属示例与random形状相同仅将策略值替换为Strategy.ROUND_ROBIN九、延伸阅读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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考