Patrón de Agente Reflexivo

Implementa el patrón de reflexión en LangGraph, bucles de generar, criticar y refinar para agentes que mejoran iterativamente su salida.

El patrón de reflexión permite que los agentes se automejoren generando una salida, criticándola y refinándola a partir de la crítica. Esto crea un ciclo de calidad que produce mejores resultados que la generación en una sola pasada.

El Bucle de Reflexión

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

Each iteration:

Generar: Produce la salida inicial (o refina la anterior)
Criticar: Analiza la salida en busca de problemas, lagunas o mejoras
Refinar: Mejora a partir de la crítica
Repetir: Hasta que se alcance el umbral de calidad o el número máximo de iteraciones

Grafo de Reflexión 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

La crítica se retroalimenta en la siguiente llamada de generación como contexto. El generador aborda los puntos de la crítica, creando un ciclo de mejora continua.

Crítica Estructurada con Puntuación

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)}"]
    }

✅Éxito

Las críticas estructuradas con puntuaciones te permiten seguir la mejora a lo largo de las iteraciones y definir umbrales de calidad para la terminación anticipada.

Terminación por Umbral de Calidad

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últiples Perspectivas

Usa varios ángulos de crítica para un comentario 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}"]}

💡Consejo

Varios críticos aportan perspectivas distintas. El generador recibe todas las críticas y debe abordarlas todas en la siguiente iteración.

Mejora de Código con Reflexión

El patrón de reflexión también funciona para la generación 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 Reflexión 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()