7. Subgraphs como Agent Modules

Descripción

En las cápsulas anteriores construiste agentes como grafos: nodos, edges, conditional routing, ciclos, Functional API. Todo vivía en un solo StateGraph. Mientras tu agente tenía 4-5 nodos, eso funcionaba. Pero piensa en lo que viene: en M8 vas a construir un sistema multi-agente donde un supervisor orquesta agentes especializados — un agente de research, uno de análisis, uno de síntesis. Si cada agente es un bloque de 15 nodos aplanados en un solo grafo, tu state machine se convierte en un monstruo inmanejable.

Subgraphs resuelven este problema. Un subgraph es un StateGraph completo — con sus propios nodos, edges, estado, y lógica — que se usa como nodo dentro de otro grafo. Es la misma idea que las funciones en programación: encapsulas una capacidad completa detrás de una interfaz clara. La diferencia es que aquí no es una función — es un grafo entero con su propio flujo, sus propios ciclos, y su propia lógica de routing.

Esta cápsula es el puente directo hacia M8 (Multi-Agent Orchestration). El pattern parent.add_node("agent_name", compiled_subgraph) es exactamente cómo se implementan agentes en un sistema multi-agente. Si dominas subgraphs aquí, M8 será una extensión natural.

El Problema: Grafos Monolíticos

Cómo se ve un grafo que creció demasiado

Imagina que evolucionas el Research Agent. Empezaste con 4 nodos. Ahora research necesita buscar en web, bases de datos, y validar fuentes. Analysis necesita extraer entidades, clasificar, y evaluar. Aplanado: 10+ nodos, 15+ conditional edges, 1 TypedDict con 20+ campos.

Los problemas concretos

1. Estado contaminado. Todos los nodos comparten un solo TypedDict. source_validity que solo importa a research también lo ve synthesis. Un bug en uno corrompe datos de otro.

2. No puedes testear aislado. Para probar que el análisis funciona, necesitas levantar el grafo completo.

3. No puedes reutilizar. El módulo de análisis está acoplado al estado global y a los edges del grafo.

4. Routing explosion. Con 10 nodos y conditional edges, debugging se vuelve rastrear qué camino tomó entre docenas de posibilidades.

La solución: dividir y componer

En vez de un grafo monolítico, construyes 3 subgraphs de 3-4 nodos cada uno:

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│   RESEARCH   │     │   ANALYSIS   │     │  SYNTHESIS   │
│   subgraph   │     │   subgraph   │     │   subgraph   │
│              │     │              │     │              │
│ search →     │     │ extract →    │     │ draft →      │
│ validate →   │     │ classify →   │     │ review →     │
│ collect      │     │ evaluate     │     │ format       │
└──────┬───────┘     └──────┬───────┘     └──────┬───────┘
       └────────────────────┼────────────────────┘
                   ┌────────▼────────┐
                   │  PARENT GRAPH   │
                   │ plan → research │
                   │  → analysis     │
                   │  → synthesis    │
                   └─────────────────┘

Cada subgraph tiene su propio estado, su propia lógica, y su propia interfaz. El parent graph solo sabe que existen, no cómo funcionan internamente.

Subgraphs: Grafos dentro de Grafos

El concepto

Un subgraph es un StateGraph que: (1) se define con su propio estado tipado, (2) tiene sus propios nodos y edges, (3) se compila con .compile(), y (4) se usa como nodo en otro grafo con add_node(). LangGraph trata un subgraph compilado como cualquier otro nodo.

Tu primer subgraph

from dotenv import load_dotenv
load_dotenv()

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

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

class ResearchState(TypedDict):
    messages: Annotated[list, add_messages]

def search_node(state: ResearchState) -> dict:
    response = model.invoke(
        [SystemMessage(content="Busca información relevante sobre el tema.")]
        + state["messages"]
    )
    return {"messages": [response]}

def process_node(state: ResearchState) -> dict:
    response = model.invoke(
        [SystemMessage(content="Extrae los 3 puntos clave de la investigación.")]
        + state["messages"]
    )
    return {"messages": [response]}

