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

Editable State: Modificar Estado Durante Pausa

Siguiente →

Descripción de la cápsula

El agente está pausado. Ya sabes aprobarlo o rechazarlo. Pero hay una tercera opción que cambia todo: modificar el estado del agente antes de que continúe. No reiniciar. No re-ejecutar. Abrir el estado, corregir lo que está mal, y dejar que el agente siga con información correcta.

Imagina esto: tu agente investigó una empresa y encontró que el CEO es "John Smith." Tú sabes que el CEO cambió a "Jane Doe" el mes pasado. Sin editable state, tienes que reiniciar toda la investigación. Con editable state, corriges ese dato en el estado y el agente continúa — genera el reporte con "Jane Doe" sin repetir las búsquedas que ya hizo.

Esto no es un feature avanzado ni un caso borde. Es el escenario más práctico de HITL: el agente hizo trabajo útil, pero tiene un dato incorrecto. Corregir y continuar es más eficiente que empezar de cero.

El escenario que lo hace imprescindible

Sin editable state:
  1. Agente investiga empresa (3 minutos, $0.50 en API calls)
  2. Agente encuentra CEO = "John Smith" ← incorrecto
  3. Tú detectas el error
  4. Opción: reiniciar toda la investigación
  5. 3 minutos más, $0.50 más
  6. Total: 6 minutos, $1.00

Con editable state:
  1. Agente investiga empresa (3 minutos, $0.50 en API calls)
  2. Agente encuentra CEO = "John Smith" ← incorrecto
  3. Tú detectas el error
  4. Editas: CEO = "Jane Doe"
  5. Agente continúa desde donde se pausó
  6. Total: 3 minutos + segundos, $0.50

La diferencia no es solo eficiencia. Es que preservas todo el trabajo que el agente ya hizo: las búsquedas, el análisis, los datos correctos. Solo corriges lo incorrecto.

Inspeccionar el estado durante una pausa

Antes de editar, necesitas ver qué tiene el agente. Usas graph.get_state(config):

from dotenv import load_dotenv
load_dotenv()

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

class ResearchState(TypedDict):
    company: str
    ceo: str
    revenue: str
    employees: int
    findings: Annotated[list[str], operator.add]
    report: str

def research_company(state: ResearchState) -> dict:
    return {
        "ceo": "John Smith",
        "revenue": "$5.2B (FY2024)",
        "employees": 12000,
        "findings": [
            "Líder en el segmento enterprise AI",
            "Crecimiento 40% interanual",
            "Expansión a mercado europeo en Q2",
        ],
    }

def generate_report(state: ResearchState) -> dict:
    report = f"REPORTE: {state['company']}\n"
    report += f"  CEO: {state['ceo']}\n"
    report += f"  Revenue: {state['revenue']}\n"
    report += f"  Empleados: {state['employees']}\n"
    report += f"  Hallazgos: {len(state['findings'])}\n"
    for f in state["findings"]:
        report += f"    - {f}\n"
    return {"report": report}

graph_builder = StateGraph(ResearchState)
graph_builder.add_node("research", research_company)
graph_builder.add_node("report", generate_report)

graph_builder.add_edge(START, "research")
graph_builder.add_edge("research", "report")
graph_builder.add_edge("report", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["research"],
)

config = {"configurable": {"thread_id": "inspect-demo"}}

graph.invoke(
    {"company": "TechCorp AI", "ceo": "", "revenue": "", "employees": 0,
     "findings": [], "report": ""},
    config,
)

print("=== Inspeccionar estado durante la pausa ===\n")

state = graph.get_state(config)

print(f"Estado completo:")
for key, value in state.values.items():
    if isinstance(value, list):
        print(f"  {key}: [{len(value)} items]")
        for item in value:
            print(f"    - {item}")
    else:
        print(f"  {key}: {value}")

print(f"\nSiguiente nodo: {state.next}")
print(f"Checkpoint ID: {state.config['configurable']['checkpoint_id'][:20]}...")
# Output esperado:
# === Inspeccionar estado durante la pausa ===
#
# Estado completo:
#   company: TechCorp AI
#   ceo: John Smith
#   revenue: $5.2B (FY2024)
#   employees: 12000
#   findings: [3 items]
#     - Líder en el segmento enterprise AI
#     - Crecimiento 40% interanual
#     - Expansión a mercado europeo en Q2
#   report:
#
# Siguiente nodo: ('report',)
# Checkpoint ID: 1ef8a1b2c3d4e5f6ab...

state.values te da el estado completo. state.next te dice qué nodo está pendiente. Con esto puedes decidir qué corregir.

Modificar el estado: update_state

Viste que el CEO es "John Smith" pero debería ser "Jane Doe." Corriges con graph.update_state():

from dotenv import load_dotenv
load_dotenv()

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

class ResearchState(TypedDict):
    company: str
    ceo: str
    revenue: str
    employees: int
    findings: Annotated[list[str], operator.add]
    report: str

def research_company(state: ResearchState) -> dict:
    return {
        "ceo": "John Smith",
        "revenue": "$5.2B (FY2024)",
        "employees": 12000,
        "findings": [
            "Líder en el segmento enterprise AI",
            "Crecimiento 40% interanual",
        ],
    }

def generate_report(state: ResearchState) -> dict:
    report = f"REPORTE: {state['company']}\n"
    report += f"  CEO: {state['ceo']}\n"
    report += f"  Revenue: {state['revenue']}\n"
    report += f"  Empleados: {state['employees']}\n"
    report += f"  Hallazgos ({len(state['findings'])}):\n"
    for f in state["findings"]:
        report += f"    - {f}\n"
    return {"report": report}

graph_builder = StateGraph(ResearchState)
graph_builder.add_node("research", research_company)
graph_builder.add_node("report", generate_report)

graph_builder.add_edge(START, "research")
graph_builder.add_edge("research", "report")
graph_builder.add_edge("report", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["research"],
)

config = {"configurable": {"thread_id": "edit-demo"}}

graph.invoke(
    {"company": "TechCorp AI", "ceo": "", "revenue": "", "employees": 0,
     "findings": [], "report": ""},
    config,
)

state = graph.get_state(config)
print(f"ANTES de editar:")
print(f"  CEO: {state.values['ceo']}")
print(f"  Findings: {len(state.values['findings'])}")

print("\n--- Editando estado ---\n")
graph.update_state(config, {"ceo": "Jane Doe"})

state = graph.get_state(config)
print(f"DESPUÉS de editar:")
print(f"  CEO: {state.values['ceo']}")
print(f"  Findings: {len(state.values['findings'])}  ← no se modificaron")

print("\n--- Continuando ejecución ---\n")
result = graph.invoke(None, config)

