How to Build an Orchestrator‑Workers AI Agent Workflow with Pydantic AI

Understanding the Orchestrator‑Workers Pattern

The Orchestrator‑Workers mode consists of a large language model (LLM) that first analyses a task, decomposes it into several sub‑tasks, and then hands each sub‑task to multiple “enhanced” LLM workers. The workers complete their assigned sub‑tasks in parallel and the orchestrator merges the results into a final output.

Differences from Routing and Parallel Modes

Routing mode: a single LLM call routes the whole task to one downstream branch; the Orchestrator‑Workers mode splits the task and distributes it to many branches.

Parallel mode: parallel sub‑tasks are predefined and predictable, so no LLM is needed for decomposition; in the Orchestrator‑Workers mode the parallel sub‑tasks are dynamic and require the orchestrator LLM to create them.

Relation to Supervisor Agents in Multi‑Agent Systems

A Supervisor agent has autonomous planning ability; it does not merely split tasks but also plans the sequence and path of multiple sub‑tasks.

Sub‑tasks generated by a Supervisor are usually executed sequentially rather than in parallel.

The planning performed by a Supervisor is dynamic, adapting to the environment after each sub‑task completes.

Outputs of Supervisor sub‑tasks are fed back to the Supervisor for further decision‑making.

Implementation with Pydantic AI

1. Define output models

# Define task model
class Task(BaseModel):
    type: str = Field(..., description='任务类型。即用来区分子任务的某个具体角度')
    description: str = Field(..., description='执行此任务的清晰描述与要求。')

# Define orchestrator response model
class OrchestratorResponse(BaseModel):
    analysis: str = Field(..., description='解释你对任务的理解，判断应该从哪个角度分解任务, 并说明不同子任务如何服务于不同的角度。')
    tasks: List[Task] = Field(..., description="任务列表")

2. Configure prompts and models

# Global configuration
config = {
    'orchestrator': {
        'system_prompt': '分析输入任务并将其分解为3-5种不同的并行子任务。',
        'model': OpenAIModel(model_name='gpt-4o-mini')
    },
    'worker': {
        'system_prompt': '根据任务描述与要求完成并输出。',
        'model': OpenAIModel(model_name='gpt-4o-mini')
    }
}

3. Implement the orchestrator method

async def orchestrate(task: str) -> OrchestratorResponse:
    """通过分解任务来处理任务。"""
    # Create orchestrator agent
    orchestrator_agent = Agent(
        config['orchestrator']['model'],
        system_prompt=config['orchestrator']['system_prompt'],
        result_type=OrchestratorResponse,
    )
    # Run orchestrator agent
    orchestrator_response = await orchestrator_agent.run(task)

    analysis = orchestrator_response.data.analysis
    tasks = orchestrator_response.data.tasks

    print(colored('分析:', 'cyan'))
    print(colored(analysis, 'yellow'))
    print(colored('任务列表:', 'cyan'))
    for idx, task in enumerate(tasks, 1):
        print(colored(f'子任务 {idx}:', 'green'))
        pprint.pprint(task.model_dump())
    print('
')
    return tasks

4. Implement the worker method

async def execute_tasks(task: str, tasks: List[Task]):
    """并行运行子任务并收集结果。"""
    # Create worker agent
    worker_agent = Agent(
        config['worker']['model'],
        system_prompt=config['worker']['system_prompt'],
    )

    # Parallel execution of subtasks
    worker_responses = await asyncio.gather(*[
        worker_agent.run(json.dumps({
            'original_task': task,
            'task_info': task_info.model_dump()
        }))
        for task_info in tasks
    ])

    # Print each task's result
    for task, response in zip(tasks, worker_responses):
        print(f"工作结果 ({task.type}): {response.data}
")

5. Test the workflow

if __name__ == "__main__":
    task = '用多种语言风格创建一句2025的新春祝福语。'
    sub_tasks = asyncio.run(orchestrate(task))
    asyncio.run(execute_tasks(task, sub_tasks))

Test Run and Observations

Running the script with the example task splits it into four distinct sub‑tasks, as shown in the first screenshot.

The workers then invoke the LLM to complete each sub‑task, and their outputs are displayed in the second screenshot.

The implementation is straightforward for simple content‑generation tasks, but real‑world scenarios may require more sophisticated role definitions for workers, dynamic prompt selection, and tighter integration between the orchestrator’s analysis and the workers’ responsibilities.

Future articles will explore the remaining foundational pattern – the Evaluator‑Optimizer (reflection) mode.