research_graph = StateGraph(ResearchState)
research_graph.add_node("search", search_node)
research_graph.add_node("process", process_node)
research_graph.add_edge(START, "search")
research_graph.add_edge("search", "process")
research_graph.add_edge("process", END)

research_subgraph = research_graph.compile()

research_subgraph es un grafo completamente funcional. Puedes invocarlo directamente:

result = research_subgraph.invoke({
    "messages": [HumanMessage(content="¿Qué es RAG?")]
})
print(result["messages"][-1].content)

Usarlo como nodo en un parent graph

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

def planning_node(state: AgentState) -> dict:
    response = model.invoke(
        [SystemMessage(content="Genera un plan de investigación con 2 sub-preguntas.")]
        + state["messages"]
    )
    return {"messages": [response]}

def synthesis_node(state: AgentState) -> dict:
    response = model.invoke(
        [SystemMessage(content="Sintetiza toda la información en una respuesta final.")]
        + state["messages"]
    )
    return {"messages": [response]}

parent_graph = StateGraph(AgentState)
parent_graph.add_node("plan", planning_node)
parent_graph.add_node("research", research_subgraph)  # Subgraph como nodo
parent_graph.add_node("synthesize", synthesis_node)

parent_graph.add_edge(START, "plan")
parent_graph.add_edge("plan", "research")
parent_graph.add_edge("research", "synthesize")
parent_graph.add_edge("synthesize", END)

agent = parent_graph.compile()
result = agent.invoke({
    "messages": [HumanMessage(content="¿Cuáles son los beneficios de RAG?")]
})
print(result["messages"][-1].content)

Para el parent graph, "research" es un nodo como cualquier otro. No sabe que internamente tiene search → process.

Visualización

from IPython.display import Image, display

display(Image(agent.get_graph().draw_mermaid_png()))

display(Image(agent.get_graph(xray=True).draw_mermaid_png()))

Sin xray, el diagrama muestra plan → research → synthesize. Con xray=True, LangGraph expande los subgraphs y muestra sus nodos internos. Obtienes ambas vistas: arquitectura de alto nivel y detalle de implementación.

State Mapping entre Parent y Subgraph

El mecanismo: claves compartidas

Cuando el parent invoca un subgraph, LangGraph mapea el estado usando claves con el mismo nombre. Si el parent tiene messages y el subgraph tiene messages, el valor fluye entre ellos.

Parent State                    Subgraph State
┌─────────────────┐            ┌─────────────────┐
│ messages ───────│────────────│─── messages      │
│ plan            │            │                  │
│ iteration_count │            │                  │
└─────────────────┘            └─────────────────┘

El subgraph solo ve las claves que comparte. Las claves que no comparte (plan, iteration_count) son invisibles. Esto es exactamente el aislamiento que buscamos.

Claves compartidas vs privadas

class ParentState(TypedDict):
    messages: Annotated[list, add_messages]
    plan: list[str]
    iteration_count: int

class ResearchState(TypedDict):
    messages: Annotated[list, add_messages]  # Compartida — mismo nombre y tipo

class AnalysisState(TypedDict):
    messages: Annotated[list, add_messages]  # Compartida
    analysis_depth: int                      # Solo del subgraph — privada

Cuando el parent invoca el subgraph de research: messages se pasa (clave compartida), plan e iteration_count no se pasan (no existen en ResearchState). Cuando el subgraph termina, su messages actualizado regresa al parent.

Input/output explícito con funciones wrapper

Cuando los estados difieren significativamente, usas una función wrapper:

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

class AnalysisSubState(TypedDict):
    messages: Annotated[list, add_messages]
    input_text: str

def analysis_wrapper(state: ParentState) -> dict:
    sub_result = analysis_subgraph.invoke({
        "messages": state["messages"],
        "input_text": state.get("raw_data", ""),
    })
    return {"messages": sub_result["messages"],
            "raw_data": sub_result["messages"][-1].content}

El wrapper traduce raw_data → input_text en la entrada, dando control total sobre el mapping cuando las interfaces no coinciden.

Diseñando Módulos para el Research Agent

