Orquestrador e Trabalhadores

Intenção

Um modelo orquestrador decide, em tempo de execução, em quais subtarefas dividir o trabalho, entrega cada uma a um trabalhador e junta os resultados. Serve para tarefas cuja divisão não dá para prever antes de olhar a entrada.

Problema

Paula lidera a migração de um sistema legado de um framework antigo para um novo. São 400 arquivos. Alguns precisam só trocar imports; outros têm lógica de ciclo de vida que muda de semântica; alguns dependem de outros e precisam migrar juntos.

A primeira tentativa foi um agente único percorrendo os arquivos. Na metade, o contexto estava cheio e as decisões do começo (como tratar o novo sistema de eventos) tinham se perdido. A segunda tentativa foi um script que mandava cada arquivo para uma chamada isolada. Rápido, mas os arquivos que dependiam uns dos outros saíam incoerentes.

O que funciona tem as duas partes: alguém que olha o todo e decide como dividir, e trabalhadores que fazem cada parte com foco. A divisão muda a cada módulo, então não dá para fixá-la no código.

A pergunta que este padrão responde: como dividir dinamicamente uma tarefa grande entre vários modelos, mantendo coerência no resultado final?

Solução

Separe três papéis:

• Orquestrador: lê a tarefa e o contexto, produz uma lista de subtarefas com instruções autocontidas e dependências. Ele não faz o trabalho.
• Trabalhadores: cada um recebe uma subtarefa, faz num contexto próprio e devolve um resultado num formato combinado. Subtarefas sem dependência rodam em paralelo.
• Sintetizador: junta os resultados, resolve conflitos e entrega o produto final. Muitas vezes é o próprio orquestrador numa segunda chamada.

A diferença para o Seccionamento é quem decide a divisão: lá, o código; aqui, o modelo, olhando a entrada. Se você consegue escrever o for que divide o trabalho antes de ver a entrada, não precisa de orquestrador.

O post “Building effective agents” da Anthropic (dezembro de 2024) descreve este padrão com dois exemplos: produtos de código que mudam vários arquivos a cada pedido e buscas que juntam informação de várias fontes. Em junho de 2025, a Anthropic publicou como construiu o sistema de pesquisa do Claude com um agente líder que dispara subagentes em paralelo. O mesmo texto conta o custo: o sistema multiagente gastava cerca de 15 vezes mais tokens que uma conversa comum. O padrão compra tempo e cobertura com dinheiro.

Exemplo trabalhado

Paula pede: “Migre o módulo payments para o framework novo”. O orquestrador lê a árvore do módulo e devolve este plano:

{
  "conventions": [
    "Eventos usam o novo EventBus; nunca chame emit() direto",
    "Erros de domínio herdam de PaymentError"
  ],
  "subtasks": [
    { "id": "models",  "instruction": "Migre models/*.py para o novo ORM", "depends_on": [], "done_when": "testes de models passam" },
    { "id": "events",  "instruction": "Troque emit() por EventBus em events.py", "depends_on": [], "done_when": "events.py compila" },
    { "id": "service", "instruction": "Migre service.py usando os models e eventos novos", "depends_on": ["models", "events"], "done_when": "testes de service passam" },
    { "id": "api",     "instruction": "Atualize as rotas de api.py", "depends_on": ["service"], "done_when": "testes de api passam" }
  ]
}

O código transforma as dependências em ondas. A onda 1 roda models e events em paralelo. A onda 2 roda service, que recebe o resultado das duas. A onda 3 roda api. São quatro trabalhadores, três ondas e uma síntese no fim.

Repare em duas coisas. Cada instrução é autocontida: o trabalhador de events não precisa ler o resto do módulo para saber o que fazer. E o campo conventions leva para todos os trabalhadores as decisões que valem para o módulo inteiro.

A armadilha das decisões compartilhadas

Sem conventions, cada trabalhador decide sozinho como tratar o sistema de eventos. Um usa o EventBus, outro mantém emit() com um adaptador, e os dois resultados compilam. A síntese recebe duas soluções corretas e incompatíveis.

É o mesmo problema da primeira tentativa de Paula, só que em paralelo. A regra prática: toda decisão que mais de um trabalhador precisa tomar sai do orquestrador, no plano, antes de qualquer trabalhador começar.

Estrutura

flowchart TD
  IN(["1. Tarefa"]):::io --> O(["2. Orquestrador planeja"]):::llm
  O --> P[("3. Plano: subtarefas e dependências")]:::store
  P --> W1(["4. Trabalhador"]):::llm
  P --> W2(["4. Trabalhador"]):::llm
  P --> W3(["4. Trabalhador"]):::llm
  W1 --> V["5. Validar resultados"]:::tool
  W2 --> V
  W3 --> V
  V --> S(["6. Sintetizar"]):::llm
  S --> OUT(["7. Resultado"]):::io

