Subgraphs: Composición Modular

Descripción de la cápsula

Tu patrón "buscar + parsear + resumir" se repite para cada fuente. Tienes los mismos 3 pasos (fetch → parse → summarize) duplicados 3 veces — una vez para web, otra para academic, otra para news. Si necesitas cambiar la lógica de parsing, tienes que modificar 3 lugares. Si agregas una cuarta fuente, copias y pegas los mismos 3 nodos otra vez.

¿Y si pudieras definir ese pipeline una sola vez y reutilizarlo? Así como una función puede llamar a otra función, un grafo puede contener otro grafo. Encapsulas lógica compleja en un grafo independiente y lo usas como nodo dentro de un grafo más grande. Eso es un subgraph.

Los subgraphs son el equivalente de funciones en programación modular: encapsulan, abstraen y permiten reutilización. La diferencia con una función regular es que un subgraph te da checkpointing por cada paso interno, visualización del flujo, y streaming — cosas que una función Python normal no ofrece.

El problema: lógica duplicada

Observa este grafo con 3 pipelines de búsqueda idénticos en estructura:

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    query: str
    results: Annotated[list[str], operator.add]

def fetch_web(state: State) -> dict:
    return {"results": [f"[fetch_web] Raw data from web for '{state['query']}'"]}

def parse_web(state: State) -> dict:
    raw = state["results"][-1]
    return {"results": [f"[parse_web] Parsed: {raw}"]}

def summarize_web(state: State) -> dict:
    parsed = state["results"][-1]
    return {"results": [f"[summarize_web] Summary: {parsed}"]}

def fetch_academic(state: State) -> dict:
    return {"results": [f"[fetch_academic] Raw data from academic for '{state['query']}'"]}

def parse_academic(state: State) -> dict:
    raw = state["results"][-1]
    return {"results": [f"[parse_academic] Parsed: {raw}"]}

def summarize_academic(state: State) -> dict:
    parsed = state["results"][-1]
    return {"results": [f"[summarize_academic] Summary: {parsed}"]}

def fetch_news(state: State) -> dict:
    return {"results": [f"[fetch_news] Raw data from news for '{state['query']}'"]}

def parse_news(state: State) -> dict:
    raw = state["results"][-1]
    return {"results": [f"[parse_news] Parsed: {raw}"]}

def summarize_news(state: State) -> dict:
    parsed = state["results"][-1]
    return {"results": [f"[summarize_news] Summary: {parsed}"]}

graph = StateGraph(State)

graph.add_node("fetch_web", fetch_web)
graph.add_node("parse_web", parse_web)
graph.add_node("summarize_web", summarize_web)
graph.add_node("fetch_academic", fetch_academic)
graph.add_node("parse_academic", parse_academic)
graph.add_node("summarize_academic", summarize_academic)
graph.add_node("fetch_news", fetch_news)
graph.add_node("parse_news", parse_news)
graph.add_node("summarize_news", summarize_news)

graph.add_edge(START, "fetch_web")
graph.add_edge("fetch_web", "parse_web")
graph.add_edge("parse_web", "summarize_web")
graph.add_edge("summarize_web", "fetch_academic")
graph.add_edge("fetch_academic", "parse_academic")
graph.add_edge("parse_academic", "summarize_academic")
graph.add_edge("summarize_academic", "fetch_news")
graph.add_edge("fetch_news", "parse_news")
graph.add_edge("parse_news", "summarize_news")
graph.add_edge("summarize_news", END)

app = graph.compile()
result = app.invoke({"query": "Python 3.12", "results": []})
print(f"Nodos totales: 9")
print(f"Resultados: {len(result['results'])}")
# Output esperado:
# Nodos totales: 9
# Resultados: 9

9 nodos para 3 pipelines que hacen lo mismo con datos diferentes. Si necesitas agregar validación entre parse y summarize, tienes que tocar 3 lugares. Si necesitas una cuarta fuente, copias 3 nodos más. Esto no escala.

Subgraphs = funciones: la analogía fundamental

Piensa en funciones de Python:

def process_source(source_name: str, query: str) -> str:
    raw = fetch(source_name, query)
    parsed = parse(raw)
    summary = summarize(parsed)
    return summary

result_web = process_source("web", "Python 3.12")
result_academic = process_source("academic", "Python 3.12")
result_news = process_source("news", "Python 3.12")

Defines la lógica una vez y la llamas 3 veces con diferentes argumentos. Un subgraph hace exactamente lo mismo, pero con las ventajas de LangGraph:

Crear un subgraph: define un StateGraph completo

Un subgraph es un StateGraph normal — lo defines, le agregas nodos y edges, y lo compilas. La única diferencia es que después lo usas como nodo dentro de otro grafo:

from dotenv import load_dotenv
load_dotenv()

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

class SearchPipelineState(TypedDict):
    source_name: str
    query: str
    raw_data: str
    parsed_data: str
    summary: str

def fetch(state: SearchPipelineState) -> dict:
    return {"raw_data": f"Raw data from {state['source_name']} for '{state['query']}'"}

def parse(state: SearchPipelineState) -> dict:
    return {"parsed_data": f"Parsed: {state['raw_data']}"}

def summarize(state: SearchPipelineState) -> dict:
    return {"summary": f"Summary of {state['source_name']}: {state['parsed_data'][:50]}..."}

search_pipeline = StateGraph(SearchPipelineState)
search_pipeline.add_node("fetch", fetch)
search_pipeline.add_node("parse", parse)
search_pipeline.add_node("summarize", summarize)

search_pipeline.add_edge(START, "fetch")
search_pipeline.add_edge("fetch", "parse")
search_pipeline.add_edge("parse", "summarize")
search_pipeline.add_edge("summarize", END)

search_app = search_pipeline.compile()

result = search_app.invoke({
    "source_name": "web",
    "query": "Python 3.12",
    "raw_data": "",
    "parsed_data": "",
    "summary": "",
})
print(result["summary"])
# Output esperado:
# Summary of web: Parsed: Raw data from web for 'Python 3.12'...