Tres módulos, tres responsabilidades

Research Module (con loop reason-act)

from langchain_core.tools import tool
from langchain_core.messages import ToolMessage

@tool
def search_web(query: str) -> str:
    """Busca información en la web."""
    return f"Resultados para '{query}': datos relevantes encontrados."

class ResearchModuleState(TypedDict):
    messages: Annotated[list, add_messages]

research_tools = [search_web]
research_model = model.bind_tools(research_tools)
tools_by_name = {t.name: t for t in research_tools}

def research_reason(state: ResearchModuleState) -> dict:
    response = research_model.invoke(
        [SystemMessage(content="Investiga el tema. Usa herramientas. "
                       "Cuando tengas suficiente, responde directamente.")]
        + state["messages"]
    )
    return {"messages": [response]}

def research_tools_node(state: ResearchModuleState) -> dict:
    last = state["messages"][-1]
    results = []
    for tc in last.tool_calls:
        result = str(tools_by_name[tc["name"]].invoke(tc["args"]))
        results.append(ToolMessage(content=result, tool_call_id=tc["id"]))
    return {"messages": results}

def research_route(state: ResearchModuleState) -> str:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return "done"

rg = StateGraph(ResearchModuleState)
rg.add_node("reason", research_reason)
rg.add_node("tools", research_tools_node)
rg.add_edge(START, "reason")
rg.add_conditional_edges("reason", research_route, {"tools": "tools", "done": END})
rg.add_edge("tools", "reason")
research_module = rg.compile()

Analysis y Synthesis Modules

El mismo pattern — estado propio, nodos internos, compile, listo:

class AnalysisModuleState(TypedDict):
    messages: Annotated[list, add_messages]

def extract_key_points(state: AnalysisModuleState) -> dict:
    return {"messages": [model.invoke(
        [SystemMessage(content="Extrae los 5 puntos clave.")] + state["messages"]
    )]}

def evaluate_quality(state: AnalysisModuleState) -> dict:
    return {"messages": [model.invoke(
        [SystemMessage(content="Evalúa la calidad. ¿Suficiente? ¿Qué falta?")]
        + state["messages"]
    )]}

ag = StateGraph(AnalysisModuleState)
ag.add_node("extract", extract_key_points)
ag.add_node("evaluate", evaluate_quality)
ag.add_edge(START, "extract")
ag.add_edge("extract", "evaluate")
ag.add_edge("evaluate", END)
analysis_module = ag.compile()

class SynthesisModuleState(TypedDict):
    messages: Annotated[list, add_messages]

def generate_draft(state: SynthesisModuleState) -> dict:
    return {"messages": [model.invoke(
        [SystemMessage(content="Genera respuesta completa y estructurada.")]
        + state["messages"]
    )]}

def refine_response(state: SynthesisModuleState) -> dict:
    return {"messages": [model.invoke(
        [SystemMessage(content="Revisa: mejora claridad, elimina redundancias.")]
        + state["messages"]
    )]}

syg = StateGraph(SynthesisModuleState)
syg.add_node("draft", generate_draft)
syg.add_node("refine", refine_response)
syg.add_edge(START, "draft")
syg.add_edge("draft", "refine")
syg.add_edge("refine", END)
synthesis_module = syg.compile()

Testing aislado

Cada módulo se testea por separado sin levantar el agente completo:

test_msg = [HumanMessage(content="Explica retrieval-augmented generation")]

r1 = research_module.invoke({"messages": test_msg})
r2 = analysis_module.invoke({"messages": r1["messages"]})
r3 = synthesis_module.invoke({"messages": r2["messages"]})
print(r3["messages"][-1].content[:200])

Composing Subgraphs

Parent graph con routing condicional

Ahora unimos los tres módulos con un parent graph que incluye un quality check y ciclo de re-investigación:

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    quality_score: float
    cycle_count: int

def planning_node(state: AgentState) -> dict:
    response = model.invoke(
        [SystemMessage(content="Analiza la pregunta y genera un plan con sub-preguntas.")]
        + state["messages"]
    )
    return {"messages": [response]}

