Shared vs Isolated State

Descripción de la cápsula

Esta es la decisión de diseño MÁS DIFÍCIL en sistemas multi-agente: ¿qué puede ver y modificar cada agente? Comparte demasiado y los agentes se interfieren entre sí — un agente sobrescribe el trabajo de otro, el estado crece sin control, y debuggear se vuelve imposible. Comparte muy poco y los agentes no pueden coordinarse — cada uno trabaja en su burbuja sin acceso al contexto que necesita.

Acertar en el diseño de estado determina si tu sistema funciona o crea caos. No es exageración: la mayoría de bugs en sistemas multi-agente no son errores de lógica en los agentes individuales, sino problemas de cómo comparten (o no comparten) información.

En las cápsulas anteriores usaste estado compartido sin pensarlo mucho — todos los agentes leían y escribían en el mismo TypedDict. Ahora vas a entender por qué eso funciona para sistemas pequeños pero se rompe para sistemas grandes, y vas a aprender tres patrones concretos para diseñar estado: shared, isolated, y hybrid.

La analogía de microservicios

Si vienes del mundo del desarrollo backend, esta analogía te va a clarificar todo:

En microservicios, compartir una base de datos entre servicios es un anti-pattern conocido: produce acoplamiento, conflictos, y hace imposible escalar servicios independientemente. La misma lógica aplica a agentes.

La diferencia: en microservicios, separar databases requiere infraestructura real. En LangGraph, separar estado es una decisión de diseño en tu TypedDict. Es más fácil de implementar, pero igual de importante.

Pattern 1: Shared state — todos ven todo

El patrón más simple: un solo TypedDict, todos los agentes leen y escriben los mismos campos.

Implementación

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from IPython.display import Image, display

class SharedState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    research_results: Annotated[list[str], operator.add]
    analysis: str
    final_report: str
    current_agent: str

model = init_chat_model("openai:gpt-4.1-mini")

def researcher(state: SharedState) -> dict:
    response = model.invoke([
        SystemMessage(content="Eres un investigador. Encuentra 3 datos clave sobre el tema."),
        state["messages"][-1],
    ])
    findings = [f.strip() for f in response.content.split("\n") if f.strip()]
    return {
        "research_results": findings,
        "current_agent": "analyst",
    }

def analyst(state: SharedState) -> dict:
    findings_text = "\n".join(state["research_results"])
    response = model.invoke(
        f"Analiza estos hallazgos y extrae la conclusión principal:\n{findings_text}"
    )
    return {
        "analysis": response.content,
        "current_agent": "writer",
    }

def writer(state: SharedState) -> dict:
    response = model.invoke(
        f"Escribe un resumen ejecutivo basado en:\n"
        f"Hallazgos: {', '.join(state['research_results'][:3])}\n"
        f"Análisis: {state['analysis']}"
    )
    return {
        "final_report": response.content,
        "current_agent": "done",
    }

graph = StateGraph(SharedState)
graph.add_node("researcher", researcher)
graph.add_node("analyst", analyst)
graph.add_node("writer", writer)
graph.add_edge(START, "researcher")
graph.add_edge("researcher", "analyst")
graph.add_edge("analyst", "writer")
graph.add_edge("writer", END)

app = graph.compile()
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({
    "messages": [HumanMessage(content="Impacto de la IA en la educación")],
    "research_results": [],
    "analysis": "",
    "final_report": "",
    "current_agent": "researcher",
})

print(f"Research results: {len(result['research_results'])} items")
print(f"Analysis: {result['analysis'][:100]}...")
print(f"Report: {result['final_report'][:100]}...")
# Output:
# Research results: 3 items
# Analysis: La IA está transformando la educación en tres dimensiones principales...
# Report: Resumen ejecutivo: La integración de IA en educación muestra impacto...

Cuándo funciona

• ✅ Sistemas pequeños (2-4 agentes)
• ✅ Los agentes trabajan en secuencia (uno termina antes de que el siguiente empiece)
• ✅ Cada agente necesita contexto de los anteriores (el analyst necesita ver research_results)
• ✅ El orden de ejecución es fijo y predecible

El riesgo: interferencia entre agentes

El problema aparece cuando dos agentes modifican el mismo campo. Mira qué pasa si dos researchers trabajan en paralelo:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class DangerousState(TypedDict):
    topic: str
    findings: str

def researcher_a(state: DangerousState) -> dict:
    return {"findings": "Hallazgo A: La IA mejora el aprendizaje personalizado"}

def researcher_b(state: DangerousState) -> dict:
    return {"findings": "Hallazgo B: La IA reduce costos educativos"}

graph = StateGraph(DangerousState)
graph.add_node("researcher_a", researcher_a)
graph.add_node("researcher_b", researcher_b)

graph.add_edge(START, "researcher_a")
graph.add_edge(START, "researcher_b")
graph.add_edge("researcher_a", END)
graph.add_edge("researcher_b", END)

app = graph.compile()
result = app.invoke({"topic": "IA en educación", "findings": ""})
print(result["findings"])
# Output: Hallazgo B: La IA reduce costos educativos
# ¡Hallazgo A se PERDIÓ! El último en escribir gana.

Sin un reducer, findings se sobrescribe. Si ambos nodos corren "en paralelo" (LangGraph ejecuta ambos en el mismo superstep), el resultado depende del orden interno de ejecución. Esto es exactamente el mismo problema que una race condition en una base de datos compartida.