Este grafo funciona de forma independiente. Tiene su propio estado (SearchPipelineState), sus propios nodos, y se puede invocar directamente. Ahora lo vamos a usar como nodo de un grafo padre.

Usar un subgraph como nodo

Para usar un subgraph compilado como nodo, lo pasas directamente a add_node:

from dotenv import load_dotenv
load_dotenv()

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

class SearchPipelineState(TypedDict):
    source_name: str
    query: str
    raw_data: str
    parsed_data: str
    summary: str

def fetch(state: SearchPipelineState) -> dict:
    return {"raw_data": f"Raw data from {state['source_name']} for '{state['query']}'"}

def parse(state: SearchPipelineState) -> dict:
    return {"parsed_data": f"Parsed: {state['raw_data']}"}

def summarize(state: SearchPipelineState) -> dict:
    return {"summary": f"Summary of {state['source_name']}: {state['parsed_data'][:50]}..."}

search_pipeline = StateGraph(SearchPipelineState)
search_pipeline.add_node("fetch", fetch)
search_pipeline.add_node("parse", parse)
search_pipeline.add_node("summarize", summarize)
search_pipeline.add_edge(START, "fetch")
search_pipeline.add_edge("fetch", "parse")
search_pipeline.add_edge("parse", "summarize")
search_pipeline.add_edge("summarize", END)

search_subgraph = search_pipeline.compile()

class ParentState(TypedDict):
    query: str
    source_name: str
    raw_data: str
    parsed_data: str
    summary: str
    all_summaries: Annotated[list[str], operator.add]

def collect_summary(state: ParentState) -> dict:
    return {"all_summaries": [state["summary"]]}

parent = StateGraph(ParentState)
parent.add_node("search_pipeline", search_subgraph)
parent.add_node("collect", collect_summary)

parent.add_edge(START, "search_pipeline")
parent.add_edge("search_pipeline", "collect")
parent.add_edge("collect", END)

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

result = app.invoke({
    "query": "LangGraph subgraphs",
    "source_name": "web",
    "raw_data": "",
    "parsed_data": "",
    "summary": "",
    "all_summaries": [],
})
print(f"Summary: {result['summary']}")
print(f"All summaries: {result['all_summaries']}")
# Output esperado:
# Summary: Summary of web: Parsed: Raw data from web for 'LangGraph subgrap...
# All summaries: ['Summary of web: Parsed: Raw data from web for \'LangGraph subgrap...']

parent.add_node("search_pipeline", search_subgraph) — el grafo compilado se usa directamente como nodo. LangGraph se encarga de ejecutar todos los nodos internos del subgraph cuando llega el turno de ese "nodo" en el grafo padre.

Interfaz de estado entre padre y subgraph

Cuando el padre y el subgraph comparten los mismos campos en su TypedDict, LangGraph mapea el estado automáticamente. Los campos que tienen el mismo nombre se pasan directamente:

class SearchPipelineState(TypedDict):
    source_name: str    # ← compartido con padre
    query: str          # ← compartido con padre
    raw_data: str       # interno del subgraph
    parsed_data: str    # interno del subgraph
    summary: str        # ← compartido con padre

class ParentState(TypedDict):
    query: str          # ← compartido con subgraph
    source_name: str    # ← compartido con subgraph
    summary: str        # ← compartido con subgraph
    all_summaries: Annotated[list[str], operator.add]  # solo del padre

Los campos que existen en ambos (query, source_name, summary) se comparten. Los campos que solo existen en el subgraph (raw_data, parsed_data) son internos — el padre no los ve. Los campos que solo existen en el padre (all_summaries) no afectan al subgraph.

Regla de mapeo

Estado diferente: subgraph con su propio TypedDict

A veces el subgraph necesita un estado completamente diferente al del padre. En ese caso, usas una función wrapper que traduce entre estados:

from dotenv import load_dotenv
load_dotenv()

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

class AnalysisState(TypedDict):
    text: str
    word_count: int
    sentiment: str

def count_words(state: AnalysisState) -> dict:
    return {"word_count": len(state["text"].split())}

def detect_sentiment(state: AnalysisState) -> dict:
    text = state["text"].lower()
    if any(w in text for w in ["great", "excellent", "amazing", "good"]):
        return {"sentiment": "positive"}
    elif any(w in text for w in ["bad", "terrible", "awful", "poor"]):
        return {"sentiment": "negative"}
    return {"sentiment": "neutral"}

analysis_graph = StateGraph(AnalysisState)
analysis_graph.add_node("count", count_words)
analysis_graph.add_node("sentiment", detect_sentiment)
analysis_graph.add_edge(START, "count")
analysis_graph.add_edge("count", "sentiment")
analysis_graph.add_edge("sentiment", END)
analysis_subgraph = analysis_graph.compile()

class ParentState(TypedDict):
    query: str
    search_result: str
    analysis: dict
    summaries: Annotated[list[str], operator.add]

def search(state: ParentState) -> dict:
    return {"search_result": f"Great results about '{state['query']}': Python is an amazing language."}

def analyze_wrapper(state: ParentState) -> dict:
    analysis_input = {"text": state["search_result"], "word_count": 0, "sentiment": ""}
    result = analysis_subgraph.invoke(analysis_input)
    return {"analysis": {"word_count": result["word_count"], "sentiment": result["sentiment"]}}

def report(state: ParentState) -> dict:
    a = state["analysis"]
    return {"summaries": [f"Analysis: {a['word_count']} words, sentiment: {a['sentiment']}"]}

parent = StateGraph(ParentState)
parent.add_node("search", search)
parent.add_node("analyze", analyze_wrapper)
parent.add_node("report", report)

parent.add_edge(START, "search")
parent.add_edge("search", "analyze")
parent.add_edge("analyze", "report")
parent.add_edge("report", END)

app = parent.compile()
result = app.invoke({"query": "Python", "search_result": "", "analysis": {}, "summaries": []})
print(f"Result: {result['analysis']}")
print(f"Report: {result['summaries']}")
# Output esperado:
# Result: {'word_count': 9, 'sentiment': 'positive'}
# Report: ["Analysis: 9 words, sentiment: positive"]