1. A tarefa chega com o material necessário ou com acesso a ele.
2. O orquestrador analisa e decide a divisão.
3. O plano é um objeto estruturado: subtarefas, instrução de cada uma, dependências e critério de pronto.
4. Trabalhadores executam em paralelo quando não há dependência, cada um com contexto e orçamento próprios.
5. O código valida cada resultado (schema, testes, compilação) antes de seguir.
6. O sintetizador junta tudo e resolve inconsistências.
7. O resultado sai com o plano e os resultados parciais registrados no trace.

Pseudocódigo

schema Subtask { id: string, instruction: string, depends_on: list[string], done_when: string }
schema Plan { conventions: list[string], subtasks: list[Subtask] }
schema WorkResult { id: string, output: string, ok: bool, notes: string }

function orchestrate(task):
    # 2 e 3. o modelo decide a divisão
    plan = llm(ORCHESTRATOR_PROMPT, task) -> Plan
    validate_dag(plan)                                   # sem ciclos, sem ids soltos
    trace.span("plan", plan)

    results = {}
    for wave in topological_waves(plan.subtasks):
        # 4. cada onda roda em paralelo
        parallel {
            for st in wave:
                ctx = plan.conventions + [st.instruction] + [results[d].output for d in st.depends_on]
                results[st.id] = llm(WORKER_PROMPT, ctx) -> WorkResult
        }
        # 5. validar antes da próxima onda
        for st in wave:
            if not check(st.done_when, results[st.id]):
                results[st.id] = retry_once(st, results)
        budget.check()

    # 6. sintetizar
    return llm(SYNTH_PROMPT, task, plan, results)        # 7

Cenários reais

• Pesquisa profunda com citações. O orquestrador quebra a pergunta em subperguntas que dependem do que a primeira rodada encontrar. Trabalhadores pesquisam cada uma; o sintetizador escreve o relatório com as citações.
• Agente de código. Para mudanças em vários módulos, o orquestrador decide quais arquivos mudam e em que ordem. Cada trabalhador edita um módulo e roda os testes dele.
• Migração de código legado. O plano agrupa arquivos que dependem entre si, define a ordem das ondas e o critério de pronto (compila e passa nos testes do módulo). É o caso de Paula.

Aplicabilidade

• A tarefa é grande demais para um contexto e a divisão depende do conteúdo.
• As partes são em boa medida independentes, mas não totalmente.
• Há um critério de pronto por parte que o código consegue checar.
• Paralelismo reduz o tempo total de forma relevante.
• O valor da tarefa justifica gastar várias vezes os tokens de uma chamada só.

Quando não usar

• A divisão é sempre a mesma: por arquivo, por página, por seção. Use Seccionamento ou Map-Reduce de Documentos, com a divisão em código.
• A tarefa cabe num agente com contexto limpo. Um orquestrador acrescenta custo e pontos de falha. Comece por um Loop ReAct e só divida quando o contexto encher.
• As partes dependem tanto umas das outras que nenhuma pode rodar sozinha. Aí o paralelismo some, e você paga o plano e a síntese sem ganhar tempo.
• Os trabalhadores seriam papéis decorativos repassando texto. Um “analista”, um “revisor” e um “redator” que leem o mesmo contexto e reescrevem a mesma resposta não dividem trabalho nenhum. Veja Multi-Agente Decorativo.

Orquestrar ou planejar e executar

O Planejar e Executar também começa com um plano. A diferença está em quem executa. Lá, um único agente segue o plano passo a passo, no mesmo contexto. Aqui, cada subtarefa vai para um trabalhador com contexto próprio, e as independentes rodam ao mesmo tempo.

Use Planejar e Executar quando os passos são sequenciais e precisam do que veio antes. Use Orquestrador e Trabalhadores quando as partes cabem em contextos separados e o volume justifica o paralelismo.

Como implementar

1. Escreva o schema do plano: subtarefas, instrução autocontida, dependências, critério de pronto.
2. Escreva o prompt do orquestrador com limites: número máximo de subtarefas, tamanho de cada uma.
3. Valide o plano em código: sem ciclos, sem dependências inexistentes, dentro dos limites.
4. Escreva o prompt do trabalhador como função genérica que recebe instrução e contexto das dependências.
5. Execute em ondas, com paralelismo dentro de cada onda e orçamento total compartilhado.
6. Valide cada resultado com checagem em código antes de liberar a próxima onda.
7. Escreva o sintetizador e dê a ele o plano junto com os resultados.
8. Monte um eval com tarefas reais, medindo qualidade final, custo e tempo total.

