6. Shared vs Isolated State

Descripción

Ya conoces los 4 patterns de orquestación multi-agente: Supervisor coordina centralmente, Handoffs transfiere control entre agentes, Subagents delega con contexto aislado, Router clasifica y dirige. Cada pattern define quién decide qué agente trabaja. Pero hay una decisión de diseño aún más fundamental que el pattern de orquestación: cómo se comunican los agentes entre sí. Es la decisión que más impacta la calidad, escalabilidad y debuggeabilidad de tu sistema.

La comunicación entre agentes se reduce a una pregunta: ¿qué información ve cada agente? En un extremo tienes shared state — todos los agentes leen y escriben el mismo estado. Implementación simple, coordinación trivial, pero cada agente ve todo lo que producen los demás, incluyendo información que no necesita. En el otro extremo tienes isolated state — cada agente tiene su propio estado privado. Contexto limpio, sin interferencia, pero coordinar entre agentes requiere trabajo explícito. Entre ambos extremos hay variantes: message passing, shared memory con namespaces, y el approach híbrido que vas a usar en el proyecto.

Esta cápsula no es sobre implementar un pattern más. Es sobre tomar la decisión arquitectónica que determina si tu sistema multi-agente escala o colapsa. En producción, el 80% de los problemas de calidad en sistemas multi-agente no vienen de agentes mal configurados — vienen de comunicación mal diseñada.

Shared State

La idea: todos ven todo

Shared state es el approach más directo. Defines un TypedDict con todos los campos que cualquier agente necesita, y todos los nodos del grafo leen y escriben en ese mismo estado:

import operator
from typing import Annotated, TypedDict
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages

class SharedResearchState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    research_data: str
    analysis_result: str
    report_draft: str
    quality_score: float
    iteration_count: int
    sources_found: list[str]
    current_agent: str

Cada agente accede a todo. Mira el researcher — su trabajo es buscar información, pero lee campos que no necesita:

def researcher_node(state: SharedResearchState) -> dict:
    query = state["messages"][-1].content
    previous_analysis = state.get("analysis_result", "")
    sources = state.get("sources_found", [])

    response = model.invoke([
        SystemMessage(content=f"""Eres un researcher especializado.
Busca información sobre: {query}
Análisis previo: {previous_analysis}
Fuentes ya encontradas: {sources}"""),
        *state["messages"],
    ])
    return {"messages": [response], "research_data": response.content}

El analyst hace lo mismo — ve report_draft e iteration_count aunque no los necesita para analizar datos. Y el writer ve research_data, quality_score, y todo el historial de messages de los otros agentes.

Ventajas de shared state

Coordinación trivial. Cuando el analyst necesita los datos del researcher, simplemente lee state["research_data"]. No hay que pasar mensajes, no hay serialización, no hay protocolos. El dato está ahí.

Debugging simple. En cualquier punto del flujo, inspeccionas un solo objeto — el state — y ves todo el sistema.

Implementación rápida. Un TypedDict, unos nodos, unas edges. Es lo primero que implementas cuando estás prototipando:

from langgraph.graph import StateGraph, START, END

graph = StateGraph(SharedResearchState)
graph.add_node("researcher", researcher_node)
graph.add_node("analyst", analyst_node)
graph.add_node("writer", writer_node)
graph.add_edge(START, "researcher")
graph.add_edge("researcher", "analyst")
graph.add_edge("analyst", "writer")
graph.add_edge("writer", END)

app = graph.compile()

El problema: context bloat

El problema aparece cuando el sistema crece. Cada iteración acumula más datos en el state. Los messages crecen. Los campos intermedios se llenan. Y cada agente recibe todo esto en su prompt:

Iteración 1:
  researcher ve: query original (5 tokens de contexto útil)
  messages: 3 mensajes

Iteración 3:
  researcher ve: query + análisis previo + borrador + score + fuentes
  messages: 15 mensajes (incluye outputs del analyst y writer que no necesita)

Iteración 8:
  researcher ve: todo el historial acumulado
  messages: 60+ mensajes
  El 90% del contexto es ruido para la tarea de búsqueda

Esto causa tres problemas concretos:

1. Degradación de calidad. El LLM pierde foco con información irrelevante. El researcher empieza a "responder" al análisis del analyst en lugar de buscar información nueva.