Mitigación: reducers para acumulación segura

from typing import TypedDict, Annotated
import operator

class SafeSharedState(TypedDict):
    topic: str
    findings: Annotated[list[str], operator.add]

def researcher_a(state: SafeSharedState) -> dict:
    return {"findings": ["Hallazgo A: La IA mejora el aprendizaje personalizado"]}

def researcher_b(state: SafeSharedState) -> dict:
    return {"findings": ["Hallazgo B: La IA reduce costos educativos"]}

Con operator.add, ambos hallazgos se acumulan en la lista. El reducer elimina el conflicto de sobrescritura.

Pattern 2: Isolated state — cada agente tiene su scope

Cada agente opera con su propio TypedDict. El estado del agente no es visible para otros agentes directamente. La comunicación pasa por el nodo padre (supervisor o router) que mapea datos entre scopes.

Implementación con subgraph

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from IPython.display import Image, display

class ResearchSubState(TypedDict):
    query: str
    sources: Annotated[list[str], operator.add]
    findings: Annotated[list[str], operator.add]

class AnalysisSubState(TypedDict):
    input_data: list[str]
    conclusion: str
    confidence: float

class ParentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    research_output: list[str]
    analysis_conclusion: str
    analysis_confidence: float
    final_report: str

model = init_chat_model("openai:gpt-4.1-mini")

def research_node_1(state: ResearchSubState) -> dict:
    response = model.invoke(
        f"Busca 2 fuentes sobre: {state['query']}. Lista las fuentes separadas por '|'."
    )
    sources = [s.strip() for s in response.content.split("|")]
    return {"sources": sources}

def research_node_2(state: ResearchSubState) -> dict:
    response = model.invoke(
        f"Basándote en estas fuentes: {', '.join(state['sources'])}\n"
        f"Extrae los hallazgos principales sobre: {state['query']}"
    )
    findings = [f.strip() for f in response.content.split("\n") if f.strip()]
    return {"findings": findings}

research_graph = StateGraph(ResearchSubState)
research_graph.add_node("find_sources", research_node_1)
research_graph.add_node("extract_findings", research_node_2)
research_graph.add_edge(START, "find_sources")
research_graph.add_edge("find_sources", "extract_findings")
research_graph.add_edge("extract_findings", END)
research_subgraph = research_graph.compile()

def analysis_node(state: AnalysisSubState) -> dict:
    data_text = "\n".join(state["input_data"])
    response = model.invoke(
        f"Analiza estos datos y da una conclusión con nivel de confianza (0-1):\n{data_text}"
    )
    return {"conclusion": response.content, "confidence": 0.85}

analysis_graph = StateGraph(AnalysisSubState)
analysis_graph.add_node("analyze", analysis_node)
analysis_graph.add_edge(START, "analyze")
analysis_graph.add_edge("analyze", END)
analysis_subgraph = analysis_graph.compile()

def run_research(state: ParentState) -> dict:
    query = state["messages"][-1].content
    result = research_subgraph.invoke({
        "query": query,
        "sources": [],
        "findings": [],
    })
    return {"research_output": result["findings"]}

def run_analysis(state: ParentState) -> dict:
    result = analysis_subgraph.invoke({
        "input_data": state["research_output"],
        "conclusion": "",
        "confidence": 0.0,
    })
    return {
        "analysis_conclusion": result["conclusion"],
        "analysis_confidence": result["confidence"],
    }

def write_report(state: ParentState) -> dict:
    response = model.invoke(
        f"Genera un reporte ejecutivo:\n"
        f"Hallazgos: {state['research_output'][:3]}\n"
        f"Conclusión: {state['analysis_conclusion']}\n"
        f"Confianza: {state['analysis_confidence']}"
    )
    return {"final_report": response.content}

parent_graph = StateGraph(ParentState)
parent_graph.add_node("research", run_research)
parent_graph.add_node("analysis", run_analysis)
parent_graph.add_node("report", write_report)
parent_graph.add_edge(START, "research")
parent_graph.add_edge("research", "analysis")
parent_graph.add_edge("analysis", "report")
parent_graph.add_edge("report", END)

app = parent_graph.compile()
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({
    "messages": [HumanMessage(content="Estado actual de la computación cuántica")],
    "research_output": [],
    "analysis_conclusion": "",
    "analysis_confidence": 0.0,
    "final_report": "",
})

print(f"Research findings: {len(result['research_output'])} items")
print(f"Analysis confidence: {result['analysis_confidence']}")
print(f"Report: {result['final_report'][:120]}...")
# Output:
# Research findings: 3 items
# Analysis confidence: 0.85
# Report: Reporte Ejecutivo: La computación cuántica se encuentra en una fase de...

Lo clave del patrón

Observa cómo los estados están completamente separados:

• ResearchSubState tiene query, sources, findings — campos que solo el researcher necesita
• AnalysisSubState tiene input_data, conclusion, confidence — campos que solo el analyst necesita
• ParentState tiene los resultados finales de cada agente — es el "contrato" entre agentes

El researcher no puede ver conclusion. El analyst no puede ver sources. Cada agente opera en su propio mundo, y el padre mapea los datos entre ellos:

research.findings → parent.research_output → analysis.input_data

Cuándo usar isolated state