Variações

• Orquestrador com replanejamento. Depois de cada onda, o orquestrador revisa o plano com os resultados. Mais adaptável, mais caro. Bom para pesquisa, em que a primeira rodada muda as perguntas da segunda.
• Trabalhadores com ferramentas. Cada trabalhador é um agente em loop, não uma chamada. Necessário em código e pesquisa; aí cada trabalhador é um subagente, como em Delegação a Subagentes.
• Trabalhadores em modelo menor. O orquestrador e o sintetizador usam o modelo grande; os trabalhadores, um modelo mais barato. Funciona quando as instruções do plano são específicas o bastante.
• Supervisor. O orquestrador acompanha os trabalhadores durante a execução e pode reatribuir tarefas. Mais controle, mais chamadas do orquestrador.

Conta ilustrativa de custo

Suponha que o plano custe 6 mil tokens, cada trabalhador 20 mil e a síntese 15 mil. Com 4 subtarefas, a tarefa gasta cerca de 101 mil tokens. Se o orquestrador decide dividir em 20 subtarefas, o total passa de 420 mil, e a qualidade não necessariamente sobe.

Por isso o limite de subtarefas fica no prompt e também no código. O modelo escolhe a divisão; o código decide quanto ela pode custar.

Custo, latência e falhas

Como avaliar

• Qualidade final, medida pelo critério da tarefa (testes passando, relatório com fontes válidas).
• Qualidade do plano: cobertura (todas as partes da tarefa aparecem) e sobreposição (partes repetidas).
• Custo e tempo total, comparados com a versão de agente único.

{ "task": "Migre o módulo payments para o framework novo", "check": "compila e passa nos testes de payments", "max_subtasks": 12 }

Avalie o plano separado do resultado. Para 20 tarefas reais, escreva à mão a lista de partes que um bom plano precisa cobrir. Depois compare com os planos gerados: parte faltando é falha de cobertura, duas subtarefas mexendo no mesmo arquivo é sobreposição. Um plano ruim estraga tudo o que vem depois, então é ele que você corrige primeiro.

Em produção

• Span plan com o plano completo, spans worker filhos com a subtarefa e o resultado.
• Limite de subtarefas e de ondas por tarefa, com alerta quando bate.
• Painel de retries por subtarefa: um tipo de subtarefa que sempre falha pede um prompt de trabalhador melhor.
• Fila de revisão para tarefas cuja síntese teve conflitos.

Prós e contras

Código

As abas fazem o plano com saída estruturada, executam os trabalhadores em paralelo por onda e sintetizam. Python e TypeScript disparam as ondas com o runtime da linguagem. LangGraph dispara cada trabalhador com Send, e o estado junta os retornos.

import asyncio

import anthropic
from pydantic import BaseModel

client = anthropic.AsyncAnthropic()


class Subtask(BaseModel):
    id: str
    instruction: str
    depends_on: list[str]


class Plan(BaseModel):
    subtasks: list[Subtask]


async def plan(task: str) -> Plan:  # 2 e 3
    reply = await client.messages.parse(
        model="claude-opus-5-5", max_tokens=8000,
        system="Divida a tarefa em no máximo 12 subtarefas autocontidas. Liste dependências por id.",
        messages=[{"role": "user", "content": task}], output_format=Plan,
    )
    return reply.parsed_output


async def work(st: Subtask, deps: dict[str, str]) -> str:  # 4
    context = "\n\n".join(f"Resultado de {d}:\n{deps[d]}" for d in st.depends_on)
    reply = await client.messages.create(
        model="claude-sonnet-5-5", max_tokens=16000,
        system="Você executa uma subtarefa de um plano maior. Faça só o que foi pedido.",
        messages=[{"role": "user", "content": f"{st.instruction}\n\n{context}"}],
    )
    return reply.content[-1].text


async def orchestrate(task: str) -> str:
    p = await plan(task)
    pending, results = {s.id: s for s in p.subtasks}, {}
    while pending:
        wave = [s for s in pending.values() if all(d in results for d in s.depends_on)]
        if not wave:
            raise ValueError("plano com ciclo ou dependência inexistente")
        outputs = await asyncio.gather(*(work(s, results) for s in wave))  # paralelo
        for s, out in zip(wave, outputs):
            results[s.id] = out  # 5. valide aqui (testes, schema) antes de seguir
            del pending[s.id]
    joined = "\n\n".join(f"[{k}]\n{v}" for k, v in results.items())
    final = await client.messages.create(  # 6
        model="claude-opus-5-5", max_tokens=16000,
        system="Junte os resultados num produto final coerente. Aponte conflitos.",
        messages=[{"role": "user", "content": f"Tarefa: {task}\n\n{joined}"}],
    )
    return final.content[-1].text

