Router Chain Pattern com LangChain

Um prompt para tudo não escala

Um atendimento automático começa com um prompt só: "você é um agente de suporte, responda o cliente". Com o tempo, ele cresce. Entram regras de estorno, passos de diagnóstico técnico, o roteiro de retenção para quem quer cancelar. O prompt vira um documento enorme em que cada regra atrapalha as outras.

O Router Chain Pattern separa esse prompt gigante em vários prompts pequenos, um por assunto, e coloca na frente uma etapa que decide qual deles usar. É o mesmo raciocínio de uma central telefônica: primeiro você diz o assunto, depois fala com o setor certo.

Como o padrão funciona

1. A classificação transforma a mensagem num objeto com categoria e prioridade.
2. O roteador lê esse objeto e escolhe qual chain executar.
3. Só a rota escolhida roda; as outras não custam nada.

Repare na última rota: ela não chama o modelo. Um roteador não precisa levar sempre a um LLM; uma regra de negócio (ticket urgente vai para uma pessoa) é uma rota tão válida quanto as outras.

Uma rota por assunto

Cada rota é uma chain comum (prompt | llm | parser) com um prompt especializado. O detalhe importante é que todas recebem as mesmas variáveis, {message} e {summary}. Assim, o roteador pode mandar qualquer ticket para qualquer rota sem adaptar nada.

from langchain_core.prompts import ChatPromptTemplate


def _response_prompt(instructions: str) -> ChatPromptTemplate:
    """Build a response prompt that shares the same input variables."""
    system = (
        "system",
        instructions + " Answer in Portuguese, in at most 4 sentences. "
        "Ticket summary: {summary}",
    )
    user = ("user", "{message}")
    return ChatPromptTemplate.from_messages([system, user])


def create_billing_prompt() -> ChatPromptTemplate:
    """Route for invoices, charges and refunds."""
    return _response_prompt(
        "You are a billing specialist. Explain the charge, the refund "
        "process and the expected deadline."
    )


def create_technical_prompt() -> ChatPromptTemplate:
    """Route for bugs, errors and integrations."""
    return _response_prompt(
        "You are a technical support engineer. Give clear troubleshooting "
        "steps and ask for the information needed to investigate."
    )

# create_cancellation_prompt() e create_question_prompt() seguem o mesmo modelo

As rotas ficam num dicionário, com a categoria como chave. A primeira etapa de cada uma (_to_prompt_input) traduz o estado do roteador para as variáveis do prompt:

def _to_prompt_input(state: dict) -> dict:
    """Map the router state to the variables every route prompt expects."""
    return {"message": state["message"], "summary": state["ticket"].summary}


def create_routes(llm: BaseChatModel) -> dict[str, Runnable]:
    """One specialized chain per ticket category."""
    parser = StrOutputParser()
    prompts = {
        "billing": create_billing_prompt(),
        "technical": create_technical_prompt(),
        "cancellation": create_cancellation_prompt(),
        "question": create_question_prompt(),
    }
    return {
        category: RunnableLambda(_to_prompt_input) | prompt | llm | parser
        for category, prompt in prompts.items()
    }

Roteando com uma função

O jeito mais simples de rotear no LangChain é uma função Python comum dentro de um RunnableLambda. O truque: se a função devolve uma chain, o LangChain executa essa chain com a mesma entrada. A função só precisa escolher.

def create_router_chain(
    llm: BaseChatModel | None = None,
    classifier_llm: BaseChatModel | None = None,
):
    if llm is None:
        llm = init_chat_model(model="gemini-2.5-flash", model_provider="google_genai")

    classify = create_structured_chain(llm=classifier_llm or llm)  # Parte 3
    routes = create_routes(llm)
    escalate = RunnableLambda(lambda state: ESCALATION_MESSAGE)

    def route(state: dict) -> Runnable:
        ticket = state["ticket"]
        if ticket.priority == "high" and ticket.needs_human:
            return escalate
        return routes.get(ticket.category, routes["question"])

    return (
        RunnablePassthrough.assign(ticket=classify)
        | RunnablePassthrough.assign(answer=RunnableLambda(route))
    )

Lendo de cima para baixo:

• RunnablePassthrough.assign(ticket=classify) mantém a entrada e acrescenta o ticket classificado (o mesmo recurso usado na Parte 2).
• route olha o ticket e devolve a chain certa. A regra de escalonamento vem primeiro, porque tem prioridade sobre a categoria.
• routes.get(..., routes["question"]) garante uma rota padrão: nenhum ticket fica sem resposta.
• O segundo assign guarda a resposta em answer.

O resultado tem tudo o que aconteceu no caminho, o que facilita log e depuração:

result = chain.invoke({"message": "A integração com o webhook parou desde ontem"})

result["ticket"].category   # "technical"
result["ticket"].priority   # "medium"
result["answer"]            # resposta do engenheiro de suporte

Alternativa: RunnableBranch

O LangChain também tem o RunnableBranch, que recebe pares de (condição, chain) e um ramo padrão no final. Ele executa a primeira chain cuja condição for verdadeira:

from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (
        lambda s: s["ticket"].priority == "high" and s["ticket"].needs_human,
        RunnableLambda(lambda s: ESCALATION_MESSAGE),
    ),
    (lambda s: s["ticket"].category == "billing", routes["billing"]),
    (lambda s: s["ticket"].category == "technical", routes["technical"]),
    (lambda s: s["ticket"].category == "cancellation", routes["cancellation"]),
    routes["question"],  # ramo padrão
)

chain = (
    RunnablePassthrough.assign(ticket=classify)
    | RunnablePassthrough.assign(answer=branch)
)

As duas versões estão no repositório e se comportam igual nos testes. A diferença é de estilo:

Para a maioria dos casos, a função é mais simples de ler e de manter. O RunnableBranch funciona bem quando são poucas condições e você quer vê-las todas juntas.

Rodando o exemplo

O ponto de entrada manda três mensagens de assuntos diferentes de uma vez, com batch. Cada uma passa pela classificação e cai numa rota diferente:

MESSAGES = [
    "A integração com o webhook parou de enviar eventos desde ontem às 18h.",
    "Quero cancelar minha assinatura no fim do mês.",
    "Vocês emitem nota fiscal para pessoa física?",
]


def main():
    chain = create_router_chain()

    results = chain.batch([{"message": m} for m in MESSAGES])

    for result in results:
        ticket = result["ticket"]
        print(f"=== [{ticket.category} / {ticket.priority}] {result['message']}")
        print(result["answer"], end="\n\n")

python src/main_router.py

Boas práticas e cuidados

• Sempre tenha uma rota padrão. A classificação vai errar ou encontrar casos novos. Um dict.get com valor padrão, ou o último ramo do RunnableBranch, evita erro em produção.
• Feche as categorias com Literal. Como a classificação usa Structured Output, o modelo só pode devolver categorias que existem como rota.
• Dê a mesma interface a todas as rotas. Variáveis iguais em todos os prompts deixam qualquer rota pronta para qualquer ticket.
• Coloque regras de negócio antes da categoria. Urgência, cliente VIP ou bloqueio legal devem ser avaliados primeiro, e podem levar a rotas sem LLM.
• Use um modelo mais barato para classificar. O parâmetro classifier_llm existe para isso: classificar é mais simples que responder.
• Registre a rota escolhida. Guardar ticket.category junto com a resposta mostra onde a classificação erra e quais rotas precisam de ajuste.
• Lembre do custo. O roteador soma uma chamada (a classificação) antes da resposta. Ele compensa quando os prompts especializados respondem melhor do que um prompt genérico.
• Não crie rotas demais. Categorias muito parecidas confundem o classificador. Comece com poucas e divida só quando a resposta de uma rota ficar ruim para um tipo de pedido.

Resumo e referências

• O Router Chain classifica a entrada e executa só a chain especializada naquele assunto.
• A classificação com Structured Output garante categorias válidas.
• Uma função dentro de RunnableLambda pode devolver a chain a executar; o RunnableBranch faz o mesmo com pares de condição e chain.
• Rotas podem ser regras de negócio sem LLM, e sempre deve haver uma rota padrão.

Na Parte 5 (Parallel Chain), em vez de escolher uma rota, vamos executar várias etapas ao mesmo tempo e juntar os resultados.

Referências

• Repositório pattern_chain (código completo)
• Referência do LangChain: runnables (RunnableLambda, RunnableBranch)
• Documentação oficial do LangChain