• ✅ Los agentes pueden trabajar en paralelo sin dependencias
• ✅ Cada agente tiene datos internos que otros no necesitan ver
• ✅ El sistema es grande (5+ agentes) y el estado compartido sería un mess
• ✅ Necesitas que cada agente sea testeable de forma independiente

El riesgo: overhead de mapping

Cada vez que quieres pasar datos entre agentes, necesitas código de mapping explícito. Si tienes 5 agentes que comparten resultados en diferentes combinaciones, el mapping en el padre se vuelve tedioso.

Pattern 3: Hybrid — compartido para coordinar, aislado para trabajar

Este es el patrón recomendado para la mayoría de sistemas en producción. Define un estado compartido mínimo para coordinación y permite que cada agente tenga campos privados para su trabajo interno.

Diseño del estado

from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage

class HybridState(TypedDict):
    # --- COMPARTIDO: todos los agentes ven y usan ---
    messages: Annotated[list[AnyMessage], operator.add]
    task_status: str
    current_phase: str

    # --- OUTPUTS de cada agente (compartidos como resultados) ---
    research_results: Annotated[list[str], operator.add]
    analysis_summary: str
    final_report: str

    # --- AISLADO por convención: solo lo usa el agente que lo "posee" ---
    _research_sources: Annotated[list[str], operator.add]
    _research_queries: Annotated[list[str], operator.add]
    _analysis_metrics: dict
    _writer_drafts: Annotated[list[str], operator.add]

La convención _prefijo indica campos que "pertenecen" a un agente específico. No es enforcement técnico (cualquier agente PUEDE leer _research_sources), pero es un contrato de equipo: "este campo es responsabilidad del researcher, no lo toques."

Implementación completa

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from IPython.display import Image, display

class TeamState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    task_status: str
    current_phase: str
    research_results: Annotated[list[str], operator.add]
    analysis_summary: str
    final_report: str
    _research_sources: Annotated[list[str], operator.add]
    _analysis_raw_scores: Annotated[list[float], operator.add]

model = init_chat_model("openai:gpt-4.1-mini")

def researcher(state: TeamState) -> dict:
    topic = state["messages"][-1].content
    response = model.invoke([
        SystemMessage(content="Encuentra 3 datos clave. Sepáralos con '|'."),
        HumanMessage(content=topic),
    ])
    findings = [f.strip() for f in response.content.split("|") if f.strip()]
    sources = ["web_search", "academic_db", "news_api"]

    return {
        "research_results": findings,
        "_research_sources": sources,
        "task_status": "research_complete",
        "current_phase": "analysis",
    }

def analyst(state: TeamState) -> dict:
    findings_text = "\n".join(state["research_results"])
    response = model.invoke(
        f"Analiza estos hallazgos. Da un resumen de 2 oraciones:\n{findings_text}"
    )
    return {
        "analysis_summary": response.content,
        "_analysis_raw_scores": [0.8, 0.75, 0.9],
        "task_status": "analysis_complete",
        "current_phase": "writing",
    }

def writer(state: TeamState) -> dict:
    response = model.invoke(
        f"Escribe un reporte ejecutivo en 3 párrafos basado en:\n"
        f"Hallazgos: {state['research_results']}\n"
        f"Análisis: {state['analysis_summary']}"
    )
    return {
        "final_report": response.content,
        "task_status": "complete",
        "current_phase": "done",
    }

graph = StateGraph(TeamState)
graph.add_node("researcher", researcher)
graph.add_node("analyst", analyst)
graph.add_node("writer", writer)
graph.add_edge(START, "researcher")
graph.add_edge("researcher", "analyst")
graph.add_edge("analyst", "writer")
graph.add_edge("writer", END)

app = graph.compile()
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({
    "messages": [HumanMessage(content="Tendencias en energías renovables 2025")],
    "task_status": "started",
    "current_phase": "research",
    "research_results": [],
    "analysis_summary": "",
    "final_report": "",
    "_research_sources": [],
    "_analysis_raw_scores": [],
})

print(f"Status: {result['task_status']}")
print(f"Phase: {result['current_phase']}")
print(f"Sources (private to researcher): {result['_research_sources']}")
print(f"Raw scores (private to analyst): {result['_analysis_raw_scores']}")
print(f"Research results (shared): {len(result['research_results'])} items")
print(f"Report (first 100 chars): {result['final_report'][:100]}...")
# Output:
# Status: complete
# Phase: done
# Sources (private to researcher): ['web_search', 'academic_db', 'news_api']
# Raw scores (private to analyst): [0.8, 0.75, 0.9]
# Research results (shared): 3 items
# Report (first 100 chars): Reporte Ejecutivo: Las energías renovables continúan su expansión global...

Por qué hybrid es el default recomendado

1. Coordinación visible: task_status y current_phase están disponibles para todos — cualquier agente puede saber en qué paso estamos
2. Resultados accesibles: research_results y analysis_summary son outputs compartidos — el writer necesita ver ambos
3. Internals protegidos: _research_sources y _analysis_raw_scores son datos de trabajo que solo importan a su agente dueño
4. Debugging claro: puedes inspeccionar el estado final y saber exactamente qué produjo cada agente

Diseñando el contrato de estado

Antes de escribir un solo nodo, hazte estas tres preguntas para cada agente:

1. ¿Qué NECESITA ver este agente? (inputs mínimos)