2. Costo innecesario. Cada LLM call incluye tokens que el agente no necesita. Con 4 agentes y 5 iteraciones, son 20 calls con historial completo.

3. Interferencia entre agentes. Un agente lee campos que otro dejó en estado inconsistente. Bug clásico: el researcher ve report_draft y empieza a buscar información para "mejorar el reporte" en lugar de buscar sobre el tema original.

Cuándo usar shared state

Shared state funciona bien en sistemas pequeños con flujos cortos:

• Prototipos y MVPs. Cuando estás validando una idea, shared state te permite iterar rápido sin diseñar interfaces entre agentes.
• Pipelines lineales cortos. Si tu sistema es researcher → analyst → writer sin loops, el contexto no crece de forma descontrolada.
• Equipos de 2-3 agentes. Con pocos agentes, la interferencia es manejable. Con 6+, se vuelve caótica.
• Dependencias fuertes entre agentes. Si el analyst necesita exactamente lo que el researcher produce y el writer necesita ambos, shared state evita duplicación.

Isolated State

La idea: cada agente en su burbuja

Isolated state invierte la premisa: cada agente tiene su propio TypedDict con solo los campos que necesita. No ve el historial de otros, no accede a sus resultados, no conoce el estado global. Ya viste esto en la cápsula de Subagents — ahora lo formalizamos como pattern de comunicación.

class ResearcherState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    search_query: str
    max_sources: int

class AnalystState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    raw_data: str
    analysis_type: str

class WriterState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    key_points: list[str]
    target_length: int
    tone: str

Cada agente es un subgraph independiente con su propio state:

from langgraph.graph import StateGraph, START, END

def build_researcher():
    def search(state: ResearcherState) -> dict:
        response = model.invoke([
            SystemMessage(content=f"Busca información sobre: {state['search_query']}"),
            *state["messages"],
        ])
        return {"messages": [response]}

    g = StateGraph(ResearcherState)
    g.add_node("search", search)
    g.add_edge(START, "search")
    g.add_edge("search", END)
    return g.compile()

researcher = build_researcher()
analyst = build_analyst()  # mismo pattern con AnalystState

El parent como traductor

Con isolated state, el parent se encarga de traducir entre agentes — extrae lo relevante del resultado de uno y lo empaqueta como input del siguiente:

from langchain_core.messages import HumanMessage

class OrchestratorState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    research_output: str
    analysis_output: str

def delegate_research(state: OrchestratorState) -> dict:
    query = state["messages"][-1].content
    result = researcher.invoke({
        "messages": [HumanMessage(content=query)],
        "search_query": query, "max_sources": 5,
    })
    return {"research_output": result["messages"][-1].content}

def delegate_analysis(state: OrchestratorState) -> dict:
    result = analyst.invoke({
        "messages": [HumanMessage(content="Analiza estos datos")],
        "raw_data": state["research_output"], "analysis_type": "comparative",
    })
    return {"analysis_output": result["messages"][-1].content}

Cada agente recibe exactamente lo que necesita, nada más.

Ventajas de isolated state

Contexto constante. No importa si el sistema lleva 1 iteración o 50. Cada agente recibe la misma cantidad de contexto.

Sin interferencia. El researcher no puede leer report_draft porque ese campo no existe en ResearcherState. Imposición estructural, no disciplina.

Testing independiente. Puedes testear cada agente en aislamiento con inputs conocidos. Si falla, el bug está aislado.

Escalabilidad. Agregar un agente nuevo no afecta a los existentes. Defines su TypedDict, implementas su subgraph, y agregas la lógica de traducción en el parent.

El costo: coordinación explícita

La desventaja es que toda la comunicación pasa por el parent. El parent se convierte en un "traductor" que debe saber qué output produce cada agente, qué input necesita el siguiente, y cómo transformar uno en otro. Si cambias el formato de output del researcher, tienes que actualizar la lógica de traducción del parent. Con 2 agentes es trivial. Con 8, es un punto de fallo significativo — el parent está acoplado a las interfaces de todos los children.

Message Passing

Más allá de shared state y aislamiento total

Message passing es un approach intermedio donde los agentes no comparten estado directo, pero se comunican a través de mensajes estructurados. Cada agente tiene un "inbox" (lo que recibe) y un "outbox" (lo que produce). La comunicación es explícita y tipada.