final = graph.get_state(config)
print(final.values["report"])
# Output esperado:
# ANTES de editar:
#   CEO: John Smith
#   Findings: 2
#
# --- Editando estado ---
#
# DESPUÉS de editar:
#   CEO: Jane Doe
#   Findings: 2  ← no se modificaron
#
# --- Continuando ejecución ---
#
# REPORTE: TechCorp AI
#   CEO: Jane Doe
#   Revenue: $5.2B (FY2024)
#   Empleados: 12000
#   Hallazgos (2):
#     - Líder en el segmento enterprise AI
#     - Crecimiento 40% interanual

Tres observaciones importantes:

  1. update_state(config, {"ceo": "Jane Doe"}) solo modifica el campo ceo. Los demás campos no se tocan
  2. Los findings siguen siendo 2 — el trabajo previo del agente se preserva
  3. El reporte final dice "Jane Doe" — el agente usó el dato corregido

Cómo funciona update_state con reducers

update_state sigue las mismas reglas de reducers que el grafo normal:

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    name: str
    count: int
    tags: Annotated[list[str], operator.add]

def process(state: State) -> dict:
    return {
        "name": "original",
        "count": 10,
        "tags": ["tag1", "tag2"],
    }

def finalize(state: State) -> dict:
    return {"name": f"final_{state['name']}_count_{state['count']}_tags_{len(state['tags'])}"}

graph_builder = StateGraph(State)
graph_builder.add_node("process", process)
graph_builder.add_node("finalize", finalize)

graph_builder.add_edge(START, "process")
graph_builder.add_edge("process", "finalize")
graph_builder.add_edge("finalize", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["process"],
)

config = {"configurable": {"thread_id": "reducer-demo"}}

graph.invoke({"name": "", "count": 0, "tags": []}, config)

state = graph.get_state(config)
print(f"Antes: name={state.values['name']}, count={state.values['count']}, tags={state.values['tags']}")

graph.update_state(config, {
    "name": "editado",
    "count": 99,
    "tags": ["nuevo_tag"],
})

state = graph.get_state(config)
print(f"Después: name={state.values['name']}, count={state.values['count']}, tags={state.values['tags']}")
# Output esperado:
# Antes: name=original, count=10, tags=['tag1', 'tag2']
# Después: name=editado, count=99, tags=['tag1', 'tag2', 'nuevo_tag']

Observa la diferencia:

  • ✅ name (sin reducer): se reemplaza — de "original" a "editado"
  • ✅ count (sin reducer): se reemplaza — de 10 a 99
  • ✅ tags (con operator.add): se agrega — "nuevo_tag" se añade a la lista existente

El update_state respeta los reducers del TypedDict. Si un campo usa operator.add, la actualización se añade, no reemplaza. Si no tiene reducer, se reemplaza directamente.

Caso práctico: corregir datos incorrectos del agente

El escenario completo del CEO, con un flujo HITL realista:

from dotenv import load_dotenv
load_dotenv()

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

class CompanyResearch(TypedDict):
    company: str
    ceo: str
    founded: int
    hq_location: str
    key_products: Annotated[list[str], operator.add]
    market_position: str
    report: str

def research(state: CompanyResearch) -> dict:
    return {
        "ceo": "John Smith",
        "founded": 2018,
        "hq_location": "San Francisco, CA",
        "key_products": ["Enterprise AI Platform", "AI Analytics Suite", "MLOps Toolkit"],
        "market_position": "Top 5 en enterprise AI, compitiendo con OpenAI y Anthropic",
    }

def human_review(state: CompanyResearch) -> dict:
    review = interrupt({
        "message": "Revisa los datos de investigación antes de generar el reporte",
        "data": {
            "company": state["company"],
            "ceo": state["ceo"],
            "founded": state["founded"],
            "hq_location": state["hq_location"],
            "key_products": state["key_products"],
            "market_position": state["market_position"],
        },
        "instructions": "Responde {approved: true} o {corrections: {campo: valor}}",
    })

    if review.get("approved"):
        return {}
    return {}

def generate_report(state: CompanyResearch) -> dict:
    report = f"{'='*50}\n"
    report += f"COMPANY RESEARCH REPORT: {state['company']}\n"
    report += f"{'='*50}\n"
    report += f"CEO: {state['ceo']}\n"
    report += f"Founded: {state['founded']}\n"
    report += f"HQ: {state['hq_location']}\n"
    report += f"Products: {', '.join(state['key_products'])}\n"
    report += f"Market Position: {state['market_position']}\n"
    report += f"{'='*50}"
    return {"report": report}

graph_builder = StateGraph(CompanyResearch)
graph_builder.add_node("research", research)
graph_builder.add_node("human_review", human_review)
graph_builder.add_node("generate_report", generate_report)

graph_builder.add_edge(START, "research")
graph_builder.add_edge("research", "human_review")
graph_builder.add_edge("human_review", "generate_report")
graph_builder.add_edge("generate_report", END)

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

config = {"configurable": {"thread_id": "company-research"}}

print("=== PASO 1: Agente investiga ===\n")
graph.invoke(
    {"company": "TechCorp AI", "ceo": "", "founded": 0, "hq_location": "",
     "key_products": [], "market_position": "", "report": ""},
    config,
)

state = graph.get_state(config)
print(f"Agente encontró: CEO = {state.values['ceo']}")
print(f"Esperando en: {state.next}")

print("\n=== PASO 2: Humano detecta errores y corrige ===\n")

print("Correcciones:")
print("  - CEO: 'John Smith' → 'Jane Doe' (cambió el mes pasado)")
print("  - HQ: agregar 'oficina europea en Londres'\n")

graph.update_state(config, {
    "ceo": "Jane Doe",
    "hq_location": "San Francisco, CA (HQ) + Londres, UK (oficina europea)",
})

state = graph.get_state(config)
print(f"Estado corregido:")
print(f"  CEO: {state.values['ceo']}")
print(f"  HQ: {state.values['hq_location']}")

print("\n=== PASO 3: Continuar con datos corregidos ===\n")
result = graph.invoke(Command(resume={"approved": True}), config)

final = graph.get_state(config)
print(final.values["report"])
# Output esperado:
# === PASO 1: Agente investiga ===
#
# Agente encontró: CEO = John Smith
# Esperando en: ('human_review',)
#
# === PASO 2: Humano detecta errores y corrige ===
#
# Correcciones:
#   - CEO: 'John Smith' → 'Jane Doe' (cambió el mes pasado)
#   - HQ: agregar 'oficina europea en Londres'
#
# Estado corregido:
#   CEO: Jane Doe
#   HQ: San Francisco, CA (HQ) + Londres, UK (oficina europea)
#
# === PASO 3: Continuar con datos corregidos ===
#
# ==================================================
# COMPANY RESEARCH REPORT: TechCorp AI
# ==================================================
# CEO: Jane Doe
# Founded: 2018
# HQ: San Francisco, CA (HQ) + Londres, UK (oficina europea)
# Products: Enterprise AI Platform, AI Analytics Suite, MLOps Toolkit
# Market Position: Top 5 en enterprise AI, compitiendo con OpenAI y Anthropic
# ==================================================