def quality_check_node(state: AgentState) -> dict:
    response = model.invoke(
        [SystemMessage(content="Evalúa calidad 0.0-1.0. Responde SOLO el número.")]
        + state["messages"]
    )
    try:
        score = float(response.content.strip())
    except ValueError:
        score = 0.5
    return {
        "quality_score": score,
        "cycle_count": state.get("cycle_count", 0) + 1,
    }

def route_by_quality(state: AgentState) -> str:
    if state.get("cycle_count", 0) >= 3:
        return "synthesis"
    if state.get("quality_score", 0.0) >= 0.7:
        return "synthesis"
    return "research"

parent = StateGraph(AgentState)
parent.add_node("plan", planning_node)
parent.add_node("research", research_module)      # Subgraph
parent.add_node("analysis", analysis_module)       # Subgraph
parent.add_node("quality_check", quality_check_node)
parent.add_node("synthesis", synthesis_module)     # Subgraph

parent.add_edge(START, "plan")
parent.add_edge("plan", "research")
parent.add_edge("research", "analysis")
parent.add_edge("analysis", "quality_check")
parent.add_conditional_edges("quality_check", route_by_quality, {
    "research": "research",
    "synthesis": "synthesis",
})
parent.add_edge("synthesis", END)

agent = parent.compile()

result = agent.invoke({
    "messages": [HumanMessage(content="¿Cómo funcionan los AI agents con LangGraph?")],
    "quality_score": 0.0,
    "cycle_count": 0,
})
print(result["messages"][-1].content)

El parent tiene un ciclo: research → analysis → quality_check → (research si calidad baja). Cada iteración invoca subgraphs completos. El research subgraph internamente puede hacer 3-4 iteraciones de su propio loop reason-tools, pero el parent no lo sabe ni le importa.

from IPython.display import Image, display
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))

Con xray=True ves la arquitectura completa: el parent con su ciclo quality-driven, y dentro de cada subgraph sus nodos internos.

Subgraphs vs Funciones Normales

Una función normal también encapsula lógica. ¿Por qué molestarse con un subgraph?

Criterios de decisión

Reglas prácticas

Usa una función normal cuando:

1. La lógica es una sola llamada al modelo o una secuencia de 2 pasos sin ciclos
2. No necesitas checkpointing intermedio
3. No vas a reutilizar esa lógica en otros agentes

Usa un subgraph cuando:

1. La lógica tiene ciclos internos (loop reason-act propio)
2. Necesitas checkpointing por paso (producción, tolerancia a fallos)
3. El módulo será reutilizado en otros agentes o en M8 como agente independiente
4. El módulo tiene 3+ nodos con routing entre ellos

La pregunta clave: "¿Este módulo es un mini-agente con su propio flujo, o es un paso simple?" Mini-agente → Subgraph. Paso simple → Función normal.

Conexión con Proyecto

En el proyecto de este módulo (cápsula 08, Research Agent State Machine), vas a construir el Research Agent como un StateGraph con nodos de planning, research, analysis, y synthesis. Ahora sabes que puedes implementar research y analysis como subgraphs. El proyecto te deja elegir: grafo plano o composición con subgraphs.

En módulos posteriores:

• M5 (Planning): El módulo de planning puede ser un subgraph con su propio ciclo plan → evaluate → re-plan.
• M6 (Memory): Cada subgraph puede tener su propio checkpointer. Research persiste su progreso independientemente del parent.
• M7 (MCP): El subgraph de research puede integrar MCP tools internamente sin que el parent lo sepa.
• M8 (Multi-Agent): Cada agente es un subgraph compilado. El supervisor es el parent graph. parent.add_node("researcher", researcher_subgraph) es literal — es el código. Lo que aprendiste aquí se aplica directamente.

Troubleshooting

Problema 1: El subgraph no recibe el estado del parent

Causa: Las claves del estado no coinciden. LangGraph mapea por nombre de clave.

# ❌ Nombres diferentes — no se mapea
class ParentState(TypedDict):
    messages: Annotated[list, add_messages]
class SubState(TypedDict):
    msgs: Annotated[list, add_messages]

