Módulo 9: Human-in-the-Loop

User Feedback Loops

Descripción de la cápsula

Hasta ahora, HITL ha sido binario: el humano aprueba o rechaza. Pero la interacción más valiosa no es "sí/no" — es feedback iterativo. "Este resumen es demasiado técnico. Simplificalo." El agente reescribe. "Mejor, pero agrega más ejemplos." El agente agrega ejemplos. Este ciclo de refinamiento produce resultados que ni el humano ni la IA podrían lograr por separado.

El patrón es simple pero poderoso: Review → Feedback → Revise → Review again → Approve. Cada ronda de feedback se acumula en el estado del grafo, y el agente usa todo el historial de correcciones para mejorar su siguiente intento. No empieza de cero — aprende de lo que ya le dijiste.

Esto transforma al agente de un generador de "primer intento" a un colaborador que refina su trabajo contigo. Y lo mejor: la mecánica es una extensión directa de los interrupts y el estado editable que ya dominas.

El patrón: Review → Feedback → Revise → Approve

Visualiza el flujo completo:

generate → interrupt (muestra output al usuario)
                ↓
        ¿Qué dice el usuario?
        ├── "approved" → continuar al siguiente paso
        └── "demasiado técnico" → loop de vuelta a generate
                                    (con el feedback como contexto adicional)
                                    → interrupt (muestra nueva versión)
                                    ↓
                            ¿Qué dice el usuario?
                            ├── "approved" → continuar
                            └── "agrega ejemplos" → loop de vuelta...

Cada iteración del loop es un checkpoint. Si el proceso se cae a mitad de una revisión, el agente resume exactamente donde estaba — con todo el feedback acumulado.

Implementando un feedback loop básico

El feedback loop más simple: un nodo genera contenido, un interrupt lo muestra al usuario, y según la respuesta, continúa o regenera:

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class FeedbackState(TypedDict):
    topic: str
    draft: str
    feedback: Annotated[list[str], operator.add]
    iteration: int
    max_iterations: int
    approved: bool

def generate_summary(state: FeedbackState) -> dict:
    """Genera o regenera un resumen incorporando feedback previo."""
    topic = state["topic"]
    feedback_history = state.get("feedback", [])
    iteration = state.get("iteration", 0)

    if iteration == 0:
        draft = (
            f"Resumen técnico sobre '{topic}': "
            f"Los modelos de lenguaje utilizan arquitecturas transformer con mecanismos "
            f"de atención multi-cabeza para procesar secuencias de tokens. La función "
            f"de pérdida cross-entropy optimiza la distribución de probabilidad sobre "
            f"el vocabulario. El fine-tuning con RLHF alinea las salidas del modelo "
            f"con preferencias humanas mediante una función de recompensa aprendida."
        )
    else:
        latest_feedback = feedback_history[-1] if feedback_history else ""
        all_feedback = " | ".join(feedback_history)

        if "simplifica" in latest_feedback.lower() or "técnico" in latest_feedback.lower():
            draft = (
                f"Resumen sobre '{topic}': "
                f"Los modelos de lenguaje son programas que aprenden a escribir texto "
                f"leyendo millones de documentos. Funcionan prediciendo la siguiente palabra "
                f"más probable. Después de entrenarlos, se ajustan con feedback humano para "
                f"que sus respuestas sean más útiles y seguras."
            )
        elif "ejemplo" in latest_feedback.lower():
            draft = (
                f"Resumen sobre '{topic}': "
                f"Los modelos de lenguaje aprenden a escribir texto prediciendo la "
                f"siguiente palabra. Por ejemplo, si les das 'El gato se sentó en el...', "
                f"predicen 'sofá' o 'piso'. Después se ajustan con feedback humano — "
                f"como un editor que revisa borradores. Ejemplo: ChatGPT usa este proceso "
                f"para responder preguntas de forma conversacional."
            )
        else:
            draft = (
                f"Resumen revisado sobre '{topic}' "
                f"(iteración {iteration + 1}, incorporando: {all_feedback}): "
                f"Versión mejorada del contenido basada en todas las correcciones recibidas."
            )

    return {
        "draft": draft,
        "iteration": iteration + 1,
    }

def review_draft(state: FeedbackState) -> dict:
    """Muestra el borrador al usuario y espera feedback."""
    iteration = state["iteration"]
    draft = state["draft"]

    response = interrupt({
        "action": "review_draft",
        "iteration": iteration,
        "draft": draft,
        "instruction": "Responde 'approved' para aceptar, o escribe tu feedback para pedir cambios.",
        "feedback_so_far": state.get("feedback", []),
    })

    if response.lower() == "approved":
        return {"approved": True}

    return {
        "feedback": [response],
        "approved": False,
    }

def should_continue(state: FeedbackState) -> str:
    if state.get("approved", False):
        return "deliver"

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

    return "revise"

def deliver(state: FeedbackState) -> dict:
    """Entrega el resultado final."""
    return {}

graph_builder = StateGraph(FeedbackState)
graph_builder.add_node("generate", generate_summary)
graph_builder.add_node("review", review_draft)
graph_builder.add_node("deliver", deliver)

graph_builder.add_edge(START, "generate")
graph_builder.add_edge("generate", "review")
graph_builder.add_conditional_edges(
    "review", should_continue,
    {"revise": "generate", "deliver": "deliver"},
)
graph_builder.add_edge("deliver", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "feedback-loop-001"}}

print("=== Ronda 1: generación inicial ===")
result = graph.invoke(
    {
        "topic": "modelos de lenguaje",
        "draft": "",
        "feedback": [],
        "iteration": 0,
        "max_iterations": 3,
        "approved": False,
    },
    config,
)
state = graph.get_state(config)
print(f"Borrador: {state.values['draft'][:80]}...")
print(f"Iteración: {state.values['iteration']}")

