Structured Output Pattern com LangChain

O problema do texto livre

Nas partes anteriores, toda chain terminava com StrOutputParser: o resultado era um texto. Isso serve para mostrar a resposta a uma pessoa, mas não para o código tomar decisões com ela.

Imagine um sistema de suporte que precisa saber a categoria e a prioridade de cada mensagem para decidir para onde ela vai. A primeira tentativa costuma ser pedir JSON no prompt e torcer:

prompt = "Classifique a mensagem e responda em JSON com category e priority..."
texto = chain.invoke({"message": mensagem})

dados = json.loads(texto)          # quebra se vier "```json" ou um comentário
prioridade = dados["priority"]     # quebra se o modelo escrever "Priority"
if prioridade == "alta": ...       # e se vier "high"? ou "Alta"?

Cada linha é um ponto de falha. O Structured Output Pattern resolve isso trocando o texto por um contrato: você define o formato em código e o LangChain devolve um objeto que já segue esse formato, ou avisa claramente quando não segue.

Como o padrão funciona

1. Um modelo Pydantic define os campos, os tipos e os valores permitidos.
2. O LangChain envia esse schema ao modelo junto com o prompt. A descrição de cada campo funciona como instrução.
3. with_structured_output devolve uma instância do modelo Pydantic, já validada. Não existe parser de texto no caminho.

Definindo o schema

O exemplo deste tutorial faz a triagem de mensagens de suporte. O schema descreve exatamente o que queremos de volta:

from typing import Literal

from pydantic import BaseModel, Field


class TicketAnalysis(BaseModel):
    """Triage of a customer support message."""

    category: Literal["billing", "technical", "cancellation", "question"] = Field(
        description="Main subject of the message"
    )
    priority: Literal["low", "medium", "high"] = Field(
        description="high when the customer is blocked or losing money"
    )
    summary: str = Field(
        description="One-sentence summary of the request, in Portuguese",
        max_length=200,
    )
    products: list[str] = Field(
        default_factory=list,
        description="Products or features mentioned in the message",
    )
    needs_human: bool = Field(
        description="True when the request cannot be solved by an automatic answer"
    )

Cada detalhe tem um papel:

O prompt fica mais simples, porque não precisa descrever o JSON. Ele só explica a tarefa:

from langchain_core.prompts import ChatPromptTemplate


def create_ticket_prompt() -> ChatPromptTemplate:
    """Create a prompt that triages a customer support message."""
    system = (
        "system",
        "You are a support analyst. Classify the customer message "
        "and extract the requested information.",
    )
    user = ("user", "{message}")
    return ChatPromptTemplate.from_messages([system, user])

Montando a chain

A mudança em relação às partes anteriores é pequena: em vez de llm | StrOutputParser(), a chain usa llm.with_structured_output(TicketAnalysis).

from langchain.chat_models import init_chat_model
from langchain_core.language_models import BaseChatModel

from structured_output.prompts import create_ticket_prompt
from structured_output.schemas import TicketAnalysis


def create_structured_chain(
    llm: BaseChatModel | None = None,
    include_raw: bool = False,
):
    """Create a chain that returns a validated TicketAnalysis instead of text."""
    if llm is None:
        llm = init_chat_model(model="gemini-2.5-flash", model_provider="google_genai")

    structured_llm = llm.with_structured_output(TicketAnalysis, include_raw=include_raw)
    return create_ticket_prompt() | structured_llm

Receber o modelo como parâmetro (llm) não é só capricho: permite trocar de provedor e, principalmente, testar a chain com um modelo falso, sem gastar chamadas de API.

O resultado é um objeto Python de verdade:

chain = create_structured_chain()

ticket = chain.invoke({
    "message": "Fui cobrado duas vezes na fatura de outubro do plano Pro "
               "e preciso do estorno urgente, senão meu cartão estoura."
})

print(type(ticket).__name__)        # TicketAnalysis
print(ticket.model_dump_json(indent=2))

if ticket.priority == "high" and ticket.needs_human:
    print("-> Encaminhar para a fila de atendimento humano")

TicketAnalysis
{
  "category": "billing",
  "priority": "high",
  "summary": "Cliente cobrado em duplicidade no plano Pro pede estorno urgente.",
  "products": ["Plano Pro"],
  "needs_human": true
}