El reporte final tiene los datos corregidos por el humano, pero preserva todo lo demás que el agente investigó correctamente (productos, market position, año de fundación).

Agregar información que el agente no encontró

No solo corriges errores — puedes agregar datos que el agente no pudo obtener:

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    topic: str
    sources: Annotated[list[str], operator.add]
    analysis: str
    summary: str

def search(state: State) -> dict:
    return {
        "sources": [
            "Wikipedia: información general sobre el tema",
            "arXiv: 2 papers relevantes encontrados",
        ],
    }

def analyze(state: State) -> dict:
    return {"analysis": f"Análisis de {len(state['sources'])} fuentes sobre '{state['topic']}'"}

def summarize(state: State) -> dict:
    return {"summary": f"Resumen: {state['analysis']}. Fuentes: {', '.join(state['sources'][:3])}..."}

graph_builder = StateGraph(State)
graph_builder.add_node("search", search)
graph_builder.add_node("analyze", analyze)
graph_builder.add_node("summarize", summarize)

graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "analyze")
graph_builder.add_edge("analyze", "summarize")
graph_builder.add_edge("summarize", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["search"],
)

config = {"configurable": {"thread_id": "add-info"}}

graph.invoke(
    {"topic": "AI regulation in EU", "sources": [], "analysis": "", "summary": ""},
    config,
)

state = graph.get_state(config)
print(f"Fuentes encontradas por el agente: {len(state.values['sources'])}")
for s in state.values["sources"]:
    print(f"  - {s}")

print("\n--- Humano agrega fuentes que el agente no encontró ---\n")

graph.update_state(config, {
    "sources": [
        "INSIDER: Documento interno de la Comisión Europea (no público)",
        "INSIDER: Entrevista con regulador principal (off the record)",
    ],
})

state = graph.get_state(config)
print(f"Fuentes después de agregar: {len(state.values['sources'])}")
for s in state.values["sources"]:
    print(f"  - {s}")

print("\n--- Continuando con fuentes enriquecidas ---\n")
graph.invoke(None, config)

final = graph.get_state(config)
print(f"Análisis: {final.values['analysis']}")
print(f"Resumen: {final.values['summary']}")
# Output esperado:
# Fuentes encontradas por el agente: 2
#   - Wikipedia: información general sobre el tema
#   - arXiv: 2 papers relevantes encontrados
#
# --- Humano agrega fuentes que el agente no encontró ---
#
# Fuentes después de agregar: 4
#   - Wikipedia: información general sobre el tema
#   - arXiv: 2 papers relevantes encontrados
#   - INSIDER: Documento interno de la Comisión Europea (no público)
#   - INSIDER: Entrevista con regulador principal (off the record)
#
# --- Continuando con fuentes enriquecidas ---
#
# Análisis: Análisis de 4 fuentes sobre 'AI regulation in EU'
# Resumen: Resumen: Análisis de 4 fuentes sobre 'AI regulation in EU'. Fuentes: Wikipedia: información general sobre el tema, arXiv: 2 papers relevantes encontrados, INSIDER: Documento interno de la Comisión Europea (no público)...

Como sources usa operator.add, las nuevas fuentes se agregan a las existentes. El agente ahora trabaja con 4 fuentes: 2 que encontró él y 2 que el humano aportó. Información privilegiada que ningún agente podría obtener por su cuenta.

update_state con as_node: controlar qué nodo viene después

Cuando haces update_state, puedes especificar as_node para que LangGraph trate la actualización como si viniera de un nodo específico. Esto afecta qué nodo se ejecuta después:

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    data: str
    step_a_result: str
    step_b_result: str
    final: str
    log: Annotated[list[str], operator.add]

def step_a(state: State) -> dict:
    return {"step_a_result": f"A procesó: {state['data']}", "log": ["step_a"]}

def step_b(state: State) -> dict:
    return {"step_b_result": f"B analizó: {state['step_a_result']}", "log": ["step_b"]}

def step_final(state: State) -> dict:
    return {"final": f"Final: {state['step_b_result']}", "log": ["step_final"]}

graph_builder = StateGraph(State)
graph_builder.add_node("step_a", step_a)
graph_builder.add_node("step_b", step_b)
graph_builder.add_node("step_final", step_final)

graph_builder.add_edge(START, "step_a")
graph_builder.add_edge("step_a", "step_b")
graph_builder.add_edge("step_b", "step_final")
graph_builder.add_edge("step_final", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["step_b"],
)

config = {"configurable": {"thread_id": "as-node-demo"}}

graph.invoke(
    {"data": "raw input", "step_a_result": "", "step_b_result": "",
     "final": "", "log": []},
    config,
)

state = graph.get_state(config)
print(f"Pausado antes de: {state.next}")
print(f"step_a_result: {state.values['step_a_result']}")

print("\n--- Opción 1: update_state SIN as_node ---")
print("  La actualización no cambia qué nodo viene después")
print(f"  Siguiente sigue siendo: {state.next}")

print("\n--- Opción 2: update_state CON as_node='step_b' ---")
print("  LangGraph trata el update como si step_b ya ejecutó")

graph.update_state(
    config,
    {"step_b_result": "B resultado manual (humano reemplazó step_b)"},
    as_node="step_b",
)

state = graph.get_state(config)
print(f"  Siguiente ahora es: {state.next}")
print(f"  step_b_result: {state.values['step_b_result']}")

print("\n--- Continuando: step_final ejecuta directamente ---\n")
graph.invoke(None, config)

final = graph.get_state(config)
print(f"Final: {final.values['final']}")
print(f"Log: {final.values['log']}")
# Output esperado:
# Pausado antes de: ('step_b',)
# step_a_result: A procesó: raw input
#
# --- Opción 1: update_state SIN as_node ---
#   La actualización no cambia qué nodo viene después
#   Siguiente sigue siendo: ('step_b',)
#
# --- Opción 2: update_state CON as_node='step_b' ---
#   LangGraph trata el update como si step_b ya ejecutó
#   Siguiente ahora es: ('step_final',)
#   step_b_result: B resultado manual (humano reemplazó step_b)
#
# --- Continuando: step_final ejecuta directamente ---
#
# Final: Final: B resultado manual (humano reemplazó step_b)
# Log: ['step_a', 'step_final']

as_node="step_b" le dice a LangGraph: "esta actualización es como si step_b se ejecutó." Entonces el grafo pasa al siguiente nodo después de step_b, que es step_final. Sin as_node, el grafo seguiría queriendo ejecutar step_b.

Usos prácticos de as_node:

  • ✅ Reemplazar completamente la ejecución de un nodo con un valor manual
  • ✅ Saltarte un nodo que no necesitas (proporcionas su output directamente)
  • ✅ Redirigir el flujo: hacer que el grafo piense que vino de un nodo diferente

Editar mensajes en la conversación

Cuando el estado tiene mensajes (como en chatbots o agentes con MessagesState), puedes editarlos durante una pausa:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage

class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    context: str
    result: str

def gather_context(state: State) -> dict:
    last_msg = state["messages"][-1].content
    context = f"Contexto recopilado para: '{last_msg}'"
    return {"context": context}

def respond(state: State) -> dict:
    response = f"Respuesta basada en contexto: {state['context']}"
    return {
        "messages": [AIMessage(content=response)],
        "result": response,
    }

graph_builder = StateGraph(State)
graph_builder.add_node("gather_context", gather_context)
graph_builder.add_node("respond", respond)

graph_builder.add_edge(START, "gather_context")
graph_builder.add_edge("gather_context", "respond")
graph_builder.add_edge("respond", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["gather_context"],
)

config = {"configurable": {"thread_id": "edit-messages"}}

graph.invoke(
    {"messages": [HumanMessage(content="Investiga sobre LangChain")],
     "context": "", "result": ""},
    config,
)

state = graph.get_state(config)
print(f"Contexto actual: {state.values['context']}")
print(f"Mensajes: {len(state.values['messages'])}")

print("\n--- Humano agrega mensaje con instrucciones adicionales ---\n")

graph.update_state(config, {
    "messages": [HumanMessage(content="NOTA: enfócate en LangGraph, no LangChain general")],
    "context": "Contexto recopilado para: 'Investiga sobre LangChain'. NOTA HUMANA: enfoque en LangGraph",
})

state = graph.get_state(config)
print(f"Contexto actualizado: {state.values['context']}")
print(f"Mensajes: {len(state.values['messages'])}")
for msg in state.values["messages"]:
    print(f"  [{msg.type}] {msg.content[:60]}...")

print("\n--- Continuando con contexto enriquecido ---\n")
graph.invoke(None, config)

final = graph.get_state(config)
print(f"Resultado: {final.values['result']}")
# Output esperado:
# Contexto actual: Contexto recopilado para: 'Investiga sobre LangChain'
# Mensajes: 1
#
# --- Humano agrega mensaje con instrucciones adicionales ---
#
# Contexto actualizado: Contexto recopilado para: 'Investiga sobre LangChain'. NOTA HUMANA: enfoque en LangGraph
# Mensajes: 2
# Mensajes:
#   [human] Investiga sobre LangChain...
#   [human] NOTA: enfócate en LangGraph, no LangChain general...
#
# --- Continuando con contexto enriquecido ---
#
# Resultado: Respuesta basada en contexto: Contexto recopilado para: 'Investiga sobre LangChain'. NOTA HUMANA: enfoque en LangGraph

Como messages usa operator.add, el nuevo mensaje se agrega a la lista existente. El agente ahora tiene tanto la instrucción original como la corrección del humano.

Comparación: reiniciar vs editar estado

Para que quede claro cuándo conviene cada enfoque:

AspectoReiniciar desde ceroEditar estado y continuar
Trabajo previoSe pierde todoSe preserva
Costo de APISe repite todoSolo desde la pausa
TiempoMinutos (re-ejecutar pipeline)Segundos (editar + continuar)
Datos correctos del agenteSe re-calculanSe mantienen
Cuándo usarError fundamental en el diseñoDato incorrecto puntual
ComplejidadSimple (nuevo invoke)Requiere entender el estado

Regla práctica:

  • ✅ Edita cuando el agente hizo bien el 80%+ del trabajo y solo necesitas corregir un dato
  • ✅ Reinicia cuando el error es tan grande que invalida todo el trabajo previo
  • ⚠️ Edita con cuidado cuando los campos tienen dependencias entre sí (si cambias X, ¿Y sigue siendo válido?)

Validar ediciones antes de continuar

En un flujo de producción, quieres verificar que la edición es válida antes de continuar:

from dotenv import load_dotenv
load_dotenv()

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

class OrderState(TypedDict):
    customer: str
    items: Annotated[list[str], operator.add]
    total: float
    discount_pct: float
    final_total: float
    status: str

def calculate_order(state: OrderState) -> dict:
    total = len(state["items"]) * 29.99
    return {"total": total, "status": "calculated"}

def apply_discount(state: OrderState) -> dict:
    final = state["total"] * (1 - state["discount_pct"] / 100)
    return {"final_total": round(final, 2), "status": "discount_applied"}

def process_order(state: OrderState) -> dict:
    return {"status": f"processed_for_{state['customer']}_total_{state['final_total']}"}

graph_builder = StateGraph(OrderState)
graph_builder.add_node("calculate", calculate_order)
graph_builder.add_node("discount", apply_discount)
graph_builder.add_node("process", process_order)

graph_builder.add_edge(START, "calculate")
graph_builder.add_edge("calculate", "discount")
graph_builder.add_edge("discount", "process")
graph_builder.add_edge("process", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["calculate"],
)

config = {"configurable": {"thread_id": "validate-edit"}}

graph.invoke(
    {"customer": "Empresa ABC", "items": ["Licencia AI", "Soporte Premium", "Training"],
     "total": 0, "discount_pct": 10, "final_total": 0, "status": ""},
    config,
)

state = graph.get_state(config)
print(f"Total calculado: ${state.values['total']:.2f}")
print(f"Descuento configurado: {state.values['discount_pct']}%")

print("\n--- Humano quiere cambiar el descuento a 25% ---\n")

new_discount = 25.0

if not (0 <= new_discount <= 50):
    print(f"❌ Descuento {new_discount}% fuera de rango permitido (0-50%)")
else:
    print(f"✅ Descuento {new_discount}% válido, aplicando...")
    graph.update_state(config, {"discount_pct": new_discount})

    state = graph.get_state(config)
    print(f"  Descuento actualizado: {state.values['discount_pct']}%")

    print("\n--- Continuando con descuento modificado ---\n")
    graph.invoke(None, config)

    final = graph.get_state(config)
    print(f"Total original: ${final.values['total']:.2f}")
    print(f"Descuento: {final.values['discount_pct']}%")
    print(f"Total final: ${final.values['final_total']:.2f}")
    print(f"Status: {final.values['status']}")
# Output esperado:
# Total calculado: $89.97
# Descuento configurado: 10%
#
# --- Humano quiere cambiar el descuento a 25% ---
#
# ✅ Descuento 25.0% válido, aplicando...
#   Descuento actualizado: 25.0%
#
# --- Continuando con descuento modificado ---
#
# Total original: $89.97
# Descuento: 25.0%
# Total final: $67.48
# Status: processed_for_Empresa ABC_total_67.48

La validación antes de aplicar la edición evita estados inválidos. En producción, podrías validar tipos, rangos, relaciones entre campos, y cualquier restricción de negocio.

Remover datos irrelevantes antes de continuar

Otro caso práctico: el agente recopiló datos pero algunos son irrelevantes. En vez de dejar que el agente los procese, los remueves:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    query: str
    results: list[dict]
    filtered_results: list[dict]
    summary: str

