编排器-工作器模式：动态任务分解与并行执行指南

编排器-工作器模式通过中央LLM动态分析任务，生成并委派子任务给专门的工作器LLM，实现灵活的多视角内容生成。本教程详细讲解其原理、架构、实现步骤，并提供营销文案生成示例。涵盖环境配置、代码解析、常见错误排查及生产优化方案，帮助开发者掌握这一高级AI工程模式，适用于需要动态分解复杂任务的场景。

技术背景

在AI应用中，许多任务需要从多个角度生成内容或分析，但预先定义所有子任务往往不切实际。传统方法如手动多次提示或硬编码并行化，要么效率低下，要么缺乏适应性。编排器-工作器模式应运而生，它利用一个中央LLM（编排器）在运行时分析任务，动态决定哪些子任务最有价值，并委派给专门的工作器LLM执行。这种模式特别适用于子任务取决于具体输入的场景，如营销文案生成、多视角分析等。其主要限制是增加延迟和成本（N+1次LLM调用），且对提示工程要求较高。

核心原理

核心原理是两阶段处理：分析与规划阶段，编排器LLM接收任务和上下文，分析并生成XML格式的结构化子任务描述；执行阶段，每个工作器LLM接收原始任务、子任务类型和描述，生成结果。关键设计决策包括：使用XML作为结构化输出格式，因其对LLM友好且易于解析；工作器同时接收原始任务和特定指令，以保持上下文一致性；错误处理验证工作器输出，确保可靠性。数据流为：任务输入→编排器分析→子任务列表→并行/顺序执行→结果汇总。

架构分析

系统由两个核心组件构成：编排器（Orchestrator）和工作器（Worker）。编排器负责任务分解，通过提示模板生成分析结果和任务列表；工作器负责执行具体子任务。FlexibleOrchestrator类封装了协调逻辑，包含parse_tasks()解析XML，process()主流程。架构设计强调灵活性：提示模板接受运行时变量，XML通信确保结构化，错误处理增强鲁棒性。生产环境可优化为异步并行执行工作器，或使用不同模型（如Claude Opus作为编排器，Haiku作为工作器）平衡成本与质量。

完整实战教程

编排器-工作器工作流

简介

你是否曾需要对同一任务获得多个视角，但无法预先知道哪些视角最有价值？编排器-工作器模式通过让一个中央LLM分析每个独特任务，并动态决定将哪些最佳子任务委派给专门的工作器LLM来解决这个问题。

传统方法要么需要手动多次提示，要么使用硬编码的并行化，无论上下文如何都生成相同的变体。

通过这种方法，编排器LLM分析任务，确定哪些变体对此特定案例最有价值，然后委派给工作器LLM生成每个变体。

你将构建什么

一个接收产品描述请求的系统，它能够：

  1. 分析哪些类型的营销文案有价值
  2. 动态生成针对工作器的专门任务描述
  3. 生成针对不同受众优化的多个内容变体
  4. 返回所有工作器的协调结果

先决条件

何时使用此工作流

此工作流适用于无法预先预测所需子任务的复杂任务。与简单并行化的关键区别在于其灵活性——子任务不是预定义的，而是由编排器根据具体输入确定。

在以下情况下使用此模式：

在以下情况下不要使用此模式：

工作原理

编排器-工作器模式分两个阶段运行：

  1. 分析与规划阶段：编排器LLM接收任务和上下文，分析哪些方法有价值，并生成XML格式的结构化子任务描述。
  1. 执行阶段：每个工作器LLM接收：

编排器在*运行时*决定创建哪些子任务，这使得它比预定义的并行工作流更具适应性。

设置

安装

pip install anthropic

辅助函数

此示例使用 util.py 中的辅助函数进行 LLM 调用和解析 XML 响应：

这些工具处理 API 身份验证（从环境变量读取 ANTHROPIC_API_KEY），并为编排器-工作器模式提供简单接口。您可以在 util.py 中查看完整实现。

实现

FlexibleOrchestrator 类协调两阶段工作流：

关键设计决策：

实现包括：

from util import extract_xml, llm_call
# Model configuration
MODEL = "claude-sonnet-4-6"  # Fast, capable model for both orchestrator and workers
def parse_tasks(tasks_xml: str) -> list[dict]:
    """Parse XML tasks into a list of task dictionaries."""
    tasks = []
    current_task = {}
    for line in tasks_xml.split("\n"):
        line = line.strip()
        if not line:
            continue
        if line.startswith("<task>"):
            current_task = {}
        elif line.startswith("<type>"):
            current_task["type"] = line[6:-7].strip()
        elif line.startswith("<description>"):
            current_task["description"] = line[12:-13].strip()
        elif line.startswith("</task>"):
            if "description" in current_task:
                if "type" not in current_task:
                    current_task["type"] = "default"
                tasks.append(current_task)
    return tasks