# Researcher necesita:
#   - messages (para saber qué investigar)
#   - current_phase (para saber si es su turno)
# NO necesita: analysis_summary, final_report, _analysis_raw_scores

# Analyst necesita:
#   - research_results (para analizar)
# NO necesita: _research_sources, _research_queries

# Writer necesita:
#   - research_results + analysis_summary (para escribir)
# NO necesita: _research_sources, _analysis_raw_scores

2. ¿Qué PRODUCE este agente? (outputs)

# Researcher produce:
#   - research_results (compartido, otros lo necesitan)
#   - _research_sources (privado, solo para debugging)

# Analyst produce:
#   - analysis_summary (compartido, el writer lo necesita)
#   - _analysis_raw_scores (privado, solo para debugging)

# Writer produce:
#   - final_report (compartido, es el output del sistema)

3. ¿Cómo se resuelven conflictos si dos agentes escriben al mismo campo?

# research_results: operator.add → se acumulan (múltiples researchers pueden contribuir)
# analysis_summary: sin reducer → el último analyst gana (solo hay uno)
# task_status: sin reducer → el último en ejecutar define el status actual

Este análisis es equivalente a diseñar interfaces de API entre microservicios. Cada agente tiene un "contrato": qué recibe, qué retorna, y qué garantías ofrece. Dedica tiempo a esto ANTES de escribir código.

Previniendo conflictos: ejecución paralela y estado

Cuando dos agentes se ejecutan en paralelo (mismo superstep en LangGraph), los conflictos de estado son reales. Veamos las estrategias:

Estrategia 1: Reducers para acumulación segura

Si dos agentes escriben al mismo campo de tipo lista, operator.add los fusiona:

from dotenv import load_dotenv
load_dotenv()

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

class ParallelState(TypedDict):
    topic: str
    findings: Annotated[list[str], operator.add]

def web_researcher(state: ParallelState) -> dict:
    return {"findings": [f"[Web] Dato sobre {state['topic']}"]}

def academic_researcher(state: ParallelState) -> dict:
    return {"findings": [f"[Academic] Estudio sobre {state['topic']}"]}

def news_researcher(state: ParallelState) -> dict:
    return {"findings": [f"[News] Noticia sobre {state['topic']}"]}

graph = StateGraph(ParallelState)
graph.add_node("web", web_researcher)
graph.add_node("academic", academic_researcher)
graph.add_node("news", news_researcher)

graph.add_edge(START, "web")
graph.add_edge(START, "academic")
graph.add_edge(START, "news")
graph.add_edge("web", END)
graph.add_edge("academic", END)
graph.add_edge("news", END)

app = graph.compile()
result = app.invoke({"topic": "IA generativa", "findings": []})
print(result["findings"])
# Output: ['[Web] Dato sobre IA generativa', '[Academic] Estudio sobre IA generativa', '[News] Noticia sobre IA generativa']

Tres agentes corriendo en paralelo, todos escribiendo a findings. Gracias a operator.add, los tres resultados se acumulan sin conflictos.

Estrategia 2: Campos separados por agente

Si cada agente produce datos de diferente naturaleza, usa campos separados:

from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage

class SeparatedState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    web_results: Annotated[list[str], operator.add]
    academic_results: Annotated[list[str], operator.add]
    news_results: Annotated[list[str], operator.add]

Cero conflictos porque cada agente escribe a su propio campo. El nodo siguiente puede leer los tres campos para combinarlos.

Estrategia 3: Custom reducer con timestamp

Para campos escalares que múltiples agentes podrían actualizar, un reducer con timestamp preserva el más reciente:

import time

def latest_wins(current: dict, new: dict) -> dict:
    """Mantiene el valor con timestamp más reciente."""
    if not current or new.get("timestamp", 0) > current.get("timestamp", 0):
        return new
    return current

class TimestampedState(TypedDict):
    status: Annotated[dict, latest_wins]

Comunicación entre agentes: message passing a través del estado

Los agentes no se llaman directamente entre sí. Se comunican escribiendo y leyendo campos del estado. Esto es message passing implícito.

Patrón: resultados que fluyen de un agente a otro

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from IPython.display import Image, display

class PipelineState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    raw_data: str
    cleaned_data: str
    analysis: str
    recommendations: Annotated[list[str], operator.add]

model = init_chat_model("openai:gpt-4.1-mini")

def data_collector(state: PipelineState) -> dict:
    """Recolecta datos crudos. Escribe a: raw_data."""
    topic = state["messages"][-1].content
    response = model.invoke(f"Genera datos de ejemplo sobre: {topic}. Formato: lista de 3 métricas.")
    return {"raw_data": response.content}

def data_cleaner(state: PipelineState) -> dict:
    """Lee de: raw_data. Escribe a: cleaned_data."""
    response = model.invoke(
        f"Limpia y estructura estos datos:\n{state['raw_data']}"
    )
    return {"cleaned_data": response.content}

def data_analyst(state: PipelineState) -> dict:
    """Lee de: cleaned_data. Escribe a: analysis, recommendations."""
    response = model.invoke(
        f"Analiza estos datos y da 2 recomendaciones:\n{state['cleaned_data']}"
    )
    return {
        "analysis": response.content,
        "recommendations": ["Recomendación basada en análisis"],
    }