def search(state: State) -> dict:
    return {"results": [
        {"source": "arXiv", "data": "Paper relevante sobre AI agents", "score": 0.95},
        {"source": "Wikipedia", "data": "Artículo general sobre AI", "score": 0.80},
        {"source": "Reddit", "data": "Meme sobre robots", "score": 0.15},
        {"source": "Blog spam", "data": "Compra mi curso de AI!!!", "score": 0.05},
        {"source": "IEEE", "data": "Estudio técnico sobre LLM safety", "score": 0.88},
    ]}

def process(state: State) -> dict:
    results = state.get("filtered_results") or state["results"]
    sources = [r["source"] for r in results]
    return {"summary": f"Análisis de {len(results)} fuentes: {', '.join(sources)}"}

graph_builder = StateGraph(State)
graph_builder.add_node("search", search)
graph_builder.add_node("process", process)

graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "process")
graph_builder.add_edge("process", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["search"],
)

config = {"configurable": {"thread_id": "filter-results"}}

graph.invoke(
    {"query": "AI agent safety", "results": [], "filtered_results": [], "summary": ""},
    config,
)

state = graph.get_state(config)
print("Resultados del agente:")
for r in state.values["results"]:
    flag = "⚠️" if r["score"] < 0.5 else "✅"
    print(f"  {flag} {r['source']} (score: {r['score']}): {r['data'][:40]}")

print("\n--- Humano filtra resultados irrelevantes ---\n")

good_results = [r for r in state.values["results"] if r["score"] >= 0.5]
print(f"Manteniendo {len(good_results)} de {len(state.values['results'])} resultados:")
for r in good_results:
    print(f"  ✅ {r['source']}")

graph.update_state(config, {"filtered_results": good_results})

print("\n--- Continuando con resultados filtrados ---\n")
graph.invoke(None, config)

final = graph.get_state(config)
print(f"Summary: {final.values['summary']}")
# Output esperado:
# Resultados del agente:
#   ✅ arXiv (score: 0.95): Paper relevante sobre AI agents
#   ✅ Wikipedia (score: 0.80): Artículo general sobre AI
#   ⚠️ Reddit (score: 0.15): Meme sobre robots
#   ⚠️ Blog spam (score: 0.05): Compra mi curso de AI!!!
#   ✅ IEEE (score: 0.88): Estudio técnico sobre LLM safety
#
# --- Humano filtra resultados irrelevantes ---
#
# Manteniendo 3 de 5 resultados:
#   ✅ arXiv
#   ✅ Wikipedia
#   ✅ IEEE
#
# --- Continuando con resultados filtrados ---
#
# Summary: Análisis de 3 fuentes: arXiv, Wikipedia, IEEE

El humano eliminó Reddit y Blog spam antes de que el agente los procesara. El resumen final solo incluye fuentes de calidad.

Cambiar la dirección de la investigación

El caso más poderoso: el agente está investigando un tema, pero tú quieres redirigirlo a mitad de camino:

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    original_query: str
    active_query: str
    research_data: Annotated[list[str], operator.add]
    analysis: str
    report: str

def initial_research(state: State) -> dict:
    query = state["active_query"] or state["original_query"]
    return {
        "research_data": [
            f"Dato 1 sobre '{query}': panorama general",
            f"Dato 2 sobre '{query}': tendencias actuales",
        ],
    }

def deep_analysis(state: State) -> dict:
    query = state["active_query"] or state["original_query"]
    data_summary = f"{len(state['research_data'])} datos sobre '{query}'"
    return {"analysis": f"Análisis profundo: {data_summary}"}

def final_report(state: State) -> dict:
    return {"report": f"REPORTE: {state['analysis']}"}

graph_builder = StateGraph(State)
graph_builder.add_node("initial_research", initial_research)
graph_builder.add_node("deep_analysis", deep_analysis)
graph_builder.add_node("final_report", final_report)

graph_builder.add_edge(START, "initial_research")
graph_builder.add_edge("initial_research", "deep_analysis")
graph_builder.add_edge("deep_analysis", "final_report")
graph_builder.add_edge("final_report", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["initial_research"],
)

config = {"configurable": {"thread_id": "redirect-research"}}

graph.invoke(
    {"original_query": "AI market overview", "active_query": "",
     "research_data": [], "analysis": "", "report": ""},
    config,
)

state = graph.get_state(config)
print(f"Query original: {state.values['original_query']}")
print(f"Datos recopilados: {len(state.values['research_data'])}")
for d in state.values["research_data"]:
    print(f"  - {d}")

print("\n--- Humano redirige: 'enfócate en AI safety, no market overview' ---\n")

graph.update_state(config, {
    "active_query": "AI safety regulations",
    "research_data": ["HUMANO: Redirigido de 'AI market overview' a 'AI safety regulations'"],
})

state = graph.get_state(config)
print(f"Query activo ahora: {state.values['active_query']}")
print(f"Datos (preservados + nota de redirección): {len(state.values['research_data'])}")

print("\n--- Continuando con nueva dirección ---\n")
graph.invoke(None, config)

final = graph.get_state(config)
print(f"Reporte: {final.values['report']}")
# Output esperado:
# Query original: AI market overview
# Datos recopilados: 2
#   - Dato 1 sobre 'AI market overview': panorama general
#   - Dato 2 sobre 'AI market overview': tendencias actuales
#
# --- Humano redirige: 'enfócate en AI safety, no market overview' ---
#
# Query activo ahora: AI safety regulations
# Datos (preservados + nota de redirección): 3
#
# --- Continuando con nueva dirección ---
#
# Reporte: REPORTE: Análisis profundo: 3 datos sobre 'AI safety regulations'

El humano cambió la dirección de la investigación sin perder los datos ya recopilados. El análisis profundo y el reporte se generan con el nuevo enfoque.

Troubleshooting

Problema 1: "update_state no cambia el valor"

Síntoma: Llamas a update_state pero el campo sigue con el valor original.

Causa: El campo usa un reducer (operator.add). El nuevo valor se agrega, no reemplaza.

Solución: Para listas con operator.add, el valor se añade. Si quieres reemplazar la lista completa, necesitas un campo sin reducer, o diseñar tu estado con un campo separado:

# Si tags usa operator.add:
graph.update_state(config, {"tags": ["nuevo"]})
# Resultado: tags = ["existente", "nuevo"]  (se agrega)

# Para reemplazar, usa un campo sin reducer:
class State(TypedDict):
    filtered_tags: list[str]  # sin operator.add — se reemplaza

Problema 2: "Después de update_state, el nodo incorrecto se ejecuta"

Síntoma: Editas el estado pero el grafo ejecuta un nodo que no esperabas.

Causa: update_state sin as_node no cambia qué nodo viene después.

Solución: Usa as_node para indicar desde qué nodo debe continuar:

graph.update_state(config, {"data": "nuevo"}, as_node="step_b")

Problema 3: "El estado editado no persiste después de continuar"