-> Encaminhar para a fila de atendimento humano

O if no final é o ponto de tudo: o código decide com base em campos tipados, sem interpretar texto. Na Parte 4 (Router Chain), é exatamente o campo category que vai escolher qual chain responde a mensagem.

Quando o modelo erra

Mesmo com o schema, a resposta pode não passar na validação: um valor fora do Literal, um resumo longo demais. Nesse caso, por padrão, a chain lança um ValidationError do Pydantic, que diz exatamente qual campo falhou:

from pydantic import ValidationError

try:
    ticket = chain.invoke({"message": mensagem})
except ValidationError as e:
    print(e.errors()[0]["loc"], e.errors()[0]["type"])
    # ('category',) literal_error

Se você prefere tratar o erro como dado em vez de exceção, use include_raw=True. A chain passa a devolver um dicionário com três chaves:

chain = create_structured_chain(include_raw=True)
result = chain.invoke({"message": mensagem})

if result["parsing_error"]:
    logger.warning("Saída inválida: %s", result["raw"].content)
    ticket = None  # tentar de novo, usar um padrão ou mandar para um humano
else:
    ticket = result["parsed"]

Em produção, combine isso com novas tentativas: o tema da Parte 11 (Fallback e Retry).

Expondo numa API

Como o resultado já é um modelo Pydantic, ele se encaixa direto no FastAPI. O mesmo schema vira o response_model do endpoint, e a documentação do Swagger mostra o formato exato da resposta:

from fastapi import FastAPI
from pydantic import BaseModel

from structured_output import TicketAnalysis, create_structured_chain

app = FastAPI()
ticket_chain = create_structured_chain()


class TicketRequest(BaseModel):
    message: str


@app.post("/tickets/triage", response_model=TicketAnalysis)
async def triage(request: TicketRequest):
    return await ticket_chain.ainvoke({"message": request.message})

Repare no ainvoke: toda chain LCEL tem a versão assíncrona, então o endpoint não trava o servidor enquanto espera o modelo.

curl -X POST localhost:8000/tickets/triage \
  -H "Content-Type: application/json" \
  -d '{"message": "Fui cobrado duas vezes na fatura de outubro"}'

Boas práticas

1. Descreva cada campo com o critério, não só o nome

description="Priority" não ajuda em nada. description="high when the customer is blocked or losing money" diz ao modelo como decidir. As descrições são a parte do prompt que mais influencia a qualidade da resposta.

2. Prefira valores fechados

Sempre que o código vai tomar decisões com um campo, use Literal (ou Enum). Texto livre só onde o conteúdo realmente é livre, como o resumo.

3. Peça só o que você vai usar

Cada campo a mais é mais uma chance de erro e mais tokens. Se o código nunca lê um campo, tire-o do schema.

4. Não repita o formato no prompt

O schema já é enviado ao modelo. Descrever o JSON também no prompt duplica a instrução e cria o risco das duas versões divergirem.

5. Tenha um plano para saída inválida

Decida o que acontece quando a validação falha: exceção, include_raw com valor padrão, nova tentativa ou encaminhamento para uma pessoa. Não deixe isso para descobrir em produção.

6. Teste com um modelo falso

Injetar o modelo na função que monta a chain permite testar o fluxo inteiro (validação, decisões, endpoint) sem chamar a API. O exemplo do repositório foi testado assim.

7. Pydantic não é a única opção

with_structured_output também aceita TypedDict e um dicionário JSON Schema; nesses casos a saída é um dict em vez de um objeto. Pydantic costuma ser a melhor escolha em backend, porque valida os dados e conversa com o FastAPI.

Resumo e referências

• Structured Output troca o texto livre por um contrato: um modelo Pydantic que define campos, tipos e valores permitidos.
• llm.with_structured_output(Schema) devolve o objeto já validado; não há parser de texto.
• As descrições dos campos funcionam como instruções para o modelo.
• Saídas inválidas geram ValidationError, ou viram dado com include_raw=True.
• O mesmo schema serve de response_model numa API FastAPI.

Referências

• Repositório pattern_chain (código completo)
• Referência do LangChain: with_structured_output
• Documentação do LangChain: structured output em agentes
• Documentação do Pydantic: fields