graph = StateGraph(PipelineState)
graph.add_node("collector", data_collector)
graph.add_node("cleaner", data_cleaner)
graph.add_node("analyst", data_analyst)
graph.add_edge(START, "collector")
graph.add_edge("collector", "cleaner")
graph.add_edge("cleaner", "analyst")
graph.add_edge("analyst", END)

app = graph.compile()
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({
    "messages": [HumanMessage(content="Tráfico web del último mes")],
    "raw_data": "",
    "cleaned_data": "",
    "analysis": "",
    "recommendations": [],
})

print(f"Raw data: {result['raw_data'][:80]}...")
print(f"Cleaned: {result['cleaned_data'][:80]}...")
print(f"Analysis: {result['analysis'][:80]}...")
print(f"Recommendations: {result['recommendations']}")
# Output:
# Raw data: 1. Visitas totales: 45,000  2. Bounce rate: 62%  3. Tiempo promedio...
# Cleaned: Métricas de tráfico web: Visitas: 45,000 | Bounce rate: 62% | Tiempo...
# Analysis: El análisis revela un bounce rate elevado (62%) que sugiere problemas...
# Recommendations: ['Recomendación basada en análisis']

El flujo de datos es explícito:

collector → raw_data → cleaner → cleaned_data → analyst → analysis + recommendations

Cada agente lee los campos que necesita y escribe los campos que produce. No hay llamadas directas entre agentes — todo pasa a través del estado.

Ejemplo completo: sistema hybrid con 4 agentes

Este ejemplo integra todo: estado compartido para coordinación, campos dedicados por agente, reducers para acumulación, y flujo paralelo + secuencial.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from IPython.display import Image, display

class ProductionState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    phase: str
    task_description: str

    web_findings: Annotated[list[str], operator.add]
    paper_findings: Annotated[list[str], operator.add]

    all_findings: Annotated[list[str], operator.add]
    analysis: str
    report: str

model = init_chat_model("openai:gpt-4.1-mini")

def web_researcher(state: ProductionState) -> dict:
    response = model.invoke([
        SystemMessage(content="Busca información web. Da 2 hallazgos separados por '|'."),
        HumanMessage(content=state["task_description"]),
    ])
    findings = [f"[Web] {f.strip()}" for f in response.content.split("|") if f.strip()]
    return {
        "web_findings": findings,
        "all_findings": findings,
    }

def paper_researcher(state: ProductionState) -> dict:
    response = model.invoke([
        SystemMessage(content="Busca en papers académicos. Da 2 hallazgos separados por '|'."),
        HumanMessage(content=state["task_description"]),
    ])
    findings = [f"[Paper] {f.strip()}" for f in response.content.split("|") if f.strip()]
    return {
        "paper_findings": findings,
        "all_findings": findings,
    }

def analyst(state: ProductionState) -> dict:
    all_data = "\n".join(state["all_findings"])
    response = model.invoke(
        f"Analiza todos los hallazgos y sintetiza en una conclusión de 2-3 oraciones:\n{all_data}"
    )
    return {
        "analysis": response.content,
        "phase": "writing",
    }

def writer(state: ProductionState) -> dict:
    response = model.invoke(
        f"Escribe un reporte ejecutivo (3 párrafos) basado en:\n"
        f"Hallazgos web: {state['web_findings']}\n"
        f"Hallazgos académicos: {state['paper_findings']}\n"
        f"Análisis: {state['analysis']}"
    )
    return {
        "report": response.content,
        "phase": "complete",
    }

graph = StateGraph(ProductionState)
graph.add_node("web_researcher", web_researcher)
graph.add_node("paper_researcher", paper_researcher)
graph.add_node("analyst", analyst)
graph.add_node("writer", writer)

graph.add_edge(START, "web_researcher")
graph.add_edge(START, "paper_researcher")
graph.add_edge("web_researcher", "analyst")
graph.add_edge("paper_researcher", "analyst")
graph.add_edge("analyst", "writer")
graph.add_edge("writer", END)

app = graph.compile()
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({
    "messages": [HumanMessage(content="Investigar sobre IA en medicina")],
    "phase": "research",
    "task_description": "Impacto de la inteligencia artificial en diagnóstico médico",
    "web_findings": [],
    "paper_findings": [],
    "all_findings": [],
    "analysis": "",
    "report": "",
})

print(f"Phase: {result['phase']}")
print(f"Web findings: {len(result['web_findings'])} items")
print(f"Paper findings: {len(result['paper_findings'])} items")
print(f"All findings: {len(result['all_findings'])} items")
print(f"Analysis: {result['analysis'][:100]}...")
print(f"Report: {result['report'][:100]}...")
# Output:
# Phase: complete
# Web findings: 2 items
# Paper findings: 2 items
# All findings: 4 items
# Analysis: La IA está demostrando mejoras significativas en diagnóstico médico...
# Report: Reporte Ejecutivo: La integración de inteligencia artificial en el...

El diseño del estado refleja la arquitectura:

• web_findings y paper_findings: campos aislados, cada researcher escribe al suyo
• all_findings: campo compartido con operator.add donde ambos researchers acumulan
• analysis y report: outputs secuenciales, cada uno lee del anterior
• phase: coordinación global, indica en qué etapa está el sistema

Tabla de decisión: ¿qué patrón elegir?

Troubleshooting

Problema 1: Un agente lee un campo que otro agente aún no ha escrito