La función analyze_wrapper traduce del estado del padre al estado del subgraph, invoca el subgraph, y traduce el resultado de vuelta. Es la misma idea que un adapter pattern — un puente entre dos interfaces diferentes.

Cuándo usar estados compartidos vs diferentes

Reutilización: un subgraph usado múltiples veces

El verdadero poder de los subgraphs es que los defines una vez y los reutilizas. Aquí usamos el mismo pipeline de búsqueda para 3 fuentes diferentes:

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 IPython.display import Image, display

class SearchState(TypedDict):
    source_name: str
    query: str
    raw_data: str
    parsed_data: str
    summary: str

def fetch_data(state: SearchState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Simula datos crudos de la fuente '{state['source_name']}' sobre '{state['query']}'. "
        f"Genera 2-3 oraciones de datos sin procesar."
    )
    return {"raw_data": response.content}

def parse_data(state: SearchState) -> dict:
    return {"parsed_data": f"[{state['source_name']}] {state['raw_data'][:200]}"}

def summarize_data(state: SearchState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Resume en 1 oración: {state['parsed_data']}")
    return {"summary": response.content}

search_pipeline = StateGraph(SearchState)
search_pipeline.add_node("fetch", fetch_data)
search_pipeline.add_node("parse", parse_data)
search_pipeline.add_node("summarize", summarize_data)
search_pipeline.add_edge(START, "fetch")
search_pipeline.add_edge("fetch", "parse")
search_pipeline.add_edge("parse", "summarize")
search_pipeline.add_edge("summarize", END)
search_subgraph = search_pipeline.compile()

class OrchestratorState(TypedDict):
    query: str
    source_name: str
    raw_data: str
    parsed_data: str
    summary: str
    all_summaries: Annotated[list[dict], operator.add]

def set_source(name: str):
    def node(state: OrchestratorState) -> dict:
        return {"source_name": name}
    return node

def collect_result(state: OrchestratorState) -> dict:
    return {"all_summaries": [{"source": state["source_name"], "summary": state["summary"]}]}

parent = StateGraph(OrchestratorState)

parent.add_node("set_web", set_source("web"))
parent.add_node("pipeline_web", search_subgraph)
parent.add_node("collect_web", collect_result)

parent.add_node("set_academic", set_source("academic"))
parent.add_node("pipeline_academic", search_subgraph)
parent.add_node("collect_academic", collect_result)

parent.add_node("set_news", set_source("news"))
parent.add_node("pipeline_news", search_subgraph)
parent.add_node("collect_news", collect_result)

parent.add_edge(START, "set_web")
parent.add_edge("set_web", "pipeline_web")
parent.add_edge("pipeline_web", "collect_web")

parent.add_edge(START, "set_academic")
parent.add_edge("set_academic", "pipeline_academic")
parent.add_edge("pipeline_academic", "collect_academic")

parent.add_edge(START, "set_news")
parent.add_edge("set_news", "pipeline_news")
parent.add_edge("pipeline_news", "collect_news")

parent.add_edge("collect_web", END)
parent.add_edge("collect_academic", END)
parent.add_edge("collect_news", END)

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

result = app.invoke({
    "query": "RAG techniques",
    "source_name": "", "raw_data": "", "parsed_data": "", "summary": "",
    "all_summaries": [],
})
print(f"Fuentes procesadas: {len(result['all_summaries'])}")
for s in result["all_summaries"]:
    print(f"  [{s['source']}] {s['summary'][:80]}...")
# Output esperado:
# Fuentes procesadas: 3
#   [web] RAG combina recuperación de documentos con generación de texto...
#   [academic] Los estudios muestran que RAG mejora la precisión factual...
#   [news] Empresas como Google y OpenAI están adoptando técnicas RAG...

Lo que ganamos

• ❌ Antes: 9 funciones diferentes (3 × fetch, parse, summarize) con lógica duplicada
• ✅ Después: 3 funciones + 1 subgraph reutilizado 3 veces
• Si cambias la lógica de parsing, la cambias en un solo lugar

El mismo search_subgraph se usa como pipeline_web, pipeline_academic y pipeline_news. Las 3 instancias comparten la misma lógica pero operan con diferentes datos gracias a set_source.

Nesting: subgraphs dentro de subgraphs

Los subgraphs pueden contener otros subgraphs. Pero mantén la profundidad máxima en 2 niveles (padre → hijo → nieto). Más niveles hacen el debugging difícil:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from IPython.display import Image, display

class InnerState(TypedDict):
    text: str
    processed: str

def clean_text(state: InnerState) -> dict:
    return {"processed": state["text"].strip().lower()}

inner = StateGraph(InnerState)
inner.add_node("clean", clean_text)
inner.add_edge(START, "clean")
inner.add_edge("clean", END)
inner_compiled = inner.compile()

class MiddleState(TypedDict):
    text: str
    processed: str
    word_count: int

def count_words(state: MiddleState) -> dict:
    return {"word_count": len(state["processed"].split())}

middle = StateGraph(MiddleState)
middle.add_node("preprocess", inner_compiled)
middle.add_node("count", count_words)
middle.add_edge(START, "preprocess")
middle.add_edge("preprocess", "count")
middle.add_edge("count", END)
middle_compiled = middle.compile()

class OuterState(TypedDict):
    text: str
    processed: str
    word_count: int
    report: str

def make_report(state: OuterState) -> dict:
    return {"report": f"Text '{state['processed']}' has {state['word_count']} words"}

outer = StateGraph(OuterState)
outer.add_node("analysis", middle_compiled)
outer.add_node("report", make_report)
outer.add_edge(START, "analysis")
outer.add_edge("analysis", "report")
outer.add_edge("report", END)

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

result = app.invoke({"text": "  Hello World  ", "processed": "", "word_count": 0, "report": ""})
print(result["report"])
# Output esperado:
# Text 'hello world' has 2 words

Tres niveles: outer → middle → inner. El flujo es: outer.analysis → middle.preprocess → inner.clean → middle.count → outer.report. Cada nivel tiene su propio estado y sus propios checkpoints.

La regla de 2 niveles

Si necesitas más de 2 niveles, probablemente tu diseño necesita refactoring: combina niveles o extrae lógica a funciones regulares.

Subgraphs con Functional API: @entrypoint como módulo reutilizable

En la Functional API, un @entrypoint compilado puede actuar como un módulo reutilizable — similar a cómo un subgraph es un grafo dentro de otro grafo:

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

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

@task
def fetch_data(source: str, query: str) -> str:
    response = model.invoke(f"Simula datos de {source} sobre '{query}'. 2-3 oraciones.")
    return response.content

@task
def parse_data(source: str, raw: str) -> str:
    return f"[{source}] {raw[:200]}"

@task
def summarize_data(parsed: str) -> str:
    response = model.invoke(f"Resume en 1 oración: {parsed}")
    return response.content

@entrypoint()
def search_pipeline(config: dict) -> dict:
    source = config["source"]
    query = config["query"]

    raw = fetch_data(source, query).result()
    parsed = parse_data(source, raw).result()
    summary = summarize_data(parsed).result()

    return {"source": source, "summary": summary}

@entrypoint()
def research_orchestrator(query: str) -> dict:
    web_fut = search_pipeline.ainvoke({"source": "web", "query": query})
    academic_fut = search_pipeline.ainvoke({"source": "academic", "query": query})
    news_fut = search_pipeline.ainvoke({"source": "news", "query": query})

    results = []
    for label, fut in [("web", web_fut), ("academic", academic_fut), ("news", news_fut)]:
        try:
            r = fut.result()
            results.append(r)
        except Exception as e:
            results.append({"source": label, "summary": f"Error: {e}"})

    return {"results": results, "total": len(results)}

result = research_orchestrator.invoke("transformer architectures")
print(f"Total: {result['total']}")
for r in result["results"]:
    print(f"  [{r['source']}] {r['summary'][:80]}...")
# Output esperado:
# Total: 3
#   [web] Transformer architectures have revolutionized natural language...
#   [academic] Research shows transformers outperform RNNs in sequence...
#   [news] Major tech companies are investing heavily in transformer-based...

El search_pipeline es un @entrypoint que encapsula la lógica de buscar-parsear-resumir. El research_orchestrator lo llama 3 veces con diferentes fuentes. Cada invocación tiene sus propios checkpoints internos.

Subgraph vs función regular: cuándo usar cada uno

La pregunta clave: ¿por qué no usar una función Python normal en vez de un subgraph?

def search_and_summarize(source: str, query: str) -> str:
    raw = fetch(source, query)
    parsed = parse(raw)
    summary = summarize(parsed)
    return summary

Esta función hace exactamente lo mismo que el subgraph. Es más simple, más corta y más fácil de entender. Entonces, ¿cuándo vale la pena usar un subgraph?

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model

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

def search_as_function(source: str, query: str) -> str:
    raw = model.invoke(f"Simula datos de {source} sobre '{query}'").content
    parsed = f"[{source}] {raw[:200]}"
    summary = model.invoke(f"Resume en 1 oración: {parsed}").content
    return summary

class SubgraphState(TypedDict):
    source_name: str
    query: str
    raw_data: str
    parsed_data: str
    summary: str

def sg_fetch(state: SubgraphState) -> dict:
    result = model.invoke(f"Simula datos de {state['source_name']} sobre '{state['query']}'")
    return {"raw_data": result.content}

def sg_parse(state: SubgraphState) -> dict:
    return {"parsed_data": f"[{state['source_name']}] {state['raw_data'][:200]}"}

def sg_summarize(state: SubgraphState) -> dict:
    result = model.invoke(f"Resume en 1 oración: {state['parsed_data']}")
    return {"summary": result.content}

search_graph = StateGraph(SubgraphState)
search_graph.add_node("fetch", sg_fetch)
search_graph.add_node("parse", sg_parse)
search_graph.add_node("summarize", sg_summarize)
search_graph.add_edge(START, "fetch")
search_graph.add_edge("fetch", "parse")
search_graph.add_edge("parse", "summarize")
search_graph.add_edge("summarize", END)
search_subgraph = search_graph.compile()

func_result = search_as_function("web", "Python 3.12")
print(f"Función: {func_result[:80]}...")

sg_result = search_subgraph.invoke({
    "source_name": "web", "query": "Python 3.12",
    "raw_data": "", "parsed_data": "", "summary": "",
})
print(f"Subgraph: {sg_result['summary'][:80]}...")
# Output esperado:
# Función: Python 3.12 incluye mejoras significativas en tipado...
# Subgraph: Python 3.12 incluye mejoras significativas en tipado...

Ambos producen el mismo resultado. La diferencia está en lo que pasa cuando algo sale mal o cuando necesitas observabilidad:

Regla de decisión

• ✅ Usa una función regular cuando la lógica es simple (1-3 pasos), no necesitas checkpointing, y no necesitas visualizar el flujo interno
• ✅ Usa un subgraph cuando la lógica es compleja (3+ pasos), necesitas recovery ante fallos, quieres visualización/streaming del flujo interno, o el pipeline se reutiliza en múltiples grafos por diferentes equipos

Ejemplo completo: Research Agent modular con subgraphs

Combinemos todo: subgraphs reutilizables, branching paralelo (de la cápsula anterior) y merge inteligente:

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 IPython.display import Image, display

class PipelineState(TypedDict):
    source_name: str
    query: str
    raw_data: str
    summary: str

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

def pipeline_fetch(state: PipelineState) -> dict:
    response = model.invoke(
        f"Simula una búsqueda en {state['source_name']} sobre '{state['query']}'. "
        f"Genera 2-3 oraciones informativas."
    )
    return {"raw_data": response.content}

def pipeline_summarize(state: PipelineState) -> dict:
    response = model.invoke(
        f"Resume en 1 oración lo siguiente de [{state['source_name']}]: {state['raw_data']}"
    )
    return {"summary": response.content}

pipeline = StateGraph(PipelineState)
pipeline.add_node("fetch", pipeline_fetch)
pipeline.add_node("summarize", pipeline_summarize)
pipeline.add_edge(START, "fetch")
pipeline.add_edge("fetch", "summarize")
pipeline.add_edge("summarize", END)
pipeline_compiled = pipeline.compile()

class ResearchState(TypedDict):
    query: str
    source_name: str
    raw_data: str
    summary: str
    all_results: Annotated[list[dict], operator.add]
    final_report: str

def set_source(name: str):
    def node(state: ResearchState) -> dict:
        return {"source_name": name, "raw_data": "", "summary": ""}
    return node

def collect(state: ResearchState) -> dict:
    return {"all_results": [{"source": state["source_name"], "summary": state["summary"]}]}

def final_merge(state: ResearchState) -> dict:
    context = "\n".join(f"- [{r['source']}]: {r['summary']}" for r in state["all_results"])
    response = model.invoke(
        f"Genera un resumen ejecutivo de 3-4 oraciones basado en estas fuentes "
        f"sobre '{state['query']}':\n\n{context}"
    )
    return {"final_report": response.content}

research = StateGraph(ResearchState)

for source in ["web", "academic", "news"]:
    research.add_node(f"set_{source}", set_source(source))
    research.add_node(f"pipeline_{source}", pipeline_compiled)
    research.add_node(f"collect_{source}", collect)

    research.add_edge(START, f"set_{source}")
    research.add_edge(f"set_{source}", f"pipeline_{source}")
    research.add_edge(f"pipeline_{source}", f"collect_{source}")
    research.add_edge(f"collect_{source}", "merge")

research.add_node("merge", final_merge)
research.add_edge("merge", END)

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

result = app.invoke({
    "query": "¿Cómo mejora RAG la precisión de los LLMs?",
    "source_name": "", "raw_data": "", "summary": "",
    "all_results": [], "final_report": "",
})

print(f"Fuentes: {len(result['all_results'])}")
for r in result["all_results"]:
    print(f"  [{r['source']}] {r['summary'][:80]}...")
print(f"\nReporte final: {result['final_report'][:200]}...")
# Output esperado:
# Fuentes: 3
#   [web] RAG combina búsqueda de documentos con generación de texto para...
#   [academic] Estudios recientes demuestran que RAG reduce las alucinaciones...
#   [news] Empresas como Google y Microsoft están integrando RAG en sus...
# Reporte final: RAG (Retrieval-Augmented Generation) ha demostrado ser una técnica...

Arquitectura del grafo

START → set_web → pipeline_web (subgraph: fetch → summarize) → collect_web ─┐
START → set_academic → pipeline_academic (subgraph: fetch → summarize) → collect_academic ─┤→ merge → END
START → set_news → pipeline_news (subgraph: fetch → summarize) → collect_news ─┘

• Subgraph: definido una vez, usado 3 veces
• Branching: las 3 ramas se ejecutan en paralelo
• Merge: recibe los 3 resultados y sintetiza con LLM

Si mañana necesitas agregar una cuarta fuente ("social_media"), agregas 3 líneas al loop for source in [...] y listo.

Troubleshooting

Problema 1: "El subgraph no recibe los campos del estado padre"

Síntoma: El subgraph recibe campos vacíos o con valores por defecto.

Causa: Los nombres de los campos en el TypedDict del subgraph no coinciden con los del padre.

Solución: Verifica que los campos compartidos tengan exactamente el mismo nombre y tipo en ambos TypedDict:

# ❌ Nombres diferentes
class ParentState(TypedDict):
    search_query: str    # "search_query"

class SubgraphState(TypedDict):
    query: str           # "query" — no coincide

# ✅ Mismos nombres
class ParentState(TypedDict):
    query: str           # "query"

class SubgraphState(TypedDict):
    query: str           # "query" — coincide

Problema 2: "El subgraph sobreescribe campos del padre que no debería tocar"

Síntoma: Después de ejecutar el subgraph, campos del padre que no están en el subgraph cambian o se pierden.

Causa: El subgraph solo puede modificar campos que comparte con el padre. Si un campo del padre no existe en el subgraph, el subgraph no lo afecta. Pero si un campo compartido tiene un reducer en el padre (operator.add) y el subgraph retorna un valor para ese campo, el reducer lo procesa.

Solución: Ten cuidado con reducers en campos compartidos. Si no quieres que el subgraph afecte un campo con reducer, no incluyas ese campo en el TypedDict del subgraph.

Problema 3: "No puedo reutilizar el mismo subgraph con diferentes configuraciones"

Síntoma: Quieres usar el mismo subgraph para "web" y "academic", pero ambas instancias reciben los mismos datos.

Causa: El subgraph lee del estado del padre, que es compartido. Si no cambias source_name antes de cada instancia, ambas leen el mismo valor.

Solución: Usa un nodo set_source antes de cada instancia del subgraph que configure los campos necesarios:

def set_source(name: str):
    def node(state):
        return {"source_name": name}
    return node

graph.add_node("set_web", set_source("web"))
graph.add_node("pipeline_web", search_subgraph)
graph.add_edge("set_web", "pipeline_web")

Problema 4: "El diagrama no muestra los nodos internos del subgraph"

Síntoma: draw_mermaid_png() muestra el subgraph como una caja opaca sin detalle interno.

Causa: Por defecto, la visualización del grafo padre muestra los subgraphs como nodos colapsados.

Solución: Usa el parámetro xray para expandir los subgraphs en la visualización:

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

Con xray=True, verás los nodos internos de cada subgraph dentro de sus respectivas cajas.

Problema 5: "El subgraph falla y no se captura el error en el padre"

Síntoma: Un error dentro del subgraph se propaga sin manejo al nivel del padre y detiene todo el workflow.

Causa: Las excepciones dentro de nodos del subgraph se propagan hacia arriba como en cualquier call stack de Python.

Solución: Si usas el subgraph como nodo directo (add_node("pipeline", subgraph)), los errores se propagan. Para capturarlos, usa un wrapper:

def safe_pipeline(state):
    try:
        result = search_subgraph.invoke({"source_name": state["source_name"], "query": state["query"], ...})
        return {"summary": result["summary"]}
    except Exception as e:
        return {"summary": f"Error en pipeline: {str(e)}"}

Ejercicios

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

Crea un subgraph que procese texto en 2 pasos: (1) limpiar (strip + lowercase), (2) contar palabras. Usa este subgraph como nodo de un grafo padre que le agrega un paso final de reporte. Visualiza ambos grafos.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from IPython.display import Image, display

class TextState(TypedDict):
    text: str
    cleaned: str
    word_count: int

def clean(state: TextState) -> dict:
    return {"cleaned": state["text"].strip().lower()}

def count(state: TextState) -> dict:
    return {"word_count": len(state["cleaned"].split())}

text_processor = StateGraph(TextState)
text_processor.add_node("clean", clean)
text_processor.add_node("count", count)
text_processor.add_edge(START, "clean")
text_processor.add_edge("clean", "count")
text_processor.add_edge("count", END)
text_subgraph = text_processor.compile()

print("=== Subgraph ===")
display(Image(text_subgraph.get_graph().draw_mermaid_png()))

class ParentState(TypedDict):
    text: str
    cleaned: str
    word_count: int
    report: str

def make_report(state: ParentState) -> dict:
    return {"report": f"'{state['cleaned']}' tiene {state['word_count']} palabras"}

parent = StateGraph(ParentState)
parent.add_node("process", text_subgraph)
parent.add_node("report", make_report)
parent.add_edge(START, "process")
parent.add_edge("process", "report")
parent.add_edge("report", END)

app = parent.compile()
print("=== Grafo padre ===")
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({"text": "  Hello World From LangGraph  ", "cleaned": "", "word_count": 0, "report": ""})
print(result["report"])
# Output esperado:
# 'hello world from langgraph' tiene 4 palabras

Ejercicio 2: Subgraph con estado diferente (Fácil)

Crea un subgraph de análisis de sentimiento con su propio TypedDict (campos: text, sentiment, confidence). El grafo padre tiene un TypedDict diferente (campos: query, search_result, analysis). Usa una función wrapper para conectarlos.

from dotenv import load_dotenv
load_dotenv()

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

class SentimentState(TypedDict):
    text: str
    sentiment: str
    confidence: float

def analyze_sentiment(state: SentimentState) -> dict:
    text = state["text"].lower()
    positive_words = ["good", "great", "excellent", "amazing", "love", "best"]
    negative_words = ["bad", "terrible", "awful", "hate", "worst", "poor"]
    pos = sum(1 for w in positive_words if w in text)
    neg = sum(1 for w in negative_words if w in text)
    total = pos + neg
    if total == 0:
        return {"sentiment": "neutral", "confidence": 0.5}
    if pos > neg:
        return {"sentiment": "positive", "confidence": round(pos / total, 2)}
    return {"sentiment": "negative", "confidence": round(neg / total, 2)}

sentiment_graph = StateGraph(SentimentState)
sentiment_graph.add_node("analyze", analyze_sentiment)
sentiment_graph.add_edge(START, "analyze")
sentiment_graph.add_edge("analyze", END)
sentiment_subgraph = sentiment_graph.compile()

class ParentState(TypedDict):
    query: str
    search_result: str
    analysis: dict

def simulate_search(state: ParentState) -> dict:
    return {"search_result": f"Great results! Python is an excellent and amazing language for AI."}

def analyze_wrapper(state: ParentState) -> dict:
    result = sentiment_subgraph.invoke({
        "text": state["search_result"],
        "sentiment": "",
        "confidence": 0.0,
    })
    return {"analysis": {"sentiment": result["sentiment"], "confidence": result["confidence"]}}

parent = StateGraph(ParentState)
parent.add_node("search", simulate_search)
parent.add_node("analyze", analyze_wrapper)
parent.add_edge(START, "search")
parent.add_edge("search", "analyze")
parent.add_edge("analyze", END)

app = parent.compile()
result = app.invoke({"query": "Python for AI", "search_result": "", "analysis": {}})
print(f"Search: {result['search_result'][:60]}...")
print(f"Analysis: {result['analysis']}")
# Output esperado:
# Search: Great results! Python is an excellent and amazing language...
# Analysis: {'sentiment': 'positive', 'confidence': 1.0}

Ejercicio 3: Subgraph reutilizado en paralelo (Medio)

Usa un mismo subgraph de "fetch + summarize" en 3 ramas paralelas (web, academic, news). Cada rama configura la fuente antes de ejecutar el subgraph. Los resultados se acumulan con operator.add. Visualiza con xray=True.

from dotenv import load_dotenv
load_dotenv()

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

class FetchState(TypedDict):
    source: str
    query: str
    raw: str
    summary: str

def fetch(state: FetchState) -> dict:
    return {"raw": f"Data from {state['source']} about '{state['query']}'"}

def summarize(state: FetchState) -> dict:
    return {"summary": f"[{state['source']}] Summary: {state['raw'][:50]}"}

fetch_graph = StateGraph(FetchState)
fetch_graph.add_node("fetch", fetch)
fetch_graph.add_node("summarize", summarize)
fetch_graph.add_edge(START, "fetch")
fetch_graph.add_edge("fetch", "summarize")
fetch_graph.add_edge("summarize", END)
fetch_subgraph = fetch_graph.compile()

class OrchestratorState(TypedDict):
    query: str
    source: str
    raw: str
    summary: str
    all_summaries: Annotated[list[str], operator.add]

def set_src(name: str):
    def node(state: OrchestratorState) -> dict:
        return {"source": name, "raw": "", "summary": ""}
    return node

def collect(state: OrchestratorState) -> dict:
    return {"all_summaries": [state["summary"]]}

graph = StateGraph(OrchestratorState)
graph.add_node("merge", lambda state: {})

for src in ["web", "academic", "news"]:
    graph.add_node(f"set_{src}", set_src(src))
    graph.add_node(f"pipe_{src}", fetch_subgraph)
    graph.add_node(f"collect_{src}", collect)
    graph.add_edge(START, f"set_{src}")
    graph.add_edge(f"set_{src}", f"pipe_{src}")
    graph.add_edge(f"pipe_{src}", f"collect_{src}")
    graph.add_edge(f"collect_{src}", "merge")

graph.add_edge("merge", END)

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

result = app.invoke({
    "query": "LangGraph subgraphs", "source": "", "raw": "", "summary": "",
    "all_summaries": [],
})
print(f"Summaries: {len(result['all_summaries'])}")
for s in result["all_summaries"]:
    print(f"  {s}")
# Output esperado:
# Summaries: 3
#   [web] Summary: Data from web about 'LangGraph subgraphs'
#   [academic] Summary: Data from academic about 'LangGraph subgrap
#   [news] Summary: Data from news about 'LangGraph subgraphs'

Ejercicio 4: Functional API con módulo reutilizable (Medio)

Implementa el mismo patrón de la Functional API: un @entrypoint llamado process_source que hace fetch + summarize, y un orquestador que lo llama 3 veces en paralelo con diferentes fuentes. Maneja errores individuales.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

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

@task
def fetch_from_source(source: str, query: str) -> str:
    response = model.invoke(f"Simula datos de {source} sobre '{query}'. 2 oraciones.")
    return response.content

@task
def summarize_result(source: str, raw: str) -> str:
    response = model.invoke(f"Resume en 1 oración: [{source}] {raw}")
    return response.content

@entrypoint()
def process_source(config: dict) -> dict:
    source = config["source"]
    query = config["query"]
    raw = fetch_from_source(source, query).result()
    summary = summarize_result(source, raw).result()
    return {"source": source, "summary": summary}

@task
def final_synthesis(results: list) -> str:
    context = "\n".join(f"- [{r['source']}]: {r['summary']}" for r in results)
    response = model.invoke(f"Sintetiza en 2 oraciones:\n{context}")
    return response.content

@entrypoint()
def research_agent(query: str) -> dict:
    sources = ["web", "academic", "news"]
    futures = [process_source.ainvoke({"source": s, "query": query}) for s in sources]

    results = []
    errors = []
    for source, fut in zip(sources, futures):
        try:
            results.append(fut.result())
        except Exception as e:
            errors.append({"source": source, "error": str(e)})

    synthesis = final_synthesis(results).result()

    return {
        "results": results,
        "errors": errors,
        "synthesis": synthesis,
    }

result = research_agent.invoke("fine-tuning techniques for LLMs")
print(f"Fuentes OK: {len(result['results'])}, Errores: {len(result['errors'])}")
for r in result["results"]:
    print(f"  [{r['source']}] {r['summary'][:80]}...")
print(f"Síntesis: {result['synthesis'][:150]}...")
# Output esperado:
# Fuentes OK: 3, Errores: 0
#   [web] Fine-tuning techniques for LLMs include LoRA and full fine-tuning...
#   [academic] Recent research compares parameter-efficient fine-tuning methods...
#   [news] Companies are adopting fine-tuning to customize LLMs for specific...
# Síntesis: Fine-tuning LLMs has evolved with parameter-efficient techniques...

Ejercicio 5: Subgraph vs función — comparación práctica (Avanzado)

Implementa el mismo pipeline (fetch → parse → summarize) tanto como función regular como subgraph. Invoca ambos con los mismos datos y compara: ¿producen el mismo resultado? ¿Qué obtienes extra con el subgraph? Muestra cómo visualizar el subgraph y cómo no puedes visualizar la función.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from IPython.display import Image, display

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

def pipeline_as_function(source: str, query: str) -> dict:
    raw = model.invoke(f"Simula datos de {source} sobre '{query}'. 1 oración.").content
    parsed = f"[{source}] {raw}"
    summary = model.invoke(f"Resume: {parsed}").content
    return {"source": source, "raw": raw, "parsed": parsed, "summary": summary}

class PipeState(TypedDict):
    source_name: str
    query: str
    raw_data: str
    parsed_data: str
    summary: str

def sg_fetch(state: PipeState) -> dict:
    result = model.invoke(f"Simula datos de {state['source_name']} sobre '{state['query']}'. 1 oración.")
    return {"raw_data": result.content}

def sg_parse(state: PipeState) -> dict:
    return {"parsed_data": f"[{state['source_name']}] {state['raw_data']}"}

def sg_summarize(state: PipeState) -> dict:
    result = model.invoke(f"Resume: {state['parsed_data']}")
    return {"summary": result.content}

pipe = StateGraph(PipeState)
pipe.add_node("fetch", sg_fetch)
pipe.add_node("parse", sg_parse)
pipe.add_node("summarize", sg_summarize)
pipe.add_edge(START, "fetch")
pipe.add_edge("fetch", "parse")
pipe.add_edge("parse", "summarize")
pipe.add_edge("summarize", END)
pipeline_subgraph = pipe.compile()

print("=== Subgraph (visualizable) ===")
display(Image(pipeline_subgraph.get_graph().draw_mermaid_png()))

print("\n=== Función (no visualizable — es una caja negra) ===")
print("No hay diagrama disponible para funciones Python regulares.\n")

func_result = pipeline_as_function("web", "Python async")
print(f"Función → {func_result['summary'][:80]}...")

sg_result = pipeline_subgraph.invoke({
    "source_name": "web", "query": "Python async",
    "raw_data": "", "parsed_data": "", "summary": "",
})
print(f"Subgraph → {sg_result['summary'][:80]}...")

print("\n=== Diferencias ===")
print("✅ Subgraph: checkpointing por paso, visualización, streaming, resume ante fallos")
print("✅ Función: más simple, menos código, suficiente para lógica trivial")
print("❌ Función: sin checkpointing, sin visualización, si falla re-ejecuta todo")
# Output esperado:
# === Subgraph (visualizable) ===
# [Diagrama: fetch → parse → summarize]
# === Función (no visualizable — es una caja negra) ===
# No hay diagrama disponible para funciones Python regulares.
# Función → Python's async capabilities enable concurrent execution...
# Subgraph → Python's async capabilities enable concurrent execution...
# === Diferencias ===
# ✅ Subgraph: checkpointing por paso, visualización, streaming, resume ante fallos
# ✅ Función: más simple, menos código, suficiente para lógica trivial
# ❌ Función: sin checkpointing, sin visualización, si falla re-ejecuta todo

Ejercicio 6: Research Agent modular completo (Avanzado)

Construye un Research Agent que combine subgraphs + branching + merge con LLM. Requisitos: (1) un subgraph de búsqueda reutilizable (fetch → summarize), (2) un grafo padre que ejecute el subgraph en paralelo para 3 fuentes, (3) un nodo de merge que use un LLM para sintetizar, (4) manejo de errores en cada rama. Visualiza con xray=True.

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 IPython.display import Image, display

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

class SearchState(TypedDict):
    source_name: str
    query: str
    raw_data: str
    summary: str

def sg_fetch(state: SearchState) -> dict:
    response = model.invoke(
        f"Simula datos de '{state['source_name']}' sobre '{state['query']}'. 2-3 oraciones."
    )
    return {"raw_data": response.content}

def sg_summarize(state: SearchState) -> dict:
    response = model.invoke(f"Resume en 1 oración: [{state['source_name']}] {state['raw_data']}")
    return {"summary": response.content}

search_graph = StateGraph(SearchState)
search_graph.add_node("fetch", sg_fetch)
search_graph.add_node("summarize", sg_summarize)
search_graph.add_edge(START, "fetch")
search_graph.add_edge("fetch", "summarize")
search_graph.add_edge("summarize", END)
search_compiled = search_graph.compile()

class AgentState(TypedDict):
    query: str
    source_name: str
    raw_data: str
    summary: str
    collected: Annotated[list[dict], operator.add]
    report: str

def set_src(name: str):
    def node(state: AgentState) -> dict:
        return {"source_name": name, "raw_data": "", "summary": ""}
    return node

def safe_collect(state: AgentState) -> dict:
    if state.get("summary"):
        return {"collected": [{"source": state["source_name"], "summary": state["summary"], "status": "ok"}]}
    return {"collected": [{"source": state["source_name"], "summary": "No data", "status": "error"}]}

def merge_report(state: AgentState) -> dict:
    ok = [r for r in state["collected"] if r["status"] == "ok"]
    if not ok:
        return {"report": "No se pudo obtener información de ninguna fuente."}
    context = "\n".join(f"- [{r['source']}]: {r['summary']}" for r in ok)
    response = model.invoke(
        f"Genera un reporte de 3-4 oraciones sobre '{state['query']}' "
        f"basado en estas fuentes:\n\n{context}"
    )
    return {"report": response.content}

agent = StateGraph(AgentState)
agent.add_node("merge", merge_report)

for src in ["web", "academic", "news"]:
    agent.add_node(f"set_{src}", set_src(src))
    agent.add_node(f"search_{src}", search_compiled)
    agent.add_node(f"collect_{src}", safe_collect)
    agent.add_edge(START, f"set_{src}")
    agent.add_edge(f"set_{src}", f"search_{src}")
    agent.add_edge(f"search_{src}", f"collect_{src}")
    agent.add_edge(f"collect_{src}", "merge")

agent.add_edge("merge", END)

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

result = app.invoke({
    "query": "¿Cómo funciona el fine-tuning de LLMs?",
    "source_name": "", "raw_data": "", "summary": "",
    "collected": [], "report": "",
})

print(f"Fuentes: {len(result['collected'])}")
for r in result["collected"]:
    print(f"  [{r['source']}] ({r['status']}) {r['summary'][:70]}...")
print(f"\nReporte: {result['report'][:200]}...")
# Output esperado:
# Fuentes: 3
#   [web] (ok) Fine-tuning adapts pre-trained LLMs to specific tasks...
#   [academic] (ok) Research shows parameter-efficient methods like LoRA...
#   [news] (ok) Companies are increasingly using fine-tuning for custom...
# Reporte: El fine-tuning de LLMs es un proceso que adapta modelos pre-entrenados...

Resumen

En esta cápsula aprendiste:

• Subgraphs = funciones: así como una función puede llamar a otra función, un grafo puede contener otro grafo. Encapsulas lógica compleja y la reutilizas
• Crear un subgraph es crear un StateGraph normal, compilarlo, y usarlo como nodo en un grafo padre con add_node("nombre", subgraph_compilado)
• Estado compartido: los campos con el mismo nombre en padre y subgraph se mapean automáticamente. Los campos exclusivos de cada uno son invisibles para el otro
• Estado diferente: cuando padre y subgraph tienen TypedDict diferentes, usa una función wrapper que traduce entre ambas interfaces
• Reutilización: defines el subgraph una vez y lo usas N veces con diferentes configuraciones. Si cambias la lógica, la cambias en un solo lugar
• Nesting: subgraphs dentro de subgraphs es posible, pero mantén máximo 2 niveles de profundidad por sanidad
• Functional API: un @entrypoint compilado puede actuar como módulo reutilizable, con la misma filosofía de encapsulación
• Subgraph vs función regular: el subgraph te da checkpointing por paso, visualización y streaming. La función es más simple pero es una caja negra. Usa subgraph cuando la lógica es compleja y necesitas observabilidad
• xray=True en draw_mermaid_png() expande los subgraphs para mostrar sus nodos internos

Próxima cápsula: Map-Reduce — cómo procesar colecciones de datos en paralelo, enviando cada elemento a su propia instancia de un subgraph.

Recursos adicionales

1. LangGraph Subgraphs — How-to Guide — Tutorial oficial de subgraphs con ejemplos completos
2. LangGraph Concepts: Subgraphs — Documentación conceptual de subgraphs
3. State Interface Between Graphs — Cómo manejar estado entre padre y subgraph
4. Visualizing Subgraphs — Usar xray para ver nodos internos
5. LangGraph Branching — Combinar branching con subgraphs
6. Functional API — @entrypoint como módulo reutilizable
7. LangGraph Persistence — Checkpointing en subgraphs (cada paso interno tiene su propio checkpoint)

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