La diferencia con shared state es sutil pero importante: en shared state, el analyst lee state["research_data"] directamente. En message passing, el researcher envía un mensaje con su resultado, y el analyst recibe ese mensaje. El researcher decide qué incluir en el mensaje. El analyst solo ve mensajes dirigidos a él.

Implementación con mensajes tipados

from dataclasses import dataclass

@dataclass
class AgentMessage:
    sender: str
    receiver: str
    content: str
    message_type: str  # "result", "request", "feedback"

class MessagePassingState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    agent_inbox: dict[str, list[AgentMessage]]
    current_agent: str

def send_message(state: dict, sender: str, receiver: str,
                 content: str, msg_type: str = "result") -> dict:
    inbox = state.get("agent_inbox", {})
    inbox.setdefault(receiver, []).append(
        AgentMessage(sender=sender, receiver=receiver,
                     content=content, message_type=msg_type))
    return {"agent_inbox": inbox}

def get_inbox(state: dict, agent_name: str) -> list[AgentMessage]:
    return state.get("agent_inbox", {}).get(agent_name, [])

Nodos con inbox/outbox

Cada agente lee su inbox, procesa, y envía al siguiente:

def researcher_with_messages(state: MessagePassingState) -> dict:
    my_inbox = get_inbox(state, "researcher")
    requests = [m.content for m in my_inbox if m.message_type == "request"]
    query = requests[-1] if requests else state["messages"][-1].content

    response = model.invoke([
        SystemMessage(content="Eres un researcher. Busca información relevante."),
        HumanMessage(content=query),
    ])

    updated = send_message(state, "researcher", "analyst", response.content, "result")
    return {"agent_inbox": updated["agent_inbox"], "messages": [response]}

El analyst hace lo mismo: lee mensajes del researcher en su inbox, analiza, y envía al writer. Cada agente solo ve los mensajes dirigidos a él — no hay broadcast.

Cuándo message passing tiene sentido

Message passing brilla cuando necesitas comunicación selectiva — el researcher envía al analyst pero no al writer — o feedback loops donde el supervisor pide más datos al researcher sin contaminar el inbox del writer.

┌──────────┐  result   ┌──────────┐  result   ┌──────────┐
│Researcher├──────────►│ Analyst  ├──────────►│  Writer  │
└────▲─────┘           └──────────┘           └──────────┘
     │ request
     │
┌────┴─────┐
│Supervisor│
└──────────┘

Shared Memory con LangGraph Store

Memoria compartida sin context bloat

En el Módulo 6 aprendiste el Store API de LangGraph: InMemoryStore con namespaces para guardar conocimiento persistente entre sesiones. Ese mismo mecanismo sirve para comunicación entre agentes — pero con una ventaja sobre shared state: cada agente accede solo a los namespaces que le corresponden.

La idea: los agentes comparten un Store con namespaces organizados. El researcher escribe en su namespace, el analyst lee del namespace del researcher y escribe en el suyo propio.

from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore
from langchain_core.runnables import RunnableConfig

store = InMemoryStore()

Namespaces por agente

El researcher escribe en su namespace, el analyst lee del namespace del researcher:

def researcher_with_store(state: dict, config: RunnableConfig, *, store: BaseStore):
    task_id = config["configurable"].get("task_id", "default")
    response = model.invoke([
        SystemMessage(content="Busca información relevante."), *state["messages"],
    ])
    store.put(("tasks", task_id, "researcher"), "results",
              {"data": response.content, "sources": ["source1", "source2"]})
    return {"messages": [response]}

def analyst_with_store(state: dict, config: RunnableConfig, *, store: BaseStore):
    task_id = config["configurable"].get("task_id", "default")
    research = store.get(("tasks", task_id, "researcher"), "results")
    if not research:
        return {"messages": [AIMessage(content="Sin datos de investigación.")]}

    response = model.invoke([
        SystemMessage(content=f"Analiza: {research.value['data']}"),
        HumanMessage(content="Produce un análisis estructurado."),
    ])
    store.put(("tasks", task_id, "analyst"), "analysis",
              {"result": response.content, "based_on": research.value.get("sources", [])})
    return {"messages": [response]}