import Anthropic from "@anthropic-ai/sdk";
import { zodOutputFormat } from "@anthropic-ai/sdk/helpers/zod";
import { z } from "zod";

const client = new Anthropic();

const Plan = z.object({
  subtasks: z.array(z.object({ id: z.string(), instruction: z.string(), depends_on: z.array(z.string()) })),
});
type Subtask = z.infer<typeof Plan>["subtasks"][number];

const text = (m: Anthropic.Message) => m.content.find((b) => b.type === "text")?.text ?? "";

async function work(st: Subtask, results: Map<string, string>): Promise<string> {
  const context = st.depends_on.map((d) => `Resultado de ${d}:\n${results.get(d)}`).join("\n\n");
  const reply = await client.messages.create({
    model: "claude-sonnet-5-5", max_tokens: 16000,
    system: "Você executa uma subtarefa de um plano maior. Faça só o que foi pedido.",
    messages: [{ role: "user", content: `${st.instruction}\n\n${context}` }],
  });
  return text(reply);
}

export async function orchestrate(task: string): Promise<string> {
  const { parsed_output: plan } = await client.messages.parse({
    model: "claude-opus-5-5", max_tokens: 8000,
    system: "Divida a tarefa em no máximo 12 subtarefas autocontidas. Liste dependências por id.",
    messages: [{ role: "user", content: task }],
    output_config: { format: zodOutputFormat(Plan) },
  });
  if (!plan) throw new Error("orquestrador não devolveu plano");

  const results = new Map<string, string>();
  let pending = plan.subtasks;
  while (pending.length) {
    const wave = pending.filter((s) => s.depends_on.every((d) => results.has(d)));
    if (!wave.length) throw new Error("plano com ciclo ou dependência inexistente");
    const outs = await Promise.all(wave.map((s) => work(s, results))); // 4. paralelo
    wave.forEach((s, i) => results.set(s.id, outs[i])); // 5. valide aqui
    pending = pending.filter((s) => !results.has(s.id));
  }

  const joined = [...results].map(([k, v]) => `[${k}]\n${v}`).join("\n\n");
  const final = await client.messages.create({
    model: "claude-opus-5-5", max_tokens: 16000,
    system: "Junte os resultados num produto final coerente. Aponte conflitos.",
    messages: [{ role: "user", content: `Tarefa: ${task}\n\n${joined}` }],
  });
  return text(final);
}

import operator
from typing import Annotated

from langchain.chat_models import init_chat_model
from langgraph.graph import END, START, StateGraph
from langgraph.types import Send
from pydantic import BaseModel
from typing_extensions import TypedDict

class Task(BaseModel):
    id: str
    instruction: str

class Plan(BaseModel):
    tasks: list[Task]

class State(TypedDict):
    goal: str
    plan: list[dict]
    results: Annotated[list[str], operator.add]
    report: str

planner = init_chat_model("anthropic:claude-opus-5-5", temperature=0).with_structured_output(Plan)
worker_model = init_chat_model("anthropic:claude-haiku-4-5", temperature=0)
synth = init_chat_model("anthropic:claude-opus-5-5", temperature=0)

def plan(state: State) -> dict:
    # 2 e 3. o orquestrador só planeja
    drafted = planner.invoke([
        ("system", "Divida a tarefa em subtarefas independentes. Não execute."),
        ("human", state["goal"]),
    ])
    return {"plan": [t.model_dump() for t in drafted.tasks]}

def fan(state: State):
    # 4. um Send por trabalhador; eles rodam em paralelo
    return [Send("work", {"goal": task["instruction"]}) for task in state["plan"]]

def work(state: State) -> dict:
    reply = worker_model.invoke([("human", state["goal"])])
    return {"results": [reply.text]}  # operator.add junta as listas

def synthesize(state: State) -> dict:
    # 5 e 6. validar e juntar fica no seu código antes desta chamada, se o critério for teste
    report = synth.invoke([("human", "\n".join(state["results"]))])
    return {"report": report.text}

graph = StateGraph(State)
graph.add_node("plan", plan)
graph.add_node("work", work)
graph.add_node("synthesize", synthesize)
graph.add_edge(START, "plan")
graph.add_conditional_edges("plan", fan, ["work"])
graph.add_edge("work", "synthesize")
graph.add_edge("synthesize", END)
app = graph.compile()

Suporte por framework

Teste seu entendimento