# ✅ Mismo nombre
class SubState(TypedDict):
    messages: Annotated[list, add_messages]

Problema 2: El estado del parent no se actualiza después del subgraph

Causa: El subgraph retorna claves que no existen en el parent state. Solo las claves compartidas se sincronizan. Asegúrate de que las claves de output del subgraph existan en ParentState con el mismo reducer.

Problema 3: TypeError al compilar el parent con un subgraph

Causa: Pasaste el StateGraph sin compilar.

# ❌ Sin compilar
parent.add_node("research", research_graph)

# ✅ Compilado
parent.add_node("research", research_graph.compile())

Problema 4: draw_mermaid_png no muestra nodos internos del subgraph

Causa: Por defecto, subgraphs son cajas opacas.

agent.get_graph().draw_mermaid_png()            # Solo muestra "research" como bloque
agent.get_graph(xray=True).draw_mermaid_png()   # Muestra nodos internos

Problema 5: El subgraph entra en loop infinito

Causa: El subgraph tiene un ciclo sin stop condition. El parent no puede interrumpirlo — siempre incluye un límite de iteraciones dentro del subgraph con un campo internal_iterations en su estado.

Ejercicios

Ejercicio 1: Subgraph básico como nodo (Fácil)

Crea un subgraph de traducción con dos nodos (detect_language → translate). Compílalo y úsalo como nodo en un parent graph: process_input → translate_subgraph → format_output.

class TranslateState(TypedDict):
    messages: Annotated[list, add_messages]

def detect_language(state: TranslateState) -> dict:
    return {"messages": [model.invoke(
        [SystemMessage(content="Detecta el idioma.")] + state["messages"])]}
def translate(state: TranslateState) -> dict:
    return {"messages": [model.invoke(
        [SystemMessage(content="Traduce al inglés.")] + state["messages"])]}

tg = StateGraph(TranslateState)
tg.add_node("detect", detect_language)
tg.add_node("translate", translate)
tg.add_edge(START, "detect")
tg.add_edge("detect", "translate")
tg.add_edge("translate", END)
translate_module = tg.compile()

class ParentState(TypedDict):
    messages: Annotated[list, add_messages]

parent = StateGraph(ParentState)
parent.add_node("process", lambda s: {"messages": [model.invoke(
    [SystemMessage(content="Limpia el texto.")] + s["messages"])]})
parent.add_node("translate", translate_module)
parent.add_node("format", lambda s: {"messages": [model.invoke(
    [SystemMessage(content="Formatea la traducción.")] + s["messages"])]})
parent.add_edge(START, "process")
parent.add_edge("process", "translate")
parent.add_edge("translate", "format")
parent.add_edge("format", END)
agent = parent.compile()

result = agent.invoke({"messages": [HumanMessage(content="Hola, ¿cómo estás?")]})
print(result["messages"][-1].content)

Ejercicio 2: Subgraph con ciclo interno (Medio)

Crea un subgraph de research con un loop reason-act interno (máximo 3 iteraciones, usa search_web). Insértalo en un parent: plan → research_subgraph → summarize.

class RState(TypedDict):
    messages: Annotated[list, add_messages]
    iterations: int

model_with_tools = model.bind_tools([search_web])

def reason(state: RState) -> dict:
    return {"messages": [model_with_tools.invoke(
        [SystemMessage(content="Investiga. Responde cuando tengas suficiente.")]
        + state["messages"])], "iterations": state.get("iterations", 0) + 1}

def exec_tools(state: RState) -> dict:
    last = state["messages"][-1]
    return {"messages": [ToolMessage(content=str(tools_by_name[tc["name"]].invoke(
        tc["args"])), tool_call_id=tc["id"]) for tc in last.tool_calls]}

def should_continue(state: RState) -> str:
    if state.get("iterations", 0) >= 3: return "done"
    last = state["messages"][-1]
    return "tools" if hasattr(last, "tool_calls") and last.tool_calls else "done"