Síntoma: Editas un campo, continúas, pero el nodo siguiente sobrescribe tu edición.

Causa: El nodo siguiente escribe en el mismo campo que editaste.

Solución: Verifica qué campos escribe cada nodo. Si editas ceo y el siguiente nodo también escribe ceo, tu edición se perderá. Coloca el breakpoint después del nodo que escribe en ese campo.

Problema 4: "update_state falla con error de tipo"

Síntoma: Error al hacer update_state con un valor de tipo incorrecto.

Causa: El valor no coincide con el tipo definido en el TypedDict.

Solución: Respeta los tipos del estado:

class State(TypedDict):
    count: int
    name: str

# ❌ Tipo incorrecto
graph.update_state(config, {"count": "diez"})

# ✅ Tipo correcto
graph.update_state(config, {"count": 10})

Ejercicios

Ejercicio 1: Inspeccionar y editar un campo simple (Fácil)

Crea un grafo de 2 nodos: fetch_data (obtiene el nombre de un producto como "ProdcutX" — con typo intencional) y create_listing (genera un listing). Pausa después de fetch_data, inspecciona el typo, corrígelo a "ProductX", y continúa.

Ver solución 
from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    product_id: str
    product_name: str
    listing: str

def fetch_data(state: State) -> dict:
    return {"product_name": "ProdcutX"}

def create_listing(state: State) -> dict:
    return {"listing": f"LISTING: {state['product_name']} — Disponible ahora"}

graph_builder = StateGraph(State)
graph_builder.add_node("fetch_data", fetch_data)
graph_builder.add_node("create_listing", create_listing)

graph_builder.add_edge(START, "fetch_data")
graph_builder.add_edge("fetch_data", "create_listing")
graph_builder.add_edge("create_listing", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["fetch_data"],
)

config = {"configurable": {"thread_id": "typo-fix"}}
graph.invoke({"product_id": "P001", "product_name": "", "listing": ""}, config)

state = graph.get_state(config)
print(f"Nombre obtenido: '{state.values['product_name']}' ← typo!")

graph.update_state(config, {"product_name": "ProductX"})

state = graph.get_state(config)
print(f"Nombre corregido: '{state.values['product_name']}'")

graph.invoke(None, config)

final = graph.get_state(config)
print(f"Listing: {final.values['listing']}")
# Output esperado:
# Nombre obtenido: 'ProdcutX' ← typo!
# Nombre corregido: 'ProductX'
# Listing: LISTING: ProductX — Disponible ahora

Ejercicio 2: Agregar datos a una lista con reducer (Fácil)

Crea un grafo donde un nodo collect_sources recopila 2 fuentes. Pausa, agrega 2 fuentes más como humano (usando update_state con una lista — recuerda que operator.add las agrega). Verifica que el nodo summarize trabaja con las 4 fuentes.

Ver solución 
from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    topic: str
    sources: Annotated[list[str], operator.add]
    summary: str

def collect_sources(state: State) -> dict:
    return {"sources": [
        f"Fuente 1: Wikipedia sobre {state['topic']}",
        f"Fuente 2: arXiv sobre {state['topic']}",
    ]}

def summarize(state: State) -> dict:
    return {"summary": f"Resumen basado en {len(state['sources'])} fuentes"}

graph_builder = StateGraph(State)
graph_builder.add_node("collect", collect_sources)
graph_builder.add_node("summarize", summarize)

graph_builder.add_edge(START, "collect")
graph_builder.add_edge("collect", "summarize")
graph_builder.add_edge("summarize", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["collect"],
)

config = {"configurable": {"thread_id": "add-sources"}}
graph.invoke({"topic": "HITL patterns", "sources": [], "summary": ""}, config)

state = graph.get_state(config)
print(f"Fuentes del agente: {len(state.values['sources'])}")

graph.update_state(config, {
    "sources": [
        "Fuente 3: Paper interno (no público)",
        "Fuente 4: Notas de reunión con equipo de LangGraph",
    ],
})

state = graph.get_state(config)
print(f"Fuentes después de agregar: {len(state.values['sources'])}")
for s in state.values["sources"]:
    print(f"  - {s}")

graph.invoke(None, config)

final = graph.get_state(config)
print(f"\n{final.values['summary']}")
assert "4 fuentes" in final.values["summary"]
print("✅ Las 4 fuentes fueron procesadas")
# Output esperado:
# Fuentes del agente: 2
# Fuentes después de agregar: 4
#   - Fuente 1: Wikipedia sobre HITL patterns
#   - Fuente 2: arXiv sobre HITL patterns
#   - Fuente 3: Paper interno (no público)
#   - Fuente 4: Notas de reunión con equipo de LangGraph
#
# Resumen basado en 4 fuentes
# ✅ Las 4 fuentes fueron procesadas

Ejercicio 3: Usar as_node para saltar un nodo (Medio)

Crea un grafo de 3 nodos: fetch → validate → save. Configura un breakpoint antes de validate. En vez de dejar que validate ejecute, usa update_state con as_node="validate" para proporcionar el resultado de validación directamente y saltar al nodo save.

Ver solución 
from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    data: str
    validated: bool
    validation_msg: str
    saved: bool
    log: Annotated[list[str], operator.add]

def fetch(state: State) -> dict:
    return {"data": "datos crudos del API", "log": ["fetched"]}

def validate(state: State) -> dict:
    is_valid = len(state["data"]) > 5
    return {
        "validated": is_valid,
        "validation_msg": "OK" if is_valid else "Datos insuficientes",
        "log": ["validated_by_node"],
    }

def save(state: State) -> dict:
    if state["validated"]:
        return {"saved": True, "log": ["saved"]}
    return {"saved": False, "log": ["save_skipped"]}

graph_builder = StateGraph(State)
graph_builder.add_node("fetch", fetch)
graph_builder.add_node("validate", validate)
graph_builder.add_node("save", save)

graph_builder.add_edge(START, "fetch")
graph_builder.add_edge("fetch", "validate")
graph_builder.add_edge("validate", "save")
graph_builder.add_edge("save", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["validate"],
)

config = {"configurable": {"thread_id": "skip-node"}}
graph.invoke(
    {"data": "", "validated": False, "validation_msg": "", "saved": False, "log": []},
    config,
)

state = graph.get_state(config)
print(f"Datos: {state.values['data']}")
print(f"Siguiente: {state.next}")

print("\n--- Humano valida manualmente y salta el nodo validate ---\n")

graph.update_state(
    config,
    {"validated": True, "validation_msg": "Validado manualmente por humano", "log": ["validated_by_human"]},
    as_node="validate",
)

state = graph.get_state(config)
print(f"Validated: {state.values['validated']}")
print(f"Siguiente (saltó validate): {state.next}")

graph.invoke(None, config)