Síntoma: KeyError: 'analysis_summary' en el writer, porque el analyst no ha ejecutado aún. Causa: Los edges del grafo no garantizan el orden correcto, o el campo no tiene valor default. Solución: Verifica que los edges fuerzan la secuencia correcta, y siempre incluye valores iniciales:

result = app.invoke({
    "analysis_summary": "",  # Valor default
    "research_results": [],
    # ... todos los campos con defaults
})

Problema 2: Los datos de ejecución paralela se pierden

Síntoma: Dos agentes corren en paralelo escribiendo al mismo campo de tipo lista, pero solo aparecen los resultados de uno. Causa: El campo no tiene Annotated[list, operator.add]. Solución: Agrega el reducer para campos que múltiples agentes escriben:

class State(TypedDict):
    findings: list[str]  # MAL → el último agente sobrescribe

class State(TypedDict):
    findings: Annotated[list[str], operator.add]  # BIEN → se acumulan

Problema 3: El estado crece sin control en loops largos

Síntoma: Después de 20 iteraciones de un loop supervisor → agents, el estado tiene miles de entries en las listas acumuladas, y cada llamada al LLM es más lenta y cara. Causa: operator.add acumula indefinidamente. En loops largos, las listas crecen sin límite. Solución: Usa un custom reducer que limite el tamaño:

def capped_list(current: list, new: list, max_size: int = 50) -> list:
    combined = current + new
    return combined[-max_size:]

class State(TypedDict):
    findings: Annotated[list[str], lambda c, n: (c + n)[-50:]]

Problema 4: No sé qué agente escribió qué en el estado

Síntoma: El estado final tiene research_results con 8 items, pero no sabes cuáles vinieron de qué agente. Causa: Múltiples agentes acumulan al mismo campo sin identificarse. Solución: Incluye metadata de origen en cada entrada:

def researcher_a(state) -> dict:
    return {"findings": [{"source": "researcher_a", "data": "hallazgo X"}]}

def researcher_b(state) -> dict:
    return {"findings": [{"source": "researcher_b", "data": "hallazgo Y"}]}

Problema 5: El subgraph no puede acceder al estado del padre

Síntoma: Un subgraph compilado necesita datos del parent state, pero su TypedDict es diferente y no tiene acceso. Causa: Los subgraphs tienen su propio estado aislado. Ese es el punto. Solución: El nodo del padre que invoca el subgraph debe mapear explícitamente los datos necesarios:

def run_sub_agent(state: ParentState) -> dict:
    sub_result = sub_graph.invoke({
        "query": state["messages"][-1].content,
        "context": state["research_results"],
    })
    return {"analysis": sub_result["conclusion"]}

Ejercicios

Ejercicio 1: Identificar el patrón correcto (Fácil)

Para cada escenario, decide si usarías shared, isolated, o hybrid state. Justifica.

A) 3 agentes en secuencia: researcher → analyst → writer. Cada uno necesita el output del anterior.

B) 5 agentes de traducción que traducen el mismo texto a 5 idiomas en paralelo.

C) Un supervisor con 3 agentes especializados que trabajan en fases, pero cada agente tiene cálculos internos que los otros no necesitan ver.

Ejercicio 2: Corregir el conflicto de estado (Fácil)

El siguiente código tiene un bug: dos researchers corren en paralelo pero solo los resultados de uno sobreviven. Corrígelo.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class BuggyState(TypedDict):
    topic: str
    results: list[str]

def researcher_a(state: BuggyState) -> dict:
    return {"results": [f"Resultado A sobre {state['topic']}"]}

def researcher_b(state: BuggyState) -> dict:
    return {"results": [f"Resultado B sobre {state['topic']}"]}

graph = StateGraph(BuggyState)
graph.add_node("a", researcher_a)
graph.add_node("b", researcher_b)
graph.add_edge(START, "a")
graph.add_edge(START, "b")
graph.add_edge("a", END)
graph.add_edge("b", END)

app = graph.compile()
result = app.invoke({"topic": "IA", "results": []})
print(result["results"])
# Actual: ['Resultado B sobre IA'] — ¡Resultado A se perdió!

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

class FixedState(TypedDict):
    topic: str
    results: Annotated[list[str], operator.add]  # ← Agregar reducer

def researcher_a(state: FixedState) -> dict:
    return {"results": [f"Resultado A sobre {state['topic']}"]}

def researcher_b(state: FixedState) -> dict:
    return {"results": [f"Resultado B sobre {state['topic']}"]}

graph = StateGraph(FixedState)
graph.add_node("a", researcher_a)
graph.add_node("b", researcher_b)
graph.add_edge(START, "a")
graph.add_edge(START, "b")
graph.add_edge("a", END)
graph.add_edge("b", END)

app = graph.compile()
result = app.invoke({"topic": "IA", "results": []})
print(result["results"])
# Output: ['Resultado A sobre IA', 'Resultado B sobre IA']

Ejercicio 3: Diseñar estado hybrid para un equipo de 4 agentes (Medio)

Diseña el TypedDict para un sistema de 4 agentes: data_collector (recolecta datos de APIs), validator (valida formato y calidad), transformer (transforma datos), reporter (genera reporte). Define qué campos son compartidos, cuáles privados, y cuáles necesitan reducers. No necesitas implementar los agentes.

from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage

class DataPipelineState(TypedDict):
    # COMPARTIDO: coordinación
    messages: Annotated[list[AnyMessage], operator.add]
    current_phase: str
    error_count: int

    # OUTPUTS compartidos (cada agente escribe, el siguiente lee)
    raw_data: Annotated[list[dict], operator.add]
    validation_result: dict
    transformed_data: list[dict]
    final_report: str

    # PRIVADO del collector
    _collector_api_calls: Annotated[list[str], operator.add]
    _collector_retries: int

    # PRIVADO del validator
    _validation_errors: Annotated[list[str], operator.add]
    _validation_schema: str

    # PRIVADO del transformer
    _transform_steps: Annotated[list[str], operator.add]
    _transform_duration_ms: float

    # PRIVADO del reporter
    _report_drafts: Annotated[list[str], operator.add]

Ejercicio 4: Implementar comunicación entre agentes (Medio)

Crea un sistema de 3 agentes donde el planner escribe un plan, el executor lee el plan y genera resultados, y el reviewer lee los resultados y decide si el plan se cumplió. Usa el patrón hybrid. Implementa con StateGraph.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage

class PlanExecuteState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    phase: str

    plan: str
    execution_results: Annotated[list[str], operator.add]
    review: str
    plan_fulfilled: bool

    _planner_alternatives: Annotated[list[str], operator.add]
    _executor_attempts: int

model = init_chat_model("openai:gpt-4.1-mini")

def planner(state: PlanExecuteState) -> dict:
    task = state["messages"][-1].content
    response = model.invoke([
        SystemMessage(content="Crea un plan de 3 pasos para esta tarea. Sé conciso."),
        HumanMessage(content=task),
    ])
    return {
        "plan": response.content,
        "phase": "executing",
        "_planner_alternatives": ["Plan alternativo: enfoque diferente"],
    }

def executor(state: PlanExecuteState) -> dict:
    response = model.invoke(
        f"Ejecuta este plan y reporta resultados de cada paso:\n{state['plan']}"
    )
    results = [r.strip() for r in response.content.split("\n") if r.strip()]
    return {
        "execution_results": results,
        "phase": "reviewing",
        "_executor_attempts": 1,
    }

def reviewer(state: PlanExecuteState) -> dict:
    response = model.invoke(
        f"Revisa si el plan se cumplió:\n"
        f"Plan: {state['plan']}\n"
        f"Resultados: {state['execution_results']}\n"
        f"Responde 'CUMPLIDO' o 'NO CUMPLIDO' seguido de una explicación breve."
    )
    fulfilled = "cumplido" in response.content.lower()[:20]
    return {
        "review": response.content,
        "plan_fulfilled": fulfilled,
        "phase": "complete",
    }

graph = StateGraph(PlanExecuteState)
graph.add_node("planner", planner)
graph.add_node("executor", executor)
graph.add_node("reviewer", reviewer)
graph.add_edge(START, "planner")
graph.add_edge("planner", "executor")
graph.add_edge("executor", "reviewer")
graph.add_edge("reviewer", END)

app = graph.compile()

result = app.invoke({
    "messages": [HumanMessage(content="Investigar las ventajas de Python para data science")],
    "phase": "planning",
    "plan": "",
    "execution_results": [],
    "review": "",
    "plan_fulfilled": False,
    "_planner_alternatives": [],
    "_executor_attempts": 0,
})

print(f"Phase: {result['phase']}")
print(f"Plan: {result['plan'][:100]}...")
print(f"Execution results: {len(result['execution_results'])} items")
print(f"Review: {result['review'][:100]}...")
print(f"Plan fulfilled: {result['plan_fulfilled']}")
# Output:
# Phase: complete
# Plan: 1. Investigar las principales librerías de Python para data science...
# Execution results: 3 items
# Review: CUMPLIDO. Los resultados cubren los tres pasos del plan...
# Plan fulfilled: True

Ejercicio 5: Subgraph con estado aislado (Avanzado)

Crea un sistema donde el parent graph tiene ParentState y un subgraph tiene ResearchSubState. El parent invoca el subgraph pasándole solo los datos necesarios, y recibe de vuelta solo los resultados. Demuestra que el subgraph NO puede ver campos del parent.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage

class ResearchSubState(TypedDict):
    query: str
    max_results: int
    findings: Annotated[list[str], operator.add]

class ParentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    budget: float
    deadline: str
    research_output: Annotated[list[str], operator.add]
    summary: str

model = init_chat_model("openai:gpt-4.1-mini")

def sub_search(state: ResearchSubState) -> dict:
    response = model.invoke(
        f"Busca {state['max_results']} datos sobre: {state['query']}. Separa con '|'."
    )
    findings = [f.strip() for f in response.content.split("|") if f.strip()]
    return {"findings": findings[:state["max_results"]]}

sub_graph_builder = StateGraph(ResearchSubState)
sub_graph_builder.add_node("search", sub_search)
sub_graph_builder.add_edge(START, "search")
sub_graph_builder.add_edge("search", END)
research_subgraph = sub_graph_builder.compile()

def invoke_research(state: ParentState) -> dict:
    query = state["messages"][-1].content
    sub_result = research_subgraph.invoke({
        "query": query,
        "max_results": 3,
        "findings": [],
    })
    return {"research_output": sub_result["findings"]}

def summarize(state: ParentState) -> dict:
    findings = "\n".join(state["research_output"])
    response = model.invoke(f"Resume estos hallazgos en 2 oraciones:\n{findings}")
    return {"summary": response.content}

parent_builder = StateGraph(ParentState)
parent_builder.add_node("research", invoke_research)
parent_builder.add_node("summarize", summarize)
parent_builder.add_edge(START, "research")
parent_builder.add_edge("research", "summarize")
parent_builder.add_edge("summarize", END)

app = parent_builder.compile()
result = app.invoke({
    "messages": [HumanMessage(content="Tendencias en blockchain 2025")],
    "budget": 1000.0,
    "deadline": "2025-12-31",
    "research_output": [],
    "summary": "",
})

print(f"Budget (parent only): {result['budget']}")
print(f"Deadline (parent only): {result['deadline']}")
print(f"Research output: {len(result['research_output'])} items")
print(f"Summary: {result['summary'][:100]}...")
# Output:
# Budget (parent only): 1000.0
# Deadline (parent only): 2025-12-31
# Research output: 3 items
# Summary: El blockchain continúa evolucionando con tendencias hacia...

Ejercicio 6: Custom reducer para merge de agentes paralelos (Avanzado)

Crea un custom reducer que combine resultados de múltiples agentes paralelos, dando prioridad por confianza. Cada agente retorna un dict con {"data": str, "confidence": float}. El reducer debe mantener una lista ordenada por confianza descendente, con un máximo de 5 items.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END

def priority_merge(current: list[dict], new: list[dict]) -> list[dict]:
    """Combina resultados por confianza, mantiene los top 5."""
    combined = current + new
    sorted_results = sorted(combined, key=lambda x: x["confidence"], reverse=True)
    return sorted_results[:5]

class PriorityState(TypedDict):
    topic: str
    results: Annotated[list[dict], priority_merge]

def agent_high_confidence(state: PriorityState) -> dict:
    return {"results": [
        {"data": f"Hallazgo preciso sobre {state['topic']}", "confidence": 0.95},
        {"data": f"Dato verificado de {state['topic']}", "confidence": 0.88},
    ]}

def agent_medium_confidence(state: PriorityState) -> dict:
    return {"results": [
        {"data": f"Información parcial sobre {state['topic']}", "confidence": 0.65},
        {"data": f"Referencia general de {state['topic']}", "confidence": 0.55},
    ]}

def agent_low_confidence(state: PriorityState) -> dict:
    return {"results": [
        {"data": f"Dato no verificado sobre {state['topic']}", "confidence": 0.30},
        {"data": f"Rumor sobre {state['topic']}", "confidence": 0.15},
    ]}

graph = StateGraph(PriorityState)
graph.add_node("high", agent_high_confidence)
graph.add_node("medium", agent_medium_confidence)
graph.add_node("low", agent_low_confidence)

graph.add_edge(START, "high")
graph.add_edge(START, "medium")
graph.add_edge(START, "low")
graph.add_edge("high", END)
graph.add_edge("medium", END)
graph.add_edge("low", END)

app = graph.compile()
result = app.invoke({"topic": "computación cuántica", "results": []})

print(f"Total results (capped at 5): {len(result['results'])}")
for r in result["results"]:
    print(f"  [{r['confidence']:.2f}] {r['data']}")
# Output:
# Total results (capped at 5): 5
#   [0.95] Hallazgo preciso sobre computación cuántica
#   [0.88] Dato verificado de computación cuántica
#   [0.65] Información parcial sobre computación cuántica
#   [0.55] Referencia general de computación cuántica
#   [0.30] Dato no verificado sobre computación cuántica
# (El "Rumor" con 0.15 fue descartado — solo se mantienen los top 5)

Resumen

En esta cápsula aprendiste:

• El diseño de estado es la decisión más difícil en multi-agent — determina si tu sistema funciona o crea caos
• La analogía con microservicios aclara los trade-offs: shared DB = shared state (peligroso), cada servicio con su DB = isolated state (seguro), APIs entre servicios = message passing
• Shared state (todos ven todo) funciona para sistemas pequeños y secuenciales, pero produce conflictos en ejecución paralela
• Isolated state (cada agente tiene su scope) elimina conflictos pero requiere mapping explícito entre agentes
• Hybrid state (compartido para coordinación, aislado para trabajo) es el patrón recomendado para producción
• Los reducers (operator.add, custom) son la herramienta principal para prevenir conflictos de sobrescritura
• La comunicación entre agentes es message passing implícito a través del estado: un agente escribe, otro lee
• Diseñar el estado es diseñar contratos entre agentes: qué recibe cada uno, qué produce, cómo se resuelven conflictos

Próxima cápsula: Proyecto del módulo — integrar supervisor, router, handoffs, y estado hybrid en un sistema multi-agente completo.

Recursos adicionales

1. State Management — LangGraph Docs — Referencia oficial de estado y reducers
2. Multi-Agent Architectures — LangGraph Docs — Patrones de comunicación entre agentes
3. How to pass private state between nodes — Guía oficial de estado privado
4. Subgraphs — LangGraph Docs — Documentación de subgrafos con estado aislado
5. Reducers — LangGraph Docs — Referencia detallada de reducers y Annotated
6. How to add and use subgraphs — Tutorial práctico de subgrafos
7. State Schema — LangGraph How-tos — Cómo definir esquemas de estado para grafos complejos

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