rg = StateGraph(RState)
rg.add_node("reason", reason)
rg.add_node("tools", exec_tools)
rg.add_edge(START, "reason")
rg.add_conditional_edges("reason", should_continue, {"tools": "tools", "done": END})
rg.add_edge("tools", "reason")
research_sub = rg.compile()

class PState(TypedDict):
    messages: Annotated[list, add_messages]

parent = StateGraph(PState)
parent.add_node("plan", lambda s: {"messages": [model.invoke(
    [SystemMessage(content="Genera un plan breve.")] + s["messages"])]})
parent.add_node("research", research_sub)
parent.add_node("summarize", lambda s: {"messages": [model.invoke(
    [SystemMessage(content="Resume en 3 puntos.")] + s["messages"])]})
parent.add_edge(START, "plan")
parent.add_edge("plan", "research")
parent.add_edge("research", "summarize")
parent.add_edge("summarize", END)
agent = parent.compile()

result = agent.invoke({"messages": [HumanMessage(content="¿Qué es vector search?")]})

Ejercicio 3: State mapping con wrapper (Medio)

Crea un parent con AgentState(messages, raw_data) y un subgraph con ProcessState(messages, input_text). Implementa un wrapper que traduzca raw_data → input_text al invocar el subgraph.

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

class ProcessState(TypedDict):
    messages: Annotated[list, add_messages]
    input_text: str

def process_text(state: ProcessState) -> dict:
    return {"messages": [model.invoke(
        [SystemMessage(content=f"Procesa: {state.get('input_text', '')}")]
        + state["messages"])]}

pg = StateGraph(ProcessState)
pg.add_node("process", process_text)
pg.add_edge(START, "process")
pg.add_edge("process", END)
process_sub = pg.compile()

def process_wrapper(state: AgentState) -> dict:
    sub_result = process_sub.invoke({
        "messages": state["messages"],
        "input_text": state.get("raw_data", ""),
    })
    return {"messages": sub_result["messages"],
            "raw_data": sub_result["messages"][-1].content}

parent = StateGraph(AgentState)
parent.add_node("collect", lambda s: {"raw_data": "AI agents combinan LLMs con tools."})
parent.add_node("process", process_wrapper)
parent.add_edge(START, "collect")
parent.add_edge("collect", "process")
parent.add_edge("process", END)
agent = parent.compile()
result = agent.invoke({"messages": [HumanMessage(content="Analiza")], "raw_data": ""})
print(result["raw_data"][:200])

Ejercicio 4: Tres subgraphs con routing condicional (Difícil)

Implementa research, analysis, y synthesis como subgraphs. El parent tiene quality_check después de analysis. Si quality_score < 0.7, vuelve a research (máximo 3 ciclos).

class QState(TypedDict):
    messages: Annotated[list, add_messages]
    quality_score: float
    cycle_count: int

# Usa research_module, analysis_module, synthesis_module definidos antes

def quality_check(state: QState) -> dict:
    response = model.invoke(
        [SystemMessage(content="Evalúa calidad 0.0-1.0. SOLO el número.")]
        + state["messages"])
    try: score = float(response.content.strip())
    except ValueError: score = 0.5
    return {"quality_score": score, "cycle_count": state.get("cycle_count", 0) + 1}

def quality_route(state: QState) -> str:
    if state.get("cycle_count", 0) >= 3: return "synthesis"
    return "synthesis" if state.get("quality_score", 0.0) >= 0.7 else "research"

parent = StateGraph(QState)
parent.add_node("research", research_module)
parent.add_node("analysis", analysis_module)
parent.add_node("quality_check", quality_check)
parent.add_node("synthesis", synthesis_module)
parent.add_edge(START, "research")
parent.add_edge("research", "analysis")
parent.add_edge("analysis", "quality_check")
parent.add_conditional_edges("quality_check", quality_route, {
    "research": "research", "synthesis": "synthesis"})
parent.add_edge("synthesis", END)

agent = parent.compile()
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))
result = agent.invoke({"messages": [HumanMessage(content="Explica transformers")],
    "quality_score": 0.0, "cycle_count": 0})
print(f"Ciclos: {result['cycle_count']}, Score: {result['quality_score']:.2f}")