print("\n=== Ronda 1: feedback — 'demasiado técnico' ===")
result = graph.invoke(
    Command(resume="Demasiado técnico. Simplificalo para alguien sin background en ML."),
    config,
)
state = graph.get_state(config)
print(f"Borrador: {state.values['draft'][:80]}...")
print(f"Iteración: {state.values['iteration']}")
print(f"Feedback acumulado: {state.values['feedback']}")

print("\n=== Ronda 2: feedback — 'agrega ejemplos' ===")
result = graph.invoke(
    Command(resume="Mejor, pero agrega ejemplos concretos."),
    config,
)
state = graph.get_state(config)
print(f"Borrador: {state.values['draft'][:80]}...")
print(f"Iteración: {state.values['iteration']}")
print(f"Feedback acumulado: {state.values['feedback']}")

print("\n=== Ronda 3: approved ===")
result = graph.invoke(Command(resume="approved"), config)
state = graph.get_state(config)
print(f"Borrador final: {state.values['draft']}")
print(f"Aprobado: {state.values['approved']}")
print(f"Total iteraciones: {state.values['iteration']}")
print(f"Historial de feedback: {state.values['feedback']}")
# Output esperado:
# === Ronda 1: generación inicial ===
# Borrador: Resumen técnico sobre 'modelos de lenguaje': Los modelos de lenguaje uti...
# Iteración: 1
#
# === Ronda 1: feedback — 'demasiado técnico' ===
# Borrador: Resumen sobre 'modelos de lenguaje': Los modelos de lenguaje son program...
# Iteración: 2
# Feedback acumulado: ['Demasiado técnico. Simplificalo para alguien sin background en ML.']
#
# === Ronda 2: feedback — 'agrega ejemplos' ===
# Borrador: Resumen sobre 'modelos de lenguaje': Los modelos de lenguaje aprenden a ...
# Iteración: 3
# Feedback acumulado: ['Demasiado técnico. Simplificalo para alguien sin background en ML.', 'Mejor, pero agrega ejemplos concretos.']
#
# === Ronda 3: approved ===
# Borrador final: Resumen sobre 'modelos de lenguaje': Los modelos de lenguaje aprenden a escribir texto prediciendo la siguiente palabra. Por ejemplo, si les das 'El gato se sentó en el...', predicen 'sofá' o 'piso'. Después se ajustan con feedback humano — como un editor que revisa borradores. Ejemplo: ChatGPT usa este proceso para responder preguntas de forma conversacional.
# Aprobado: True
# Total iteraciones: 3
# Historial de feedback: ['Demasiado técnico. Simplificalo para alguien sin background en ML.', 'Mejor, pero agrega ejemplos concretos.']

Observa la mecánica:

feedback: Annotated[list[str], operator.add] — acumula todo el feedback sin perder rondas anteriores
El nodo generate lee el feedback previo y adapta su output
El nodo review usa interrupt() para pausar y mostrar el borrador
El conditional edge decide: si aprobado → entregar, si hay feedback → regenerar
max_iterations previene loops infinitos

Acumulando feedback en el estado

La clave del patrón es que el feedback no reemplaza — se acumula. Cada ronda agrega contexto:

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class AccumulativeState(TypedDict):
    task: str
    output: str
    feedback_history: Annotated[list[dict], operator.add]
    revision_count: int

def produce_output(state: AccumulativeState) -> dict:
    """Genera output usando TODO el historial de feedback."""
    feedback_history = state.get("feedback_history", [])
    revision = state.get("revision_count", 0)

    context_parts = [f"Tarea: {state['task']}"]
    for entry in feedback_history:
        context_parts.append(
            f"Ronda {entry['round']}: usuario dijo '{entry['feedback']}'"
        )

    context = "\n".join(context_parts)

    if revision == 0:
        output = f"[v1] Propuesta inicial para: {state['task']}"
    else:
        corrections = [e["feedback"] for e in feedback_history]
        output = (
            f"[v{revision + 1}] Propuesta revisada para: {state['task']}. "
            f"Incorporando {len(corrections)} correcciones: {', '.join(corrections)}"
        )

    return {
        "output": output,
        "revision_count": revision + 1,
    }

def collect_feedback(state: AccumulativeState) -> dict:
    response = interrupt({
        "current_output": state["output"],
        "revision": state["revision_count"],
        "history": state.get("feedback_history", []),
        "prompt": "Escribe 'ok' para aprobar o tu corrección.",
    })

    if response.lower() in ("ok", "approved", "lgtm"):
        return {}

    return {
        "feedback_history": [{
            "round": state["revision_count"],
            "feedback": response,
        }],
    }

def route(state: AccumulativeState) -> str:
    history = state.get("feedback_history", [])
    if not history:
        return "finalize"

    latest_round = max(e["round"] for e in history)
    if latest_round < state["revision_count"]:
        return "finalize"

    if state["revision_count"] >= 5:
        return "finalize"

    return "revise"

def finalize(state: AccumulativeState) -> dict:
    return {}

graph_builder = StateGraph(AccumulativeState)
graph_builder.add_node("produce", produce_output)
graph_builder.add_node("collect", collect_feedback)
graph_builder.add_node("finalize", finalize)

graph_builder.add_edge(START, "produce")
graph_builder.add_edge("produce", "collect")
graph_builder.add_conditional_edges(
    "collect", route,
    {"revise": "produce", "finalize": "finalize"},
)
graph_builder.add_edge("finalize", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "accumulative-001"}}

graph.invoke(
    {
        "task": "Diseñar landing page para SaaS de analytics",
        "output": "",
        "feedback_history": [],
        "revision_count": 0,
    },
    config,
)

state = graph.get_state(config)
print(f"v1: {state.values['output']}")

