Padrão de Agente Reflexivo

Implemente o padrão de reflexão em LangGraph, loops de gerar, criticar e refinar para agentes que melhoram iterativamente sua saída.

O padrão de reflexão permite que os agentes se autoaprimorem gerando uma saída, criticando-a e refinando-a com base na crítica. Isso cria um ciclo de qualidade que produz resultados melhores do que a geração em uma única passagem.

O Loop de Reflexão

Generate → Critique → Refine → Critique → Refine → ... → Accept

Each iteration:

Gerar: Produza a saída inicial (ou refine a anterior)
Criticar: Analise a saída em busca de problemas, lacunas ou melhorias
Refinar: Melhore com base na crítica
Repetir: Até que o limite de qualidade seja atingido ou o número máximo de iterações seja alcançado

Grafo de Reflexão Básico

python

from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from typing_extensions import TypedDict
from typing import Annotated, List
from operator import add

llm = ChatOpenAI(model="gpt-4o")

class ReflectionState(TypedDict):
    topic: str
    drafts: Annotated[List[str], add]  # Accumulate all drafts
    critiques: Annotated[List[str], add]  # Accumulate all critiques
    final: str
    iteration: int
    max_iterations: int

def generate(state: ReflectionState) -> dict:
    context = ""
    if state["critiques"]:
        # Incorporate last critique
        last_critique = state["critiques"][-1]
        context = f"\nPrevious critique to address:\n{last_critique}"

    prompt = f"Write about: {state['topic']}. Be thorough and clear.{context}"
    draft = llm.invoke(prompt).content

    return {
        "drafts": [draft],
        "iteration": state["iteration"] + 1
    }

def critique(state: ReflectionState) -> dict:
    last_draft = state["drafts"][-1]

    prompt = f"""Critique the following text. Identify:
1. Strengths
2. Weaknesses or gaps
3. Specific improvements needed
4. Clarity and structure issues

Text to critique:
{last_draft}

Be constructive and specific."""

    critique_text = llm.invoke(prompt).content
    return {"critiques": [critique_text]}

def should_continue(state: ReflectionState) -> str:
    if state["iteration"] >= state["max_iterations"]:
        return "finalize"
    return "refine"

def finalize(state: ReflectionState) -> dict:
    return {"final": state["drafts"][-1]}

# Build graph
builder = StateGraph(ReflectionState)
builder.add_node("generate", generate)
builder.add_node("critique", critique)
builder.add_node("finalize", finalize)

builder.add_edge(START, "generate")
builder.add_edge("generate", "critique")
builder.add_conditional_edges("critique", should_continue, {
    "refine": "generate",  # Loop back for another iteration
    "finalize": "finalize"
})
builder.add_edge("finalize", END)

app = builder.compile()

# Run
result = app.invoke({
    "topic": "Benefits of functional programming",
    "drafts": [],
    "critiques": [],
    "final": "",
    "iteration": 0,
    "max_iterations": 3
})
print(result["final"])

ℹ️Nota

A crítica é alimentada de volta na próxima chamada de geração como contexto. O gerador aborda os pontos da crítica, criando um ciclo de melhoria contínua.

Crítica Estruturada com Pontuação

python

from pydantic import BaseModel, Field

class CritiqueScore(BaseModel):
    clarity: float = Field(ge=0, le=10)
    accuracy: float = Field(ge=0, le=10)
    completeness: float = Field(ge=0, le=10)
    conciseness: float = Field(ge=0, le=10)
    overall: float = Field(ge=0, le=10)
    issues: list[str]
    suggestions: list[str]

def structured_critique(state: ReflectionState) -> dict:
    from langchain_core.output_parsers import PydanticOutputParser
    parser = PydanticOutputParser(pydantic_object=CritiqueScore)

    prompt = ChatPromptTemplate.from_messages([
        ("system", "Critique the text and provide structured scores. {format_instructions}"),
        ("human", "{text}")
    ])

    chain = prompt | llm | parser
    score = chain.invoke({
        "text": state["drafts"][-1],
        "format_instructions": parser.get_format_instructions()
    })

    return {
        "critiques": [f"Score: {score.overall}/10\nIssues: {', '.join(score.issues)}\nSuggestions: {', '.join(score.suggestions)}"]
    }