Ejercicio 5: Reutilizar un subgraph en dos agentes (Difícil)

Toma el analysis_module y úsalo en dos parent graphs diferentes: uno research → analysis → synthesis, y otro user_input → analysis → response. Demuestra que el mismo subgraph compilado funciona en ambos contextos.

class SharedState(TypedDict):
    messages: Annotated[list, add_messages]

def analyze(state): return {"messages": [model.invoke(
    [SystemMessage(content="Analiza: temas, sentimiento, puntos clave.")] + state["messages"])]}
def rate(state): return {"messages": [model.invoke(
    [SystemMessage(content="Califica importancia: alta, media, baja.")] + state["messages"])]}

sag = StateGraph(SharedState)
sag.add_node("analyze", analyze)
sag.add_node("rate", rate)
sag.add_edge(START, "analyze")
sag.add_edge("analyze", "rate")
sag.add_edge("rate", END)
shared_analysis = sag.compile()

# Agent 1: Research Pipeline — research → analysis → synthesis
p1 = StateGraph(SharedState)
p1.add_node("research", lambda s: {"messages": [model.invoke(
    [SystemMessage(content="Investiga.")] + s["messages"])]})
p1.add_node("analysis", shared_analysis)
p1.add_node("synth", lambda s: {"messages": [model.invoke(
    [SystemMessage(content="Sintetiza.")] + s["messages"])]})
p1.add_edge(START, "research")
p1.add_edge("research", "analysis")
p1.add_edge("analysis", "synth")
p1.add_edge("synth", END)
agent1 = p1.compile()

# Agent 2: Feedback Pipeline — analysis → response
p2 = StateGraph(SharedState)
p2.add_node("analysis", shared_analysis)
p2.add_node("respond", lambda s: {"messages": [model.invoke(
    [SystemMessage(content="Da una recomendación.")] + s["messages"])]})
p2.add_edge(START, "analysis")
p2.add_edge("analysis", "respond")
p2.add_edge("respond", END)
agent2 = p2.compile()

r1 = agent1.invoke({"messages": [HumanMessage(content="¿Qué es fine-tuning?")]})
r2 = agent2.invoke({"messages": [HumanMessage(content="Baja retención.")]})
print("Agent 1:", r1["messages"][-1].content[:100])
print("Agent 2:", r2["messages"][-1].content[:100])

Resumen

En esta cápsula aprendiste:

• Grafos monolíticos se vuelven inmanejables cuando un agente crece. Estado contaminado, imposibilidad de testear aislado, y routing explosion son los síntomas.
• Subgraphs son StateGraphs compilados que se usan como nodos en un parent graph. Cada subgraph tiene su propio estado, nodos, edges, y lógica de routing.
• State mapping funciona por claves con el mismo nombre. Las claves compartidas fluyen entre parent y subgraph; las privadas quedan aisladas. Para mappings complejos, usas funciones wrapper.
• Composición modular: El Research Agent se descompone en research, analysis, y synthesis como subgraphs independientes que se testean, reutilizan, y reemplazan sin acoplamiento.
• Subgraph vs función normal: Ciclos, routing, o checkpointing → subgraph. Llamada simple → función normal.
• xray=True en draw_mermaid_png revela los nodos internos de cada subgraph.
• Subgraphs = agentes en M8. parent.add_node("agent_name", compiled_subgraph) es exactamente cómo implementarás multi-agent orchestration. Cada agente es un subgraph. El supervisor es el parent.

Próxima cápsula: Proyecto — Research Agent con State Machine. Pondrás en práctica todo el módulo: StateGraph, typed state, ciclos, routing, y subgraphs para construir la v1 del agente que evolucionará hasta M10.

Recursos Adicionales

1. LangGraph Subgraphs — Conceptual Guide — Documentación oficial sobre subgraphs
2. LangGraph Subgraphs — How-to Guide — Tutorial paso a paso para crear y componer subgraphs
3. LangGraph State Management — State mapping entre grafos
4. LangGraph Visualization — draw_mermaid_png y xray=True
5. LangGraph Multi-Agent — Subgraph Pattern — Preview de subgraphs en M8