graph.invoke(
    Command(resume="El hero section necesita un CTA más claro"),
    config,
)
state = graph.get_state(config)
print(f"v2: {state.values['output']}")

graph.invoke(Command(resume="ok"), config)
state = graph.get_state(config)
print(f"Final: {state.values['output']}")
print(f"Historial: {state.values['feedback_history']}")
# Output esperado:
# v1: [v1] Propuesta inicial para: Diseñar landing page para SaaS de analytics
# v2: [v2] Propuesta revisada para: Diseñar landing page para SaaS de analytics. Incorporando 1 correcciones: El hero section necesita un CTA más claro
# Final: [v2] Propuesta revisada para: Diseñar landing page para SaaS de analytics. Incorporando 1 correcciones: El hero section necesita un CTA más claro
# Historial: [{'round': 1, 'feedback': 'El hero section necesita un CTA más claro'}]

El patrón feedback_history: Annotated[list[dict], operator.add] con diccionarios que incluyen round y feedback te da trazabilidad completa. Puedes saber exactamente qué pidió el usuario en cada ronda y cómo evolucionó el output.

Incorporando feedback al prompt, no reemplazando el output

Un error común: recibir feedback y generar una respuesta completamente nueva desde cero. El patrón correcto es agregar el feedback como contexto adicional para que el modelo mejore sobre lo que ya tiene:

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from langchain.chat_models import init_chat_model

class WriterState(TypedDict):
    topic: str
    draft: str
    feedback: Annotated[list[str], operator.add]
    iteration: int
    approved: bool

llm = init_chat_model("openai:gpt-4.1-nano")

def write_draft(state: WriterState) -> dict:
    feedback = state.get("feedback", [])
    iteration = state.get("iteration", 0)

    if iteration == 0:
        prompt = (
            f"Escribe un párrafo de resumen ejecutivo sobre: {state['topic']}. "
            f"Máximo 3 oraciones. Tono profesional."
        )
    else:
        prompt = (
            f"Versión anterior:\n{state['draft']}\n\n"
            f"Feedback del usuario en esta ronda: {feedback[-1]}\n\n"
            f"Historial completo de feedback: {feedback}\n\n"
            f"Reescribe el párrafo incorporando TODO el feedback. "
            f"Mantén lo que funcionaba. Máximo 3 oraciones."
        )

    response = llm.invoke(prompt)

    return {
        "draft": response.content,
        "iteration": iteration + 1,
    }

def human_review(state: WriterState) -> dict:
    response = interrupt({
        "draft": state["draft"],
        "iteration": state["iteration"],
        "feedback_history": state.get("feedback", []),
        "instruction": "'approved' para aceptar, o escribe tu corrección.",
    })

    if response.lower() == "approved":
        return {"approved": True}

    return {"feedback": [response], "approved": False}

def route_review(state: WriterState) -> str:
    if state.get("approved", False):
        return "done"
    if state.get("iteration", 0) >= 3:
        return "done"
    return "rewrite"

def done(state: WriterState) -> dict:
    return {}

graph_builder = StateGraph(WriterState)
graph_builder.add_node("write", write_draft)
graph_builder.add_node("review", human_review)
graph_builder.add_node("done", done)

graph_builder.add_edge(START, "write")
graph_builder.add_edge("write", "review")
graph_builder.add_conditional_edges(
    "review", route_review,
    {"rewrite": "write", "done": "done"},
)
graph_builder.add_edge("done", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "writer-feedback-001"}}

print("=== Generación inicial ===")
graph.invoke(
    {
        "topic": "beneficios de microservicios vs monolitos",
        "draft": "",
        "feedback": [],
        "iteration": 0,
        "approved": False,
    },
    config,
)
state = graph.get_state(config)
print(f"Borrador v1:\n{state.values['draft']}\n")

print("=== Feedback: simplificar ===")
graph.invoke(
    Command(resume="Demasiado formal. Hazlo más conversacional, como un blog post."),
    config,
)
state = graph.get_state(config)
print(f"Borrador v2:\n{state.values['draft']}\n")

print("=== Aprobado ===")
graph.invoke(Command(resume="approved"), config)
state = graph.get_state(config)
print(f"Versión final:\n{state.values['draft']}")
print(f"Iteraciones: {state.values['iteration']}")
print(f"Feedback recibido: {state.values['feedback']}")
# Output esperado (contenido varía por LLM):
# === Generación inicial ===
# Borrador v1:
# [Párrafo profesional sobre microservicios vs monolitos]
#
# === Feedback: simplificar ===
# Borrador v2:
# [Párrafo más conversacional incorporando el feedback]
#
# === Aprobado ===
# Versión final:
# [Versión final del párrafo]
# Iteraciones: 2
# Feedback recibido: ['Demasiado formal. Hazlo más conversacional, como un blog post.']

La diferencia clave está en el prompt de reescritura:

Versión anterior: [lo que ya generó]
Feedback del usuario: [corrección específica]
Historial completo: [todas las correcciones]
→ Reescribe incorporando TODO. Mantén lo que funcionaba.

Esto le dice al modelo: "no empieces de cero, mejora lo que hay." El resultado es una evolución natural, no un reinicio.

Multi-point feedback: revisiones en diferentes etapas

El feedback no tiene que ser solo al final. Puedes pedir feedback en múltiples puntos del pipeline — cada uno sobre un aspecto diferente del trabajo:

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class ResearchState(TypedDict):
    query: str
    research_plan: str
    raw_findings: str
    final_report: str
    stage_feedback: Annotated[list[dict], operator.add]

def create_plan(state: ResearchState) -> dict:
    plan = (
        f"Plan de investigación para '{state['query']}':\n"
        f"1. Buscar papers académicos recientes\n"
        f"2. Revisar documentación oficial\n"
        f"3. Analizar benchmarks existentes\n"
        f"4. Sintetizar hallazgos en reporte"
    )
    return {"research_plan": plan}