class FlexibleOrchestrator:
    """Break down tasks and run them in parallel using worker LLMs."""
    def __init__(
        self,
        orchestrator_prompt: str,
        worker_prompt: str,
        model: str = MODEL,
    ):
        """Initialize with prompt templates and model selection."""
        self.orchestrator_prompt = orchestrator_prompt
        self.worker_prompt = worker_prompt
        self.model = model
    def _format_prompt(self, template: str, **kwargs) -> str:
        """Format a prompt template with variables."""
        try:
            return template.format(**kwargs)
        except KeyError as e:
            raise ValueError(f"Missing required prompt variable: {e}") from e
    def process(self, task: str, context: dict | None = None) -> dict:
        """Process task by breaking it down and running subtasks in parallel."""
        context = context or {}
        # Step 1: Get orchestrator response
        orchestrator_input = self._format_prompt(self.orchestrator_prompt, task=task, **context)
        orchestrator_response = llm_call(orchestrator_input, model=self.model)
        # Parse orchestrator response
        analysis = extract_xml(orchestrator_response, "analysis")
        tasks_xml = extract_xml(orchestrator_response, "tasks")
        tasks = parse_tasks(tasks_xml)
        print("\n" + "=" * 80)
        print("ORCHESTRATOR ANALYSIS")
        print("=" * 80)
        print(f"\n{analysis}\n")
        print("\n" + "=" * 80)
        print(f"IDENTIFIED {len(tasks)} APPROACHES")
        print("=" * 80)
        for i, task_info in enumerate(tasks, 1):
            print(f"\n{i}. {task_info['type'].upper()}")
            print(f"   {task_info['description']}")
        print("\n" + "=" * 80)
        print("GENERATING CONTENT")
        print("=" * 80 + "\n")
        # Step 2: Process each task
        worker_results = []
        for i, task_info in enumerate(tasks, 1):
            print(f"[{i}/{len(tasks)}] Processing: {task_info['type']}...")
            worker_input = self._format_prompt(
                self.worker_prompt,
                original_task=task,
                task_type=task_info["type"],
                task_description=task_info["description"],
                **context,
            )
            worker_response = llm_call(worker_input, model=self.model)
            worker_content = extract_xml(worker_response, "response")
            # Validate worker response - handle empty outputs
            if not worker_content or not worker_content.strip():
                print(f"⚠️  Warning: Worker '{task_info['type']}' returned no content")
                worker_content = f"[Error: Worker '{task_info['type']}' failed to generate content]"
            worker_results.append(
                {
                    "type": task_info["type"],
                    "description": task_info["description"],
                    "result": worker_content,
                }
            )
        # Display results
        print("\n" + "=" * 80)
        print("RESULTS")
        print("=" * 80)
        for i, result in enumerate(worker_results, 1):
            print(f"\n{'-' * 80}")
            print(f"Approach {i}: {result['type'].upper()}")
            print(f"{'-' * 80}")
            print(f"\n{result['result']}\n")
        return {
            "analysis": analysis,
            "worker_results": worker_results,
        }

示例用例：营销变体生成

现在让我们通过一个实际示例来了解编排器-工作器模式的应用：为产品生成多种风格的营销文案。

为什么这个示例能很好地展示该模式：

提示设计说明：

ORCHESTRATOR_PROMPT = """
Analyze this task and break it down into 2-3 distinct approaches:

Task: {task}

Return your response in this format:

<analysis>
Explain your understanding of the task and which variations would be valuable.
Focus on how each approach serves different aspects of the task.
</analysis>

<tasks>
    <task>
    <type>formal</type>
    <description>Write a precise, technical version that emphasizes specifications</description>
    </task>
    <task>
    <type>conversational</type>
    <description>Write an engaging, friendly version that connects with readers</description>
    </task>
</tasks>
"""
WORKER_PROMPT = """
Generate content based on:
Task: {original_task}
Style: {task_type}
Guidelines: {task_description}

Return your response in this format:

<response>
Your content here, maintaining the specified style and fully addressing requirements.
</response>
"""
orchestrator = FlexibleOrchestrator(
    orchestrator_prompt=ORCHESTRATOR_PROMPT,
    worker_prompt=WORKER_PROMPT,
)
results = orchestrator.process(
    task="Write a product description for a new eco-friendly water bottle",
    context={
        "target_audience": "environmentally conscious millennials",
        "key_features": ["plastic-free", "insulated", "lifetime warranty"],
    },
)

总结

你现在已经实现了一个编排者-工作者模式，该模式能够根据具体输入动态调整任务分解。该模式生成了多个营销文案变体——每个变体针对不同的受众和场景——而无需你预先定义这些变体应该是什么。

关键要点

模式优势：

该模式擅长的场景：

局限性与注意事项

成本与延迟：

何时不应使用此模式：

需要考虑的失败模式：

后续步骤

增强此实现：

  1. 使用asyncio添加并行工作者执行以获得更好的性能
  2. 为失败的工作者实现重试逻辑
  3. 添加一个综合阶段，让LLM合并工作者输出
  4. 尝试不同的编排者策略（例如，要求更多/更少的子任务）

适应你的用例：

常见问题

常见错误包括：1) 编排器未正确生成XML，导致解析失败——检查提示词是否明确指定格式，可考虑使用JSON替代；2) 工作器返回空响应——已实现验证并替换错误信息，可增加重试逻辑；3) 提示词变量缺失——_format_prompt会抛出ValueError，确保所有变量已提供；4) 环境变量未设置——确认ANTHROPIC_API_KEY已导出。排查时，先检查API调用日志，再验证XML解析结果，最后审查提示词质量。

企业应用扩展

生产环境优化：1) 使用asyncio或线程池并行执行工作器，减少总延迟；2) 为失败的工作器实现重试机制；3) 添加综合阶段，让LLM合并多个工作器输出；4) 调整编排器策略，如动态控制子任务数量；5) 针对特定领域定制提示词，提高分解质量；6) 考虑使用不同模型组合优化成本。后续可探索将模式应用于代码生成、研究分析等场景，并集成监控和日志系统。

总结

编排器-工作器模式通过中央LLM动态分析任务，生成并委派子任务给专门的工作器LLM，实现灵活的多视角内容生成。本教程详细讲解其原理、架构、实现步骤，并提供营销文案生成示例。涵盖环境配置、代码解析、常见错误排查及生产优化方案，帮助开发者掌握这一高级AI工程模式，适用于需要动态分解复杂任务的场景。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员，解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部AI 智能体教程