El writer seguiría el mismo pattern: lee de ambos namespaces (researcher y analyst). Cada agente accede solo a los namespaces que explícitamente busca — no hay broadcast.

Ventajas sobre shared state puro

1. Acceso selectivo. El researcher no puede leer lo que el writer produce. No hay broadcast — cada agente busca explícitamente lo que necesita.
2. Persistencia. El Store sobrevive entre invocaciones. Si el sistema falla a mitad de ejecución, los resultados parciales están guardados.
3. Metadata rica. Timestamps, versiones, scores de calidad junto con los datos. El analyst puede decidir si necesita re-búsqueda.
4. Auditoría. store.search(("tasks", task_id)) muestra todo lo que cada agente produjo, en qué orden, con qué metadata.

Compilación con Store

El grafo se compila pasando el store, igual que aprendiste en M6:

app = graph.compile(store=store, checkpointer=MemorySaver())
result = app.invoke(
    {"messages": [HumanMessage(content="Investiga AI agents")]},
    config={"configurable": {"thread_id": "t1", "task_id": "research-001"}},
)

Hybrid Approaches

El pattern recomendado

Ningún sistema real usa shared state puro o isolated state puro. Los sistemas en producción usan un approach híbrido: estado compartido para coordinación de alto nivel, estado aislado para ejecución. El supervisor necesita saber quién ha trabajado y el quality score (coordinación → shared). El researcher no necesita ver el borrador del writer (ejecución → isolated).

Implementación: coordination state + execution subgraphs

class CoordinationState(TypedDict):
    """Estado compartido — solo coordinación, no datos de ejecución."""
    messages: Annotated[list[AnyMessage], add_messages]
    current_phase: str
    assigned_agent: str
    iteration_count: int
    quality_score: float
    phase_results: dict[str, str]  # {"researcher": "resumen", "analyst": "resumen"}

class ResearcherExecState(TypedDict):
    """Estado aislado del researcher — solo lo que necesita para buscar."""
    messages: Annotated[list[AnyMessage], add_messages]
    search_query: str
    found_sources: list[str]

class AnalystExecState(TypedDict):
    """Estado aislado del analyst — solo datos y análisis."""
    messages: Annotated[list[AnyMessage], add_messages]
    raw_data: str
    analysis_framework: str

El supervisor trabaja con CoordinationState. Cuando delega a un agente, extrae lo necesario del coordination state, invoca el subgraph del agente con su propio state, y pone el resumen de vuelta en coordination state:

researcher_graph = build_researcher()  # subgraph con ResearcherExecState
analyst_graph = build_analyst()        # subgraph con AnalystExecState

def supervisor_delegate(state: CoordinationState) -> dict:
    agent = state["assigned_agent"]
    if agent == "researcher":
        query = state["messages"][-1].content
        result = researcher_graph.invoke({
            "messages": [HumanMessage(content=query)],
            "search_query": query, "found_sources": [],
        })
        summary = result["messages"][-1].content[:500]
    elif agent == "analyst":
        research_data = state.get("phase_results", {}).get("researcher", "")
        result = analyst_graph.invoke({
            "messages": [HumanMessage(content="Analiza estos datos")],
            "raw_data": research_data, "analysis_framework": "comparative",
        })
        summary = result["messages"][-1].content[:500]
    else:
        return {}

    return {
        "phase_results": {**state.get("phase_results", {}), agent: summary},
        "messages": [AIMessage(content=f"[{agent.title()}] {summary}")],
    }

def supervisor_decide(state: CoordinationState) -> dict:
    phases_done = list(state.get("phase_results", {}).keys())
    response = model.invoke([
        SystemMessage(content=f"""Fases completadas: {phases_done}.
Calidad: {state.get('quality_score', 0)}.
Decide: researcher, analyst, writer, o FINISH."""),
        *state["messages"][-6:],
    ])
    next_agent = response.content.strip().lower()
    if next_agent == "finish":
        return {"current_phase": "done", "assigned_agent": "none"}
    return {"assigned_agent": next_agent, "iteration_count": state.get("iteration_count", 0) + 1}

Por qué este approach funciona

El coordination state se mantiene pequeño: fase actual, agente asignado, contadores, y resúmenes (no datos completos) de cada agente. Los datos completos viven en el ResearcherExecState y mueren cuando ese subgraph termina:

┌─────────────────────────────────────────────────────┐
│  COORDINATION STATE (pequeño, solo coordinación)     │
│  current_phase: "analysis"  │  quality_score: 0.7   │
│  phase_results: {"researcher": "resumen 500 chars"}  │
│                                                      │
│    ┌──────────────┐      ┌──────────────┐            │
│    │  Researcher  │      │   Analyst    │            │
│    │  ExecState   │      │  ExecState   │            │
│    │  (aislado)   │      │  (aislado)   │            │
│    │ search_query │      │ raw_data     │            │
│    └──────────────┘      └──────────────┘            │
└─────────────────────────────────────────────────────┘

Combinación con Store

Puedes añadir el Store API al approach híbrido: el supervisor_delegate guarda el output completo en el Store (para que el siguiente agente pueda acceder a datos detallados) y pone solo un resumen en phase_results (para que el supervisor tome decisiones rápidas). Tres capas complementarias:

• Coordination state → resúmenes para decisiones rápidas del supervisor
• Store → datos completos persistentes para agentes que los necesiten
• Execution subgraphs → aislamiento con contexto mínimo

Comparación

Regla práctica

• Prototipo / 2-3 agentes lineales → Shared state
• 4+ agentes / múltiples iteraciones → Hybrid
• Feedback loops → Message passing o hybrid
• Producción con persistencia → Hybrid + Store
• Tareas 100% independientes → Isolated state (subagents)

Conexión con Proyecto

El Research Agent v5 de este módulo usa el approach híbrido:

• Coordination state del Supervisor: current_phase, assigned_agent, iteration_count, quality_score, phase_results (resúmenes). Estado pequeño y enfocado para decisiones de routing.

• Isolated execution de cada worker: el Researcher tiene ResearcherExecState con search_query y found_sources. El Analyst tiene AnalystExecState con raw_data. El Writer tiene WriterExecState con key_points. Ninguno ve el estado de los demás.

• Store para datos detallados: resultados completos en namespaces ("research", task_id, agent_name). El Supervisor lee resúmenes para decidir; cuando delega, extrae datos completos del Store.

El Researcher en la iteración 10 recibe exactamente el mismo volumen de contexto que en la iteración 1.

Troubleshooting

Problema 1: Context window del supervisor crece sin control

Síntoma: El supervisor tarda cada vez más en responder. La calidad de sus decisiones de routing se degrada en iteraciones tardías.

Causa: Aunque los workers tienen estado aislado, el coordination state acumula messages con cada iteración. Si usas add_messages, los mensajes de todas las rondas se concatenan.

Solución: Limita los mensajes del supervisor a los últimos N: state["messages"][-6:]. Usa phase_results con resúmenes en lugar de guardar outputs completos en messages.

Problema 2: El subgraph no recibe el state que espera

Síntoma: KeyError o valores None cuando el subgraph intenta acceder a campos de su estado.

Causa: El parent no inicializa todos los campos requeridos del TypedDict del child.

Solución: Pasa todos los campos con valores por defecto al invocar el subgraph. No olvides inicializar listas como [] y strings como "".

Problema 3: El Store no tiene datos cuando el agente los busca

Síntoma: store.get() retorna None aunque el agente anterior debería haber escrito datos.

Causa: El namespace o la key no coinciden exactamente. Typo en el namespace, o el task_id del config es diferente entre invocaciones.

Solución: Centraliza la construcción de namespaces en una función helper agent_namespace(task_id, agent_name) que retorne ("tasks", task_id, agent_name). Usa esa función tanto para store.put() como para store.get() — así eliminas typos.

Problema 4: Mensajes del inbox no se limpian entre iteraciones

Síntoma: En message passing, el analyst procesa mensajes de iteraciones anteriores además de la actual.

Causa: El inbox acumula mensajes sin timestamp ni mecanismo de "ya leído".

Solución: Agrega un campo read al AgentMessage y filtra por not m.read, o limpia el inbox al inicio de cada iteración. Alternativa: usa timestamps y filtra mensajes posteriores a la última lectura.

Problema 5: El approach híbrido duplica datos

Síntoma: El resumen en phase_results dice una cosa, pero los datos completos en el Store dicen otra.