✅Sucesso

Críticas estruturadas com pontuações permitem acompanhar a melhoria ao longo das iterações e definir limites de qualidade para encerramento antecipado.

Término por Limiar de Qualidade

python

def quality_router(state: ReflectionState) -> str:
    last_critique = state["critiques"][-1] if state["critiques"] else ""

    # Check if quality is acceptable (simplified — look for score)
    if "Score:" in last_critique:
        try:
            score_line = [l for l in last_critique.split("\n") if "Score:" in l][0]
            score = float(score_line.split("/")[0].split(":")[1].strip())
            if score >= 8.0:  # Quality threshold met
                return "finalize"
        except (IndexError, ValueError):
            pass

    if state["iteration"] >= state["max_iterations"]:
        return "finalize"

    return "refine"

Crítica de Múltiplas Perspectivas

Use múltiplos ângulos de crítica para um feedback completo:

python

def critic_expert(state: ReflectionState) -> dict:
    """Critique from an expert perspective."""
    prompt = f"As a domain expert, critique this: {state['drafts'][-1]}"
    return {"critiques": [f"[Expert] {llm.invoke(prompt).content}"]}

def critic_stylist(state: ReflectionState) -> dict:
    """Critique from a writing style perspective."""
    prompt = f"As a writing stylist, critique the clarity and flow: {state['drafts'][-1]}"
    return {"critiques": [f"[Style] {llm.invoke(prompt).content}"]}

def critic_fact_checker(state: ReflectionState) -> dict:
    """Critique from a factual accuracy perspective."""
    prompt = f"Fact-check this text. Identify any unsupported claims: {state['drafts'][-1]}"
    return {"critiques": [f"[Facts] {llm.invoke(prompt).content}"]}

💡Dica

Múltiplos críticos fornecem perspectivas diferentes. O gerador recebe todas as críticas e deve abordá-las todas na próxima iteração.

Melhoria de Código com Reflexão

O padrão de reflexão também funciona para geração de código:

python

def generate_code(state: ReflectionState) -> dict:
    context = ""
    if state["critiques"]:
        context = f"\nAddress these issues:\n{state['critiques'][-1]}"

    prompt = f"Write Python code for: {state['topic']}. Include tests.{context}"
    code = llm.invoke(prompt).content
    return {"drafts": [code], "iteration": state["iteration"] + 1}

def critique_code(state: ReflectionState) -> dict:
    code = state["drafts"][-1]
    prompt = f"""Review this Python code for:
1. Correctness — will it work?
2. Edge cases — what could break it?
3. Performance — can it be optimized?
4. Style — does it follow PEP 8?
5. Security — are there vulnerabilities?

Code:
{code}

Be specific about each issue."""
    critique_text = llm.invoke(prompt).content
    return {"critiques": [critique_text]}

Agente de Reflexão Completo

python

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict, Annotated
from typing import List
from operator import add

class AgentState(TypedDict):
    task: str
    outputs: Annotated[List[str], add]
    feedback: Annotated[List[str], add]
    iteration: int
    max_iterations: int

def generate(state: AgentState) -> dict:
    context = ""
    if state["feedback"]:
        context = f"\nPrevious feedback: {state['feedback'][-1]}"
    prompt = f"Task: {state['task']}. Produce your best work.{context}"
    return {"outputs": [llm.invoke(prompt).content], "iteration": state["iteration"] + 1}

def reflect(state: AgentState) -> dict:
    prompt = f"Critique this output constructively:\n{state['outputs'][-1]}"
    return {"feedback": [llm.invoke(prompt).content]}

def router(state: AgentState) -> str:
    return "refine" if state["iteration"] < state["max_iterations"] else "done"

builder = StateGraph(AgentState)
builder.add_node("generate", generate)
builder.add_node("reflect", reflect)
builder.add_edge(START, "generate")
builder.add_edge("generate", "reflect")
builder.add_conditional_edges("reflect", router, {"refine": "generate", "done": END})
app = builder.compile()