final = graph.get_state(config)
print(f"Saved: {final.values['saved']}")
print(f"Log: {final.values['log']}")
# Output esperado:
# Datos: datos crudos del API
# Siguiente: ('validate',)
#
# --- Humano valida manualmente y salta el nodo validate ---
#
# Validated: True
# Siguiente (saltó validate): ('save',)
# Saved: True
# Log: ['fetched', 'validated_by_human', 'saved']

Ejercicio 4: Corregir múltiples campos a la vez (Medio)

Crea un grafo de investigación que recopila datos de una empresa (nombre, CEO, fundación, HQ). Introduce 2 errores intencionales: CEO incorrecto y año de fundación incorrecto. Pausa, corrige ambos campos en una sola llamada a update_state, y verifica que el reporte final usa los datos correctos.

Ver solución 
from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    company: str
    ceo: str
    founded: int
    hq: str
    report: str

def research(state: State) -> dict:
    return {
        "ceo": "Mark Thompson",
        "founded": 2015,
        "hq": "Austin, TX",
    }

def generate_report(state: State) -> dict:
    report = f"{state['company']}: CEO {state['ceo']}, fundada en {state['founded']}, HQ en {state['hq']}"
    return {"report": report}

graph_builder = StateGraph(State)
graph_builder.add_node("research", research)
graph_builder.add_node("report", generate_report)

graph_builder.add_edge(START, "research")
graph_builder.add_edge("research", "report")
graph_builder.add_edge("report", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["research"],
)

config = {"configurable": {"thread_id": "multi-edit"}}
graph.invoke(
    {"company": "InnovateTech", "ceo": "", "founded": 0, "hq": "", "report": ""},
    config,
)

state = graph.get_state(config)
print(f"Datos del agente:")
print(f"  CEO: {state.values['ceo']} ← INCORRECTO")
print(f"  Founded: {state.values['founded']} ← INCORRECTO")
print(f"  HQ: {state.values['hq']} ← correcto")

print("\n--- Corrigiendo CEO y año de fundación ---\n")

graph.update_state(config, {
    "ceo": "Sarah Chen",
    "founded": 2019,
})

state = graph.get_state(config)
print(f"Datos corregidos:")
print(f"  CEO: {state.values['ceo']}")
print(f"  Founded: {state.values['founded']}")
print(f"  HQ: {state.values['hq']} (sin cambios)")

graph.invoke(None, config)

final = graph.get_state(config)
print(f"\nReporte: {final.values['report']}")

assert "Sarah Chen" in final.values["report"]
assert "2019" in final.values["report"]
assert "Austin, TX" in final.values["report"]
print("✅ Reporte con datos corregidos")
# Output esperado:
# Datos del agente:
#   CEO: Mark Thompson ← INCORRECTO
#   Founded: 2015 ← INCORRECTO
#   HQ: Austin, TX ← correcto
#
# --- Corrigiendo CEO y año de fundación ---
#
# Datos corregidos:
#   CEO: Sarah Chen
#   Founded: 2019
#   HQ: Austin, TX (sin cambios)
#
# Reporte: InnovateTech: CEO Sarah Chen, fundada en 2019, HQ en Austin, TX
# ✅ Reporte con datos corregidos

Ejercicio 5: Editar estado + redirigir con as_node (Avanzado)

Crea un pipeline de análisis: collect → analyze → recommend → deliver. Configura interrupt_after=["analyze"]. Después de analizar, el humano decide que el análisis es incorrecto y quiere que se rehaga. Usa update_state con as_node="collect" para insertar nuevos datos y hacer que el grafo re-ejecute desde analyze.

Ver solución 
from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    query: str
    raw_data: str
    analysis: str
    recommendation: str
    deliverable: str
    log: Annotated[list[str], operator.add]

def collect(state: State) -> dict:
    return {"raw_data": f"Datos básicos sobre '{state['query']}'", "log": ["collected"]}

def analyze(state: State) -> dict:
    return {"analysis": f"Análisis: {state['raw_data'][:40]}", "log": ["analyzed"]}

def recommend(state: State) -> dict:
    return {"recommendation": f"Recomendación basada en: {state['analysis'][:30]}", "log": ["recommended"]}

def deliver(state: State) -> dict:
    return {"deliverable": f"ENTREGA: {state['recommendation']}", "log": ["delivered"]}

graph_builder = StateGraph(State)
graph_builder.add_node("collect", collect)
graph_builder.add_node("analyze", analyze)
graph_builder.add_node("recommend", recommend)
graph_builder.add_node("deliver", deliver)

graph_builder.add_edge(START, "collect")
graph_builder.add_edge("collect", "analyze")
graph_builder.add_edge("analyze", "recommend")
graph_builder.add_edge("recommend", "deliver")
graph_builder.add_edge("deliver", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["analyze"],
)

config = {"configurable": {"thread_id": "redo-analysis"}}

graph.invoke(
    {"query": "AI investment strategy", "raw_data": "", "analysis": "",
     "recommendation": "", "deliverable": "", "log": []},
    config,
)

state = graph.get_state(config)
print(f"Análisis actual: {state.values['analysis']}")
print(f"Siguiente: {state.next}")

print("\n--- Humano: 'Los datos son insuficientes, inyecto mejores datos' ---\n")

graph.update_state(
    config,
    {
        "raw_data": "Datos PREMIUM: análisis de 50 empresas, métricas financieras, predicciones de mercado",
        "log": ["human_injected_premium_data"],
    },
    as_node="collect",
)

state = graph.get_state(config)
print(f"Siguiente después de as_node='collect': {state.next}")
print(f"Raw data actualizado: {state.values['raw_data'][:60]}...")

print("\n--- Continuando: analyze se re-ejecuta con datos premium ---\n")
graph.invoke(None, config)

final = graph.get_state(config)
print(f"Análisis nuevo: {final.values['analysis']}")
print(f"Recomendación: {final.values['recommendation']}")
print(f"Entregable: {final.values['deliverable']}")
print(f"Log: {final.values['log']}")
# Output esperado:
# Análisis actual: Análisis: Datos básicos sobre 'AI investment str
# Siguiente: ('recommend',)
#
# --- Humano: 'Los datos son insuficientes, inyecto mejores datos' ---
#
# Siguiente después de as_node='collect': ('analyze',)
# Raw data actualizado: Datos PREMIUM: análisis de 50 empresas, métricas financier...
#
# --- Continuando: analyze se re-ejecuta con datos premium ---
#
# Análisis nuevo: Análisis: Datos PREMIUM: análisis de 50 empresas,
# Recomendación: Recomendación basada en: Análisis: Datos PREMIUM: análi
# Entregable: ENTREGA: Recomendación basada en: Análisis: Datos PREMIUM: análi
# Log: ['collected', 'analyzed', 'human_injected_premium_data', 'analyzed', 'recommended', 'delivered']

Ejercicio 6: Flujo completo de HITL con inspección, edición y aprobación (Avanzado)