Causa: El resumen se genera en un momento y los datos del Store se actualizan después (o viceversa).

Solución: Genera el resumen del mismo output que guardas en el Store. Una sola fuente de verdad: full_output = result["messages"][-1].content, luego store.put(...) con full_output y phase_results con full_output[:500].

Ejercicios

Ejercicio 1: Diagnosticar context bloat

Tienes este sistema con shared state:

class State(TypedDict):
    messages: Annotated[list, add_messages]
    research: str
    analysis: str
    draft: str
    feedback: str
    quality: float

def researcher(state):
    response = model.invoke([
        SystemMessage(content=f"""Busca información.
Draft actual: {state.get('draft', '')}
Feedback: {state.get('feedback', '')}
Calidad: {state.get('quality', 0)}"""),
        *state["messages"],
    ])
    return {"messages": [response], "research": response.content}

¿Qué campos no debería ver el researcher? ¿Cómo lo corregirías?

class ResearcherState(TypedDict):
    messages: Annotated[list, add_messages]
    search_query: str

def researcher(state: ResearcherState):
    response = model.invoke([
        SystemMessage(content=f"Busca información sobre: {state['search_query']}"),
        *state["messages"],
    ])
    return {"messages": [response]}

Ejercicio 2: Implementar message passing con feedback

Implementa un sistema donde el supervisor pueda enviar un mensaje de tipo "feedback" al researcher pidiendo más datos, sin que ese feedback llegue al writer.

def supervisor_feedback(state: FeedbackState) -> dict:
    inbox = state.get("agent_inbox", {})
    inbox.setdefault("researcher", []).append(AgentMessage(
        sender="supervisor", receiver="researcher",
        content="Necesito más fuentes académicas, no solo blogs.",
        message_type="feedback",
    ))
    return {"agent_inbox": inbox}

def researcher_node(state: FeedbackState) -> dict:
    my_inbox = state.get("agent_inbox", {}).get("researcher", [])
    feedback = [m.content for m in my_inbox if m.message_type == "feedback"]

    prompt = "Busca información relevante."
    if feedback:
        prompt += f"\nFeedback del supervisor: {'; '.join(feedback)}"

    response = model.invoke([SystemMessage(content=prompt), *state["messages"]])

    inbox = state.get("agent_inbox", {})
    inbox.setdefault("analyst", []).append(AgentMessage(
        sender="researcher", receiver="analyst",
        content=response.content, message_type="result",
    ))
    return {"messages": [response], "agent_inbox": inbox}

Ejercicio 3: Shared memory con Store

Implementa un sistema donde el researcher escribe resultados en el Store con namespace por tarea, y el analyst lee solo los resultados más recientes.

def researcher_store(state: dict, config: RunnableConfig, *, store: BaseStore):
    task_id = config["configurable"]["task_id"]
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    response = model.invoke([
        SystemMessage(content="Busca información relevante."), *state["messages"],
    ])
    store.put(("tasks", task_id, "researcher"), f"results_{timestamp}",
              {"data": response.content, "timestamp": timestamp})
    return {"messages": [response]}

def analyst_store(state: dict, config: RunnableConfig, *, store: BaseStore):
    task_id = config["configurable"]["task_id"]
    all_results = store.search(("tasks", task_id, "researcher"))
    if not all_results:
        return {"messages": [AIMessage(content="Sin datos.")]}

    latest = sorted(all_results, key=lambda r: r.value.get("timestamp", ""), reverse=True)[0]
    response = model.invoke([
        SystemMessage(content=f"Analiza: {latest.value['data']}"),
        HumanMessage(content="Produce un análisis."),
    ])
    return {"messages": [response]}

Ejercicio 4: Diseñar estado híbrido

Diseña el TypedDict de coordination state y los execution states para un sistema de 4 agentes: researcher, analyst, writer, reviewer. El reviewer evalúa el draft y puede pedir al researcher que busque más datos o al writer que reescriba.

class CoordinationState(TypedDict):
    messages: Annotated[list, add_messages]
    current_phase: Literal["research", "analysis", "writing", "review", "done"]
    assigned_agent: str
    iteration_count: int
    quality_score: float
    phase_results: dict[str, str]
    reviewer_feedback: str  # para que el supervisor sepa qué pidió el reviewer