def review_plan(state: ResearchState) -> dict:
    """Feedback punto 1: el plan de investigación."""
    response = interrupt({
        "stage": "plan_review",
        "content": state["research_plan"],
        "instruction": "'ok' para continuar, o sugiere cambios al plan.",
    })

    if response.lower() == "ok":
        return {}

    updated_plan = f"{state['research_plan']}\n[Ajuste: {response}]"
    return {
        "research_plan": updated_plan,
        "stage_feedback": [{"stage": "plan", "feedback": response}],
    }

def execute_research(state: ResearchState) -> dict:
    findings = (
        f"Hallazgos para '{state['query']}':\n"
        f"- Paper A: resultados positivos en 85% de casos\n"
        f"- Paper B: limitaciones en datasets pequeños\n"
        f"- Docs oficiales: nueva API disponible en v2.0\n"
        f"- Benchmark: 40% mejora respecto a baseline"
    )
    return {"raw_findings": findings}

def review_findings(state: ResearchState) -> dict:
    """Feedback punto 2: hallazgos crudos."""
    response = interrupt({
        "stage": "findings_review",
        "content": state["raw_findings"],
        "instruction": "'ok' para sintetizar, o indica qué falta o sobra.",
    })

    if response.lower() == "ok":
        return {}

    updated = f"{state['raw_findings']}\n[Nota del reviewer: {response}]"
    return {
        "raw_findings": updated,
        "stage_feedback": [{"stage": "findings", "feedback": response}],
    }

def synthesize_report(state: ResearchState) -> dict:
    report = (
        f"Reporte: {state['query']}\n"
        f"Basado en: {state['raw_findings'][:100]}...\n"
        f"Conclusión: La tecnología muestra resultados prometedores con "
        f"áreas de mejora identificadas."
    )
    return {"final_report": report}

def review_report(state: ResearchState) -> dict:
    """Feedback punto 3: reporte final."""
    response = interrupt({
        "stage": "report_review",
        "content": state["final_report"],
        "instruction": "'approved' para entregar, o pide ajustes al reporte.",
    })

    if response.lower() == "approved":
        return {}

    return {
        "stage_feedback": [{"stage": "report", "feedback": response}],
    }

graph_builder = StateGraph(ResearchState)
graph_builder.add_node("plan", create_plan)
graph_builder.add_node("review_plan", review_plan)
graph_builder.add_node("research", execute_research)
graph_builder.add_node("review_findings", review_findings)
graph_builder.add_node("synthesize", synthesize_report)
graph_builder.add_node("review_report", review_report)

graph_builder.add_edge(START, "plan")
graph_builder.add_edge("plan", "review_plan")
graph_builder.add_edge("review_plan", "research")
graph_builder.add_edge("research", "review_findings")
graph_builder.add_edge("review_findings", "synthesize")
graph_builder.add_edge("synthesize", "review_report")
graph_builder.add_edge("review_report", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "multi-feedback-001"}}

print("=== Etapa 1: Review del plan ===")
graph.invoke(
    {
        "query": "Estado del arte en RAG",
        "research_plan": "",
        "raw_findings": "",
        "final_report": "",
        "stage_feedback": [],
    },
    config,
)
state = graph.get_state(config)
print(f"Plan:\n{state.values['research_plan']}\n")

print("=== Feedback al plan ===")
graph.invoke(
    Command(resume="Agrega también revisión de repos de GitHub populares"),
    config,
)
state = graph.get_state(config)
print(f"Plan ajustado:\n{state.values['research_plan']}\n")

print("=== Etapa 2: Review de hallazgos ===")
graph.invoke(Command(resume="ok"), config)
state = graph.get_state(config)
print(f"Hallazgos:\n{state.values['raw_findings']}\n")

print("=== Etapa 3: Aprobación del reporte ===")
graph.invoke(Command(resume="approved"), config)
state = graph.get_state(config)
print(f"Reporte final:\n{state.values['final_report']}")
print(f"\nFeedback por etapa: {state.values['stage_feedback']}")
# Output esperado:
# === Etapa 1: Review del plan ===
# Plan:
# Plan de investigación para 'Estado del arte en RAG':
# 1. Buscar papers académicos recientes
# 2. Revisar documentación oficial
# 3. Analizar benchmarks existentes
# 4. Sintetizar hallazgos en reporte
#
# === Feedback al plan ===
# Plan ajustado:
# Plan de investigación para 'Estado del arte en RAG':
# 1. Buscar papers académicos recientes
# 2. Revisar documentación oficial
# 3. Analizar benchmarks existentes
# 4. Sintetizar hallazgos en reporte
# [Ajuste: Agrega también revisión de repos de GitHub populares]
#
# === Etapa 2: Review de hallazgos ===
# Hallazgos:
# Hallazgos para 'Estado del arte en RAG':
# - Paper A: resultados positivos en 85% de casos
# ...
#
# === Etapa 3: Aprobación del reporte ===
# Reporte final:
# Reporte: Estado del arte en RAG
# ...
#
# Feedback por etapa: [{'stage': 'plan', 'feedback': 'Agrega también revisión de repos de GitHub populares'}]

Tres puntos de revisión, cada uno sobre un aspecto diferente:

Punto de review	Qué revisa el usuario	Valor
Plan de investigación	Las fuentes y el enfoque	Evita investigar en la dirección equivocada
Hallazgos crudos	Los datos recopilados	Detecta datos faltantes o irrelevantes antes de sintetizar
Reporte final	El producto entregable	Última oportunidad de ajustar antes de entregar

Cuándo los feedback loops perjudican

No todo merece un loop de refinamiento. Demasiadas iteraciones producen retornos decrecientes:

Iteración 1: "Demasiado técnico"    → mejora 70%  (enorme)
Iteración 2: "Agrega ejemplos"      → mejora 20%  (significativa)
Iteración 3: "Cambia una palabra"   → mejora 3%   (marginal)
Iteración 4: "Hmm, tal vez vuelve"  → mejora -5%  (contraproducente)

Regla práctica: máximo 3 rondas de feedback. Después de 3, entrega con un disclaimer:

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

MAX_FEEDBACK_ROUNDS = 3

class BoundedState(TypedDict):
    content: str
    feedback: Annotated[list[str], operator.add]
    round_num: int
    status: str

def generate(state: BoundedState) -> dict:
    round_num = state.get("round_num", 0)
    feedback = state.get("feedback", [])

    if round_num == 0:
        content = "Contenido inicial generado por el agente."
    else:
        content = (
            f"Contenido revisado (ronda {round_num + 1}). "
            f"Incorporando: {feedback[-1] if feedback else 'N/A'}"
        )

    return {"content": content, "round_num": round_num + 1}

def review(state: BoundedState) -> dict:
    round_num = state["round_num"]

    if round_num >= MAX_FEEDBACK_ROUNDS:
        print(f"  ⚠️  Máximo de {MAX_FEEDBACK_ROUNDS} rondas alcanzado. Entregando versión actual.")
        return {"status": "max_rounds_reached"}

    response = interrupt({
        "content": state["content"],
        "round": round_num,
        "remaining_rounds": MAX_FEEDBACK_ROUNDS - round_num,
        "message": f"Tienes {MAX_FEEDBACK_ROUNDS - round_num} rondas de revisión restantes.",
    })

    if response.lower() in ("approved", "ok"):
        return {"status": "approved"}

    return {
        "feedback": [response],
        "status": "needs_revision",
    }

def route(state: BoundedState) -> str:
    status = state.get("status", "")
    if status in ("approved", "max_rounds_reached"):
        return "deliver"
    return "revise"

def deliver(state: BoundedState) -> dict:
    return {}

graph_builder = StateGraph(BoundedState)
graph_builder.add_node("generate", generate)
graph_builder.add_node("review", review)
graph_builder.add_node("deliver", deliver)

graph_builder.add_edge(START, "generate")
graph_builder.add_edge("generate", "review")
graph_builder.add_conditional_edges(
    "review", route,
    {"revise": "generate", "deliver": "deliver"},
)
graph_builder.add_edge("deliver", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "bounded-feedback-001"}}

graph.invoke(
    {"content": "", "feedback": [], "round_num": 0, "status": ""},
    config,
)
graph.invoke(Command(resume="Hazlo más corto"), config)
graph.invoke(Command(resume="Agrega datos"), config)

state = graph.get_state(config)
print(f"Ronda actual: {state.values['round_num']}")
print(f"Estado: {state.values['status']}")
print(f"Contenido: {state.values['content']}")
print(f"Feedback: {state.values['feedback']}")
# Output esperado:
#   ⚠️  Máximo de 3 rondas alcanzado. Entregando versión actual.
# Ronda actual: 3
# Estado: max_rounds_reached
# Contenido: Contenido revisado (ronda 3). Incorporando: Agrega datos
# Feedback: ['Hazlo más corto', 'Agrega datos']

La protección contra loops infinitos es obligatoria en producción. Sin ella, un usuario indeciso puede generar costos de API indefinidos.

Troubleshooting

Problema 1: "El feedback no se acumula, solo veo el último"

Síntoma: Después de 3 rondas de feedback, state.feedback solo contiene la última corrección.

Causa: Usaste feedback: str en vez de feedback: Annotated[list[str], operator.add].

Solución: Usa una lista con operator.add para acumular:

# ❌ Solo guarda el último feedback
class State(TypedDict):
    feedback: str

# ✅ Acumula todo el feedback
class State(TypedDict):
    feedback: Annotated[list[str], operator.add]

Problema 2: "El agente no mejora con el feedback — genera algo completamente diferente"

Síntoma: Cada revisión ignora lo bueno de la versión anterior.

Causa: El prompt de regeneración no incluye la versión anterior ni el historial de feedback.

Solución: Incluye ambos en el prompt:

prompt = (
    f"Versión anterior:\n{state['draft']}\n\n"
    f"Feedback del usuario: {state['feedback'][-1]}\n\n"
    f"Historial de feedback: {state['feedback']}\n\n"
    f"Mejora la versión anterior incorporando el feedback. Mantén lo que ya funciona."
)

Problema 3: "El loop de feedback nunca termina"

Síntoma: El usuario sigue dando feedback y el agente sigue regenerando indefinidamente.

Causa: No hay límite de iteraciones (max_iterations).

Solución: Agrega un contador y un tope:

def route(state):
    if state.get("approved"):
        return "deliver"
    if state["iteration"] >= state["max_iterations"]:
        return "deliver"
    return "revise"

Problema 4: "El interrupt no muestra la información correcta al usuario"

Síntoma: El usuario recibe una pausa pero no sabe qué está revisando.

Causa: El valor pasado a interrupt() no tiene contexto suficiente.

Solución: Pasa un dict rico en contexto:

# ❌ Sin contexto
interrupt("Esperando feedback")

# ✅ Con contexto completo
interrupt({
    "draft": state["draft"],
    "iteration": state["iteration"],
    "feedback_so_far": state["feedback"],
    "instruction": "'approved' para aceptar, o escribe tu corrección.",
})

Ejercicios

Ejercicio 1: Feedback loop para títulos (Fácil)

Crea un grafo que genere un título para un artículo. El usuario puede dar feedback hasta 3 veces. El estado debe acumular el feedback. Simula la interacción sin LLM (usa strings hardcodeados que cambien según el feedback).

Ver solución

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class TitleState(TypedDict):
    topic: str
    title: str
    feedback: Annotated[list[str], operator.add]
    iteration: int
    approved: bool

def generate_title(state: TitleState) -> dict:
    iteration = state.get("iteration", 0)
    feedback = state.get("feedback", [])

    titles = [
        f"Guía Completa de {state['topic']}: Todo lo que Necesitas Saber",
        f"{state['topic']} Simplificado: Una Guía Práctica",
        f"Domina {state['topic']} en 2025: Guía Paso a Paso",
        f"{state['topic']}: De Cero a Producción",
    ]

    idx = min(iteration, len(titles) - 1)
    return {"title": titles[idx], "iteration": iteration + 1}

def review_title(state: TitleState) -> dict:
    response = interrupt({
        "title": state["title"],
        "iteration": state["iteration"],
        "instruction": "'approved' o escribe qué cambiar.",
    })

    if response.lower() == "approved":
        return {"approved": True}
    return {"feedback": [response], "approved": False}

def route(state: TitleState) -> str:
    if state.get("approved"):
        return "done"
    if state["iteration"] >= 3:
        return "done"
    return "regenerate"

def done(state: TitleState) -> dict:
    return {}

graph_builder = StateGraph(TitleState)
graph_builder.add_node("generate", generate_title)
graph_builder.add_node("review", review_title)
graph_builder.add_node("done", done)

graph_builder.add_edge(START, "generate")
graph_builder.add_edge("generate", "review")
graph_builder.add_conditional_edges(
    "review", route,
    {"regenerate": "generate", "done": "done"},
)
graph_builder.add_edge("done", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "title-feedback"}}

graph.invoke(
    {"topic": "Kubernetes", "title": "", "feedback": [], "iteration": 0, "approved": False},
    config,
)
state = graph.get_state(config)
print(f"v1: {state.values['title']}")

graph.invoke(Command(resume="Demasiado genérico"), config)
state = graph.get_state(config)
print(f"v2: {state.values['title']}")

graph.invoke(Command(resume="approved"), config)
state = graph.get_state(config)
print(f"Final: {state.values['title']}")
print(f"Feedback: {state.values['feedback']}")
# Output esperado:
# v1: Guía Completa de Kubernetes: Todo lo que Necesitas Saber
# v2: Kubernetes Simplificado: Una Guía Práctica
# Final: Kubernetes Simplificado: Una Guía Práctica
# Feedback: ['Demasiado genérico']

Ejercicio 2: Feedback con LLM real (Medio)

Usa un LLM real (init_chat_model) para generar un párrafo sobre un tema. Implementa un feedback loop donde el prompt de regeneración incluya la versión anterior y todo el historial de feedback. Limita a 3 rondas.

Ver solución

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from langchain.chat_models import init_chat_model

class State(TypedDict):
    topic: str
    paragraph: str
    feedback: Annotated[list[str], operator.add]
    iteration: int
    approved: bool

llm = init_chat_model("openai:gpt-4.1-nano")

def write(state: State) -> dict:
    iteration = state.get("iteration", 0)
    feedback = state.get("feedback", [])

    if iteration == 0:
        prompt = f"Escribe un párrafo de 3 oraciones sobre: {state['topic']}"
    else:
        prompt = (
            f"Versión anterior:\n{state['paragraph']}\n\n"
            f"Feedback del usuario: {feedback[-1]}\n"
            f"Todo el feedback previo: {feedback}\n\n"
            f"Reescribe mejorando según el feedback. Mantén lo bueno. 3 oraciones máximo."
        )

    response = llm.invoke(prompt)
    return {"paragraph": response.content, "iteration": iteration + 1}

def review(state: State) -> dict:
    response = interrupt({
        "paragraph": state["paragraph"],
        "round": state["iteration"],
        "remaining": 3 - state["iteration"],
    })

    if response.lower() == "approved":
        return {"approved": True}
    return {"feedback": [response], "approved": False}

def route(state: State) -> str:
    if state.get("approved") or state["iteration"] >= 3:
        return "end"
    return "rewrite"

def end(state: State) -> dict:
    return {}

builder = StateGraph(State)
builder.add_node("write", write)
builder.add_node("review", review)
builder.add_node("end", end)

builder.add_edge(START, "write")
builder.add_edge("write", "review")
builder.add_conditional_edges("review", route, {"rewrite": "write", "end": "end"})
builder.add_edge("end", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "llm-feedback"}}

graph.invoke(
    {"topic": "ventajas de Rust", "paragraph": "", "feedback": [], "iteration": 0, "approved": False},
    config,
)
state = graph.get_state(config)
print(f"v1: {state.values['paragraph'][:100]}...")

graph.invoke(Command(resume="Hazlo más casual, como si hablaras con un amigo"), config)
state = graph.get_state(config)
print(f"v2: {state.values['paragraph'][:100]}...")

graph.invoke(Command(resume="approved"), config)
state = graph.get_state(config)
print(f"Final: {state.values['paragraph']}")
print(f"Iteraciones: {state.values['iteration']}")
# Output esperado (contenido varía por LLM):
# v1: [Párrafo formal sobre Rust]...
# v2: [Párrafo más casual sobre Rust]...
# Final: [Versión final del párrafo]
# Iteraciones: 2

Ejercicio 3: Multi-point feedback pipeline (Medio)

Crea un pipeline de 4 etapas (planificar → investigar → redactar → formatear) con puntos de review después de las etapas 1 y 3. El usuario puede dar feedback en cada punto. Usa stage_feedback: Annotated[list[dict], operator.add] para rastrear de qué etapa vino cada feedback.

Ver solución

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class PipelineState(TypedDict):
    topic: str
    plan: str
    research: str
    draft: str
    formatted: str
    stage_feedback: Annotated[list[dict], operator.add]

def plan(state: PipelineState) -> dict:
    return {"plan": f"Plan para '{state['topic']}': 3 secciones, 2 fuentes por sección"}

def review_plan(state: PipelineState) -> dict:
    response = interrupt({"stage": "plan", "content": state["plan"]})
    if response.lower() == "ok":
        return {}
    return {
        "plan": f"{state['plan']} [Ajuste: {response}]",
        "stage_feedback": [{"stage": "plan", "feedback": response}],
    }

def research(state: PipelineState) -> dict:
    return {"research": f"Datos sobre '{state['topic']}': 6 fuentes encontradas"}

def draft(state: PipelineState) -> dict:
    return {"draft": f"Borrador basado en plan: {state['plan'][:50]}..."}

def review_draft(state: PipelineState) -> dict:
    response = interrupt({"stage": "draft", "content": state["draft"]})
    if response.lower() == "ok":
        return {}
    return {
        "draft": f"{state['draft']} [Corrección: {response}]",
        "stage_feedback": [{"stage": "draft", "feedback": response}],
    }

def format_output(state: PipelineState) -> dict:
    return {"formatted": f"[FORMATTED] {state['draft']}"}

builder = StateGraph(PipelineState)
builder.add_node("plan", plan)
builder.add_node("review_plan", review_plan)
builder.add_node("research", research)
builder.add_node("draft", draft)
builder.add_node("review_draft", review_draft)
builder.add_node("format", format_output)

builder.add_edge(START, "plan")
builder.add_edge("plan", "review_plan")
builder.add_edge("review_plan", "research")
builder.add_edge("research", "draft")
builder.add_edge("draft", "review_draft")
builder.add_edge("review_draft", "format")
builder.add_edge("format", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "pipeline-feedback"}}

graph.invoke(
    {"topic": "GraphQL vs REST", "plan": "", "research": "", "draft": "", "formatted": "", "stage_feedback": []},
    config,
)
graph.invoke(Command(resume="Agrega comparación de rendimiento"), config)

state = graph.get_state(config)
print(f"Plan ajustado: {state.values['plan']}")

graph.invoke(Command(resume="ok"), config)
state = graph.get_state(config)
print(f"Output: {state.values['formatted']}")
print(f"Feedback: {state.values['stage_feedback']}")
# Output esperado:
# Plan ajustado: Plan para 'GraphQL vs REST': 3 secciones, 2 fuentes por sección [Ajuste: Agrega comparación de rendimiento]
# Output: [FORMATTED] Borrador basado en plan: Plan para 'GraphQL vs REST': 3 secciones, 2 f... [Corrección: ok]
# Feedback: [{'stage': 'plan', 'feedback': 'Agrega comparación de rendimiento'}]

Ejercicio 4: Feedback loop con límite y disclaimer (Medio)

Implementa un feedback loop que, después de 3 rondas sin aprobación, entregue automáticamente con un mensaje tipo "Entregado después de 3 revisiones. Contacta soporte si necesitas más ajustes." El estado debe registrar si fue aprobado por el usuario o auto-entregado.

Ver solución

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class State(TypedDict):
    content: str
    feedback: Annotated[list[str], operator.add]
    iteration: int
    delivery_mode: str

MAX_ROUNDS = 3

def generate(state: State) -> dict:
    iteration = state.get("iteration", 0)
    feedback = state.get("feedback", [])
    content = f"Contenido v{iteration + 1}"
    if feedback:
        content += f" (incorporando: {feedback[-1]})"
    return {"content": content, "iteration": iteration + 1}

def review(state: State) -> dict:
    if state["iteration"] >= MAX_ROUNDS:
        disclaimer = (
            f"Entregado automáticamente después de {MAX_ROUNDS} revisiones. "
            f"Contacta soporte si necesitas más ajustes."
        )
        return {
            "content": f"{state['content']}\n\n⚠️ {disclaimer}",
            "delivery_mode": "auto_delivered",
        }

    response = interrupt({
        "content": state["content"],
        "round": state["iteration"],
        "remaining": MAX_ROUNDS - state["iteration"],
    })

    if response.lower() == "approved":
        return {"delivery_mode": "user_approved"}
    return {"feedback": [response], "delivery_mode": "needs_revision"}

def route(state: State) -> str:
    mode = state.get("delivery_mode", "")
    if mode in ("user_approved", "auto_delivered"):
        return "deliver"
    return "revise"

def deliver(state: State) -> dict:
    return {}

builder = StateGraph(State)
builder.add_node("generate", generate)
builder.add_node("review", review)
builder.add_node("deliver", deliver)

builder.add_edge(START, "generate")
builder.add_edge("generate", "review")
builder.add_conditional_edges("review", route, {"revise": "generate", "deliver": "deliver"})
builder.add_edge("deliver", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "bounded-delivery"}}

graph.invoke(
    {"content": "", "feedback": [], "iteration": 0, "delivery_mode": ""},
    config,
)
graph.invoke(Command(resume="Más corto"), config)
graph.invoke(Command(resume="Más datos"), config)

state = graph.get_state(config)
print(f"Contenido: {state.values['content']}")
print(f"Modo de entrega: {state.values['delivery_mode']}")
print(f"Iteraciones: {state.values['iteration']}")
print(f"Feedback: {state.values['feedback']}")
# Output esperado:
# Contenido: Contenido v3 (incorporando: Más datos)
#
# ⚠️ Entregado automáticamente después de 3 revisiones. Contacta soporte si necesitas más ajustes.
# Modo de entrega: auto_delivered
# Iteraciones: 3
# Feedback: ['Más corto', 'Más datos']

Ejercicio 5: Feedback loop con categorías de respuesta (Avanzado)

Crea un feedback loop donde el usuario no solo escribe texto libre, sino que puede elegir entre categorías: "tone" (ajustar tono), "content" (agregar/quitar contenido), "format" (cambiar estructura), "approve". El nodo de generación debe comportarse diferente según la categoría del feedback.

Ver solución

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class CategorizedState(TypedDict):
    topic: str
    output: str
    feedback_log: Annotated[list[dict], operator.add]
    iteration: int
    approved: bool

def generate(state: CategorizedState) -> dict:
    iteration = state.get("iteration", 0)
    log = state.get("feedback_log", [])

    base = f"Artículo sobre {state['topic']}"

    tone_adjustments = [e["detail"] for e in log if e["category"] == "tone"]
    content_adjustments = [e["detail"] for e in log if e["category"] == "content"]
    format_adjustments = [e["detail"] for e in log if e["category"] == "format"]

    parts = [base]
    if tone_adjustments:
        parts.append(f"[Tono ajustado: {', '.join(tone_adjustments)}]")
    if content_adjustments:
        parts.append(f"[Contenido: {', '.join(content_adjustments)}]")
    if format_adjustments:
        parts.append(f"[Formato: {', '.join(format_adjustments)}]")

    return {"output": " ".join(parts), "iteration": iteration + 1}

def review(state: CategorizedState) -> dict:
    if state["iteration"] >= 4:
        return {"approved": True}

    response = interrupt({
        "output": state["output"],
        "round": state["iteration"],
        "instruction": (
            "Responde con formato 'categoría: detalle'. "
            "Categorías: tone, content, format, approve. "
            "Ejemplo: 'tone: más casual' o 'content: agrega benchmarks' o 'approve'"
        ),
    })

    if response.lower().strip() == "approve":
        return {"approved": True}

    if ":" in response:
        category, detail = response.split(":", 1)
        category = category.strip().lower()
        detail = detail.strip()
    else:
        category = "content"
        detail = response

    valid_categories = ("tone", "content", "format")
    if category not in valid_categories:
        category = "content"

    return {
        "feedback_log": [{"category": category, "detail": detail, "round": state["iteration"]}],
        "approved": False,
    }

def route(state: CategorizedState) -> str:
    if state.get("approved"):
        return "deliver"
    return "revise"

def deliver(state: CategorizedState) -> dict:
    return {}

builder = StateGraph(CategorizedState)
builder.add_node("generate", generate)
builder.add_node("review", review)
builder.add_node("deliver", deliver)

builder.add_edge(START, "generate")
builder.add_edge("generate", "review")
builder.add_conditional_edges("review", route, {"revise": "generate", "deliver": "deliver"})
builder.add_edge("deliver", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "categorized-feedback"}}

graph.invoke(
    {"topic": "MLOps", "output": "", "feedback_log": [], "iteration": 0, "approved": False},
    config,
)
state = graph.get_state(config)
print(f"v1: {state.values['output']}")

graph.invoke(Command(resume="tone: más técnico y directo"), config)
state = graph.get_state(config)
print(f"v2: {state.values['output']}")

graph.invoke(Command(resume="content: agrega sección de CI/CD"), config)
state = graph.get_state(config)
print(f"v3: {state.values['output']}")

graph.invoke(Command(resume="approve"), config)
state = graph.get_state(config)
print(f"Final: {state.values['output']}")
print(f"Log: {state.values['feedback_log']}")
# Output esperado:
# v1: Artículo sobre MLOps
# v2: Artículo sobre MLOps [Tono ajustado: más técnico y directo]
# v3: Artículo sobre MLOps [Tono ajustado: más técnico y directo] [Contenido: agrega sección de CI/CD]
# Final: Artículo sobre MLOps [Tono ajustado: más técnico y directo] [Contenido: agrega sección de CI/CD]
# Log: [{'category': 'tone', 'detail': 'más técnico y directo', 'round': 1}, {'category': 'content', 'detail': 'agrega sección de CI/CD', 'round': 2}]

Resumen

En esta cápsula aprendiste:

Feedback loops transforman HITL de binario a iterativo — el patrón Review → Feedback → Revise → Approve produce resultados que ni el humano ni la IA lograrían solos
El feedback se acumula, no se reemplaza — usa Annotated[list[str], operator.add] para mantener el historial completo de correcciones en cada ronda
Incorpora feedback al prompt, no reemplaces desde cero — "Versión anterior + feedback del usuario → mejora lo que hay" produce evolución natural del contenido
Multi-point feedback da control granular — revisar el plan antes de ejecutar, los datos antes de sintetizar, y el reporte antes de entregar evita trabajo desperdiciado
Máximo 3 rondas de feedback — después, los retornos son marginales o contraproducentes. Auto-entrega con disclaimer es mejor que un loop infinito
La UX del interrupt importa — pasa un dict con el borrador, la ronda actual, las rondas restantes, y una instrucción clara. El usuario debe saber exactamente qué está revisando y qué opciones tiene

Próxima cápsula: Patterns HITL en Producción — cómo implementar aprobaciones asíncronas, timeouts, escalaciones y dashboards cuando "el humano" no está sentado frente a un terminal.

Recursos adicionales

LangGraph Human-in-the-Loop — Conceptos de HITL en LangGraph
How to wait for user input — Implementación de interrupts para input de usuario
How to edit graph state — Editar el estado del grafo durante la ejecución
How to review tool calls — Patrones de review para tool calls
LangGraph interrupt() — Referencia de la función interrupt
Human-AI Collaboration Patterns — Nielsen Norman Group: patrones de colaboración humano-IA

Módulo 9 — LangChain & LangGraph: From Chains to Agents

← AnteriorEditable State: Modificar Estado Durante Pausa Siguiente →Patterns HITL en Producción