Crea un agente de research con 4 nodos: plan → research → review → report. El nodo review usa interrupt() para pedir feedback al humano. Implementa el flujo completo: (1) el agente planea y ejecuta la investigación, (2) se pausa en review, (3) el humano inspecciona los datos, detecta un error en la investigación, (4) edita el estado para corregir el error, y (5) aprueba para generar el reporte.

Ver solución 
from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    objective: str
    plan: str
    findings: Annotated[list[dict], operator.add]
    corrections_applied: str
    report: str
    log: Annotated[list[str], operator.add]

def plan(state: State) -> dict:
    return {
        "plan": f"Plan para '{state['objective']}': buscar en 3 fuentes, analizar tendencias",
        "log": ["planned"],
    }

def research(state: State) -> dict:
    return {"findings": [
        {"source": "arXiv", "fact": "El mercado AI crecerá 35% en 2025", "confidence": 0.9},
        {"source": "Bloomberg", "fact": "TechCorp cotiza a $150/acción", "confidence": 0.95},
        {"source": "Blog", "fact": "El CEO de TechCorp es Robert Lee", "confidence": 0.6},
    ], "log": ["researched"]}

def review(state: State) -> dict:
    decision = interrupt({
        "message": "Revisa los hallazgos de la investigación",
        "findings": state["findings"],
        "instructions": "Responde {approved: true} después de hacer correcciones si es necesario",
    })
    return {"log": ["reviewed"]}

def report(state: State) -> dict:
    lines = [f"REPORTE: {state['objective']}"]
    lines.append(f"Plan: {state['plan']}")
    if state["corrections_applied"]:
        lines.append(f"Correcciones: {state['corrections_applied']}")
    for f in state["findings"]:
        lines.append(f"  [{f['source']}] {f['fact']} (confianza: {f['confidence']})")
    return {"report": "\n".join(lines), "log": ["reported"]}

graph_builder = StateGraph(State)
graph_builder.add_node("plan", plan)
graph_builder.add_node("research", research)
graph_builder.add_node("review", review)
graph_builder.add_node("report", report)

graph_builder.add_edge(START, "plan")
graph_builder.add_edge("plan", "research")
graph_builder.add_edge("research", "review")
graph_builder.add_edge("review", "report")
graph_builder.add_edge("report", END)

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

config = {"configurable": {"thread_id": "full-hitl"}}

print("=== PASO 1: Plan + Research + Pausa en review ===\n")
graph.invoke(
    {"objective": "Análisis de TechCorp para inversión", "plan": "",
     "findings": [], "corrections_applied": "", "report": "", "log": []},
    config,
)

state = graph.get_state(config)
print(f"Esperando en: {state.next}")
print(f"Hallazgos:")
for f in state.values["findings"]:
    flag = "⚠️" if f["confidence"] < 0.8 else "✅"
    print(f"  {flag} [{f['source']}] {f['fact']} (conf: {f['confidence']})")

print("\n=== PASO 2: Humano detecta error y corrige ===\n")

corrected_findings = []
for f in state.values["findings"]:
    if f["source"] == "Blog" and "Robert Lee" in f["fact"]:
        corrected = {**f, "fact": "El CEO de TechCorp es Amanda Reyes (desde enero 2025)", "confidence": 1.0}
        corrected_findings.append(corrected)
        print(f"  Corregido: '{f['fact']}' → '{corrected['fact']}'")
    else:
        corrected_findings.append(f)

graph.update_state(config, {
    "findings": corrected_findings,
    "corrections_applied": "CEO corregido: Robert Lee → Amanda Reyes",
})

print("\n=== PASO 3: Aprobar y generar reporte ===\n")
graph.invoke(Command(resume={"approved": True}), config)

final = graph.get_state(config)
print(final.values["report"])
print(f"\nLog: {final.values['log']}")
# Output esperado:
# === PASO 1: Plan + Research + Pausa en review ===
#
# Esperando en: ('review',)
# Hallazgos:
#   ✅ [arXiv] El mercado AI crecerá 35% en 2025 (conf: 0.9)
#   ✅ [Bloomberg] TechCorp cotiza a $150/acción (conf: 0.95)
#   ⚠️ [Blog] El CEO de TechCorp es Robert Lee (conf: 0.6)
#
# === PASO 2: Humano detecta error y corrige ===
#
#   Corregido: 'El CEO de TechCorp es Robert Lee' → 'El CEO de TechCorp es Amanda Reyes (desde enero 2025)'
#
# === PASO 3: Aprobar y generar reporte ===
#
# REPORTE: Análisis de TechCorp para inversión
# Plan: Plan para 'Análisis de TechCorp para inversión': buscar en 3 fuentes, analizar tendencias
# Correcciones: CEO corregido: Robert Lee → Amanda Reyes
#   [arXiv] El mercado AI crecerá 35% en 2025 (confianza: 0.9)
#   [Bloomberg] TechCorp cotiza a $150/acción (confianza: 0.95)
#   [Blog] El CEO de TechCorp es Amanda Reyes (desde enero 2025) (confianza: 1.0)
#
# Log: ['planned', 'researched', 'reviewed', 'reported']

Resumen

En esta cápsula aprendiste:

  • Editable state te permite modificar el estado del agente durante una pausa. No necesitas reiniciar: corriges el dato incorrecto y el agente continúa con información correcta, preservando todo el trabajo previo
  • graph.get_state(config) te da acceso completo al estado actual: todos los valores, qué nodo viene después, y el checkpoint ID. Úsalo para inspeccionar antes de editar
  • graph.update_state(config, updates) modifica el estado siguiendo las reglas de reducers: campos simples se reemplazan, campos con operator.add se agregan. Esto es fundamental para entender el resultado de tus ediciones
  • as_node controla qué nodo viene después. update_state(config, data, as_node="X") hace que LangGraph trate la edición como si el nodo X la produjo, lo que determina el siguiente nodo en el grafo
  • Los casos de uso son prácticos, no teóricos: corregir datos incorrectos, agregar información que el agente no encontró, remover resultados irrelevantes, cambiar la dirección de la investigación
  • Editar es más eficiente que reiniciar cuando el agente hizo bien la mayoría del trabajo. Preservas búsquedas, análisis, y datos correctos — solo corriges lo que está mal
  • Valida antes de editar en producción: verifica tipos, rangos, y restricciones de negocio antes de aplicar update_state

Próxima cápsula: diseño de UX para HITL — cuándo interrumpir, cuándo no, y cómo no frustrar al usuario con demasiadas pausas. El balance entre autonomía y control humano.

Recursos adicionales

  1. LangGraph — Human-in-the-Loop — Conceptos oficiales de HITL incluyendo editable state
  2. How to edit graph state — Guía práctica para inspeccionar y modificar estado durante pausas
  3. How to update graph state — Referencia completa de update_state con reducers y as_node
  4. LangGraph — State Reducers — Cómo funcionan los reducers y su impacto en update_state
  5. How to wait for user input — Combinar editable state con interrupt() para flujos interactivos

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

← AnteriorBreakpoints y ApprovalsSiguiente →User Feedback Loops