class ResearcherExec(TypedDict):
    messages: Annotated[list, add_messages]
    search_query: str
    previous_feedback: str  # feedback del reviewer si hay re-búsqueda

class AnalystExec(TypedDict):
    messages: Annotated[list, add_messages]
    raw_data: str

class WriterExec(TypedDict):
    messages: Annotated[list, add_messages]
    key_points: list[str]
    rewrite_instructions: str  # instrucciones del reviewer si hay reescritura

class ReviewerExec(TypedDict):
    messages: Annotated[list, add_messages]
    draft_to_review: str
    evaluation_criteria: list[str]

Ejercicio 5: Migrar de shared a hybrid

Tienes este sistema con shared state puro que funciona pero sufre de context bloat en la iteración 5+:

class FullSharedState(TypedDict):
    messages: Annotated[list, add_messages]
    research_data: str
    analysis: str
    draft: str
    quality: float
    iteration: int
    sources: list[str]
    feedback_history: list[str]

Migralo a un approach híbrido manteniendo el mismo flujo funcional.

class CoordState(TypedDict):
    messages: Annotated[list, add_messages]
    quality: float
    iteration: int
    phase_summaries: dict[str, str]
    current_agent: str

class ResearchExec(TypedDict):
    messages: Annotated[list, add_messages]
    query: str
    previous_sources: list[str]

class AnalysisExec(TypedDict):
    messages: Annotated[list, add_messages]
    data: str

class WritingExec(TypedDict):
    messages: Annotated[list, add_messages]
    points: list[str]
    feedback: str

def delegate(state: CoordState) -> dict:
    agent = state["current_agent"]
    sub = {"researcher": researcher_sub, "analyst": analyst_sub, "writer": writer_sub}[agent]

    if agent == "researcher":
        invoke_args = {"messages": [HumanMessage(content=state["messages"][-1].content)],
                       "query": state["messages"][-1].content, "previous_sources": []}
    elif agent == "analyst":
        invoke_args = {"messages": [HumanMessage(content="Analiza")],
                       "data": state.get("phase_summaries", {}).get("researcher", "")}
    else:
        invoke_args = {"messages": [HumanMessage(content="Escribe el reporte")],
                       "points": [state.get("phase_summaries", {}).get("analyst", "")],
                       "feedback": state["messages"][-1].content if state["iteration"] > 1 else ""}

    result = sub.invoke(invoke_args)
    output = result["messages"][-1].content
    return {"phase_summaries": {**state.get("phase_summaries", {}), agent: output[:500]}}

Resumen

En esta cápsula aprendiste:

• Shared state — un TypedDict compartido donde todos leen y escriben. Simple pero causa context bloat: cada agente ve información que no necesita, degradando calidad e incrementando costos.

• Isolated state — cada agente tiene su propio TypedDict y trabaja en un subgraph independiente. Contexto constante, sin interferencia. El parent debe actuar como traductor entre agentes.

• Message passing — comunicación selectiva con mensajes dirigidos, no broadcast. Útil para feedback loops y comunicación no-lineal.

• LangGraph Store — memoria compartida con namespaces por agente. Persiste entre invocaciones y permite auditoría.

• Hybrid — el approach recomendado: coordination state compartido + execution states aislados. Combina coordinación simple con contexto limpio.

• La decisión de comunicación entre agentes es la más importante del diseño. Shared state puro funciona en demos pero colapsa en producción. Isolated puro es limpio pero difícil de coordinar. El híbrido es el sweet spot.

Próxima cápsula: Orquestación avanzada — composición de patterns (Router + Supervisor + Subagents), parallel execution, y cómo combinar todo lo que has aprendido en este módulo para el proyecto final.

Recursos Adicionales

1. LangGraph State Management — Documentación oficial sobre cómo funciona el state en LangGraph, reducers, y TypedDict
2. LangGraph Multi-Agent — Communication — Patterns de comunicación entre agentes: shared state vs message passing
3. LangGraph Store API — Store para memoria compartida entre agentes con namespaces
4. LangGraph Subgraphs — Cómo implementar subgraphs con state mapping para isolated execution
5. Multi-Agent Architectures (LangChain Blog) — Análisis de patterns de orquestación y decisiones de diseño de estado