6. Tool Composition y Chaining

Descripción

Hasta ahora, cada tool que has creado es una unidad independiente: search busca, get_weather consulta clima, create_ticket crea un ticket. Pero en sistemas reales, las tareas no son atómicas. Un usuario dice "investiga este tema y dame un resumen" — eso requiere buscar, leer resultados, filtrar lo relevante, y resumir. ¿Quién orquesta esos pasos? Tienes tres opciones: el modelo decide en cada turno (agent loop), tú defines un pipeline explícito (chaining), o creas un tool de alto nivel que internamente coordina otros tools (composition).

Tool composition y chaining son los patterns que te permiten construir operaciones complejas a partir de tools simples — sin depender de que el modelo tome la decisión correcta en cada paso. Piensa en composition como crear un tool research que internamente llama search + read_page + summarize. Desde fuera, es un solo tool. Desde dentro, es un pipeline coordinado. El modelo solo ve "research" y lo llama una vez.

La diferencia con dejar que el agent loop haga todo es control. Cuando compones tools, tú decides el flujo: qué se ejecuta primero, qué datos pasan entre pasos, qué hacer si un paso falla. Cuando dejas que el modelo decida, es más flexible pero menos predecible — y usa más tokens. Saber cuándo usar cada uno es lo que separa un agente que "funciona a veces" de uno production-ready.

El Problema: Tools que Necesitan Otros Tools

El escenario real

Imagina un agente de investigación. El usuario dice: "¿Cuál es el estado actual de la regulación de AI en la Unión Europea?"

Con tools independientes, el flujo del agent loop sería:

1. Modelo decide llamar web_search("AI regulation EU 2026")
2. Recibe resultados → decide llamar read_page(url_1)
3. Recibe contenido → decide llamar read_page(url_2)
4. Recibe contenido → decide llamar summarize(all_content)
5. Finalmente responde

Cuatro turns de LLM. Cuatro decisiones. Cuatro oportunidades de que el modelo se distraiga, olvide un paso, o gaste tokens innecesarios. En una API con latencia de 500ms por call, esto son 2+ segundos solo de overhead de decisiones.

Lo que realmente quieres

Un tool research que haga todo eso internamente:

@tool
def research(query: str) -> str:
    """Investiga un tema: busca en web, lee las páginas relevantes y resume."""
    search_results = web_search.invoke({"query": query})
    urls = extract_urls(search_results)

    contents = []
    for url in urls[:3]:
        page = read_page.invoke({"url": url})
        contents.append(page)

    combined = "\n\n---\n\n".join(contents)
    return summarize.invoke({"text": combined, "focus": query})

El modelo llama research una vez. Un solo turn. Flujo predecible. Si quieres agregar caching, retry, o logging — lo haces dentro del tool, no en la lógica del agente.

El trade-off

Composition te da control y eficiencia, pero pierdes flexibilidad. Si el modelo pudiera decidir en cada paso, podría hacer cosas como "estos resultados de búsqueda no son buenos, voy a reformular la query". La clave es entender cuándo el flujo es suficientemente predecible para hard-codear y cuándo necesitas que el modelo decida.

Tool Composition: Tools de Alto Nivel

El pattern fundamental

Composition es crear un tool que internamente usa otros tools. Desde la perspectiva del modelo, es un solo tool. Desde la perspectiva de tu código, es un orquestador.

from langchain_core.tools import tool


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


@tool
def read_page(url: str) -> str:
    """Lee y extrae el contenido principal de una URL."""
    return f"Contenido de: {url}"


@tool
def summarize(text: str, focus: str = "") -> str:
    """Resume un texto largo, opcionalmente enfocando en un tema."""
    return f"Resumen enfocado en '{focus}': {text[:200]}..."

Ahora, el tool compuesto:

@tool
def research(query: str) -> str:
    """Investiga un tema completo: busca en web, lee páginas y genera un resumen."""
    results = web_search.invoke({"query": query})
    urls = [line.split(": ")[1] for line in results.split("\n") if "http" in line]

    contents = []
    for url in urls[:3]:
        try:
            contents.append(read_page.invoke({"url": url}))
        except Exception:
            continue

    if not contents:
        return f"No se pudo obtener contenido para: {query}"

    combined = "\n\n---\n\n".join(contents)
    return summarize.invoke({"text": combined, "focus": query})

Composition con manejo de errores

Un tool compuesto robusto no deja que el fallo de un sub-tool tire todo el pipeline:

@tool
def safe_research(query: str) -> str:
    """Investigación con fallbacks: si la web falla, usa caché. Si el resumen falla, retorna raw."""
    # Paso 1: Búsqueda con fallback
    try:
        results = web_search.invoke({"query": query})
    except Exception:
        results = check_cache(query)
        if not results:
            return f"No se pudo buscar información sobre: {query}"

    # Paso 2: Lectura con tolerancia parcial
    urls = extract_urls(results)
    contents, errors = [], []
    for url in urls[:3]:
        try:
            contents.append(read_page.invoke({"url": url}))
        except Exception as e:
            errors.append(f"{url}: {e}")

    if not contents:
        return f"Búsqueda exitosa pero sin páginas legibles. Raw:\n{results}"

    # Paso 3: Resumen con fallback a concatenación
    combined = "\n\n".join(contents)
    try:
        summary = summarize.invoke({"text": combined, "focus": query})
    except Exception:
        summary = f"Resumen no disponible. Contenido:\n{combined[:1000]}"

    if errors:
        summary += f"\n\nNota: {len(errors)} páginas no pudieron leerse."
    return summary

Cada paso tiene su propia estrategia de error. No es "todo o nada" — es degradación graceful.

Tool Chaining: Pipelines de Herramientas

Chaining vs Composition

Composition: un tool envuelve a otros. El modelo solo ve el tool externo.

Chaining: tú orquestas la secuencia desde tu código de aplicación, pasando outputs como inputs al siguiente paso. No creas un tool nuevo — defines un pipeline explícito.

from langchain_core.tools import tool


@tool
def extract_entities(text: str) -> str:
    """Extrae entidades nombradas de un texto."""
    return "persona=María García, empresa=TechCorp, monto=$45000"


@tool
def enrich_entity(entity_name: str) -> str:
    """Busca información adicional sobre una entidad."""
    return f"{entity_name}: fundada en 2020, 150 empleados"


@tool
def generate_report(enriched_data: str) -> str:
    """Genera un reporte estructurado."""
    return f"REPORTE:\n{enriched_data}\n---\nGenerado automáticamente."


def entity_enrichment_pipeline(document: str) -> str:
    """Pipeline: extraer → enriquecer → reportar."""
    entities_raw = extract_entities.invoke({"text": document})

    enriched_parts = []
    for name in parse_entity_names(entities_raw):
        enriched_parts.append(enrich_entity.invoke({"entity_name": name}))

    return generate_report.invoke({"enriched_data": "\n".join(enriched_parts)})

Pipeline con transformaciones intermedias

En la vida real, el output de un tool rara vez es exactamente el input que necesita el siguiente. Las transformaciones entre pasos son el "pegamento" — JSON parsing, reformateo, filtrado — código Python normal que no necesita ser un tool:

import json

def news_analysis_pipeline(topic: str) -> str:
    """Pipeline: buscar → analizar sentimiento → briefing."""
    raw_results = search_news.invoke({"topic": topic})
    results = json.loads(raw_results)

    # Transformación: iterar y enriquecer cada resultado
    analyses = []
    for result in results:
        sentiment = json.loads(analyze_sentiment.invoke({"text": result["snippet"]}))
        analyses.append(f"- {result['title']} ({sentiment['sentiment']}, {sentiment['confidence']:.0%})")

    return generate_briefing.invoke({"analyses": "\n".join(analyses)})

Composition vs Chaining vs Agent Loop

Composition — tu código decide dentro de un tool. El modelo ve 1 tool, gasta 1 tool call. Flujos predecibles.

Chaining — tu código decide fuera de tools. El modelo no participa. Cero tokens. Procesamiento batch.

Agent loop — el modelo decide en cada turno. Ve todos los tools, gasta N tool calls. Tareas impredecibles donde flexibilidad vale el costo.

Tabla comparativa

Cuándo combinarlos

En la práctica, mezclas los tres:

@tool
def research(query: str) -> str:
    """Composition: search + summarize siempre van juntos."""
    results = web_search.invoke({"query": query})
    return summarize.invoke({"text": results, "focus": query})


def process_request(user_query: str) -> dict:
    # Agent loop: el modelo decide si usar research, calculator, etc.
    agent = create_react_agent(model, [research, calculator, create_ticket])
    agent_result = agent.invoke({"messages": [("user", user_query)]})

    # Chaining: post-procesamiento determinista
    final_message = agent_result["messages"][-1].content
    entities = extract_entities.invoke({"text": final_message})
    return {"response": final_message, "entities": entities}

Anti-patterns: Dependencias Circulares y Over-composition

Anti-pattern 1: Dependencias circulares

El error más peligroso — un tool A que llama a tool B que llama a tool A:

# ❌ NUNCA HAGAS ESTO

@tool
def analyze(text: str) -> str:
    """Analiza texto."""
    if needs_more_context(text):
        extra = research(text)  # research llama analyze → LOOP INFINITO
        return f"Análisis con contexto: {extra}"
    return f"Análisis: {text}"

@tool
def research(query: str) -> str:
    """Investiga un tema."""
    results = search.invoke({"query": query})
    return analyze(results)  # analyze puede llamar research → BOOM

La solución es una jerarquía clara de niveles:

# ✅ Jerarquía sin ciclos

# Nivel 0: Tools atómicos (no llaman otros tools)
@tool
def search(query: str) -> str:
    """Busca en la web."""
    return f"Results for: {query}"

@tool
def extract_key_points(text: str) -> str:
    """Extrae puntos clave."""
    return f"Key points: {text[:100]}"

# Nivel 1: Solo llaman nivel 0
@tool
def research(query: str) -> str:
    """Busca y extrae puntos clave."""
    results = search.invoke({"query": query})
    return extract_key_points.invoke({"text": results})

# Nivel 2: Llaman nivel 0-1
@tool
def deep_analysis(topic: str) -> str:
    """Investigación profunda con múltiples búsquedas."""
    initial = research.invoke({"query": topic})
    additional = research.invoke({"query": f"{topic} details"})
    return f"Initial: {initial}\nAdditional: {additional}"

La regla: un tool de nivel N solo puede llamar tools de nivel N-1 o inferior. Nunca lateral, nunca hacia arriba.

Anti-pattern 2: Over-composition (el mega-tool)

# ❌ Demasiada responsabilidad

@tool
def do_everything(query: str) -> str:
    """Busca, lee, analiza, resume, traduce, formatea, envía email y archiva."""
    results = search.invoke({"query": query})
    pages = [read_page.invoke({"url": u}) for u in extract_urls(results)]
    analysis = analyze.invoke({"text": "\n".join(pages)})
    summary = summarize.invoke({"text": analysis})
    translated = translate.invoke({"text": summary, "to": "en"})
    send_email.invoke({"body": translated, "to": "team@company.com"})
    archive.invoke({"report": translated})
    return translated

Si falla translate, pierdes todo. No puedes reutilizar pasos. Imposible de testear.

# ✅ Composición modular

@tool
def research_and_summarize(query: str) -> str:
    """Busca y resume."""
    results = search.invoke({"query": query})
    pages = [read_page.invoke({"url": u}) for u in extract_urls(results)[:3]]
    return summarize.invoke({"text": "\n".join(pages), "focus": query})

@tool
def prepare_report(text: str, language: str = "es") -> str:
    """Traduce si es necesario y formatea."""
    if language != "es":
        text = translate.invoke({"text": text, "to": language})
    return format_report.invoke({"text": text})

@tool
def distribute_report(report: str, email: str) -> str:
    """Envía y archiva."""
    send_email.invoke({"body": report, "to": email})
    archive.invoke({"report": report})
    return f"Reporte enviado a {email} y archivado."

Cada tool compuesto hace una cosa coherente. Puedes usarlos independientemente.

Anti-pattern 3: Composition sin tipado

Si los pasos intermedios retornan strings opacos sin estructura, el pipeline se vuelve frágil. Valida entre pasos: verifica que el output no sea vacío, parsea JSON explícitamente, y maneja None. Agrega type hints y checks como if not raw_results.strip(): return error para que cada paso falle rápido con un mensaje claro.

Cuándo Usar Composition y Cuándo Dejar que el Agente Decida

Regla general: Si puedes dibujar el flujo sin flechas condicionales complicadas → composition o chaining. Si hay muchas ramas y decisiones → agent loop (o combinación).

Conexión con Proyecto

El proyecto de este módulo (cápsula 08) es un sistema de extraction + routing con function calling. Tool composition aparece en dos puntos:

1. Tool compuesto de extraction: un tool extract_and_classify que internamente usa extraction para sacar entidades y luego clasificación para asignar categorías — dos pasos que siempre van juntos.

2. Pipeline de routing: después de extraer, un pipeline (chaining) rutea cada entidad al processor correcto. Extraction → routing → processing es una cadena fija.

En los módulos posteriores, composition se vuelve más poderoso:

• Módulo 4 (State Machines): Los nodos del state graph son esencialmente composition
• Módulo 5 (Planning): Plan-and-execute usa composition para cada paso del plan
• Módulo 7 (MCP): Composition combina tools locales con tools remotos de MCP servers
• Módulo 8 (Multi-Agent): Cada agente es, conceptualmente, un "tool compuesto" con autonomía

Troubleshooting

Problema 1: "El tool compuesto tarda mucho porque los sub-tools son secuenciales"

Síntoma: research tarda 15 segundos porque lee 5 páginas una por una.

Solución: Usa concurrent.futures para los pasos independientes:

import concurrent.futures

@tool
def fast_research(query: str) -> str:
    """Investigación con lectura paralela."""
    results = web_search.invoke({"query": query})
    urls = extract_urls(results)[:5]

    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = {executor.submit(read_page.invoke, {"url": url}): url for url in urls}
        contents = []
        for future in concurrent.futures.as_completed(futures):
            try:
                contents.append(future.result())
            except Exception:
                continue

    return summarize.invoke({"text": "\n\n".join(contents), "focus": query})

Problema 2: "Un sub-tool falla y tira todo el tool compuesto"

Síntoma: Si read_page falla para una URL, todo research retorna error.

Solución: Envuelve cada sub-tool call en try/except y degrada gracefully (ver safe_research arriba). El tool compuesto debe retornar algo útil incluso si pasos internos fallan.

Problema 3: "No puedo debuggear qué paso del pipeline falló"

Solución: Agrega logging estructurado:

import logging
logger = logging.getLogger("tool_composition")

@tool
def debuggable_research(query: str) -> str:
    """Investigación con logging por paso."""
    logger.info(f"[research] START query={query}")

    results = web_search.invoke({"query": query})
    logger.info(f"[research] search: {len(results)} chars")

    contents = []
    for url in extract_urls(results)[:3]:
        try:
            content = read_page.invoke({"url": url})
            contents.append(content)
            logger.info(f"[research] read: {url} → {len(content)} chars")
        except Exception as e:
            logger.warning(f"[research] failed: {url} → {e}")

    summary = summarize.invoke({"text": "\n\n".join(contents), "focus": query})
    logger.info(f"[research] summary: {len(summary)} chars")
    return summary

Problema 4: "El formato de output de un tool no matchea el input del siguiente"

Solución: Define funciones de transformación y valida con Pydantic:

from pydantic import BaseModel, ValidationError

class SearchOutput(BaseModel):
    urls: list[str]
    snippets: list[str]

def transform_search_to_urls(search_raw: str) -> list[str]:
    """Transforma output de search al formato que necesita read."""
    try:
        parsed = SearchOutput.model_validate_json(search_raw)
        return parsed.urls
    except ValidationError:
        return [l.strip() for l in search_raw.split("\n") if l.startswith("http")]

Problema 5: "El tool compuesto consume demasiada memoria con textos largos"

Solución: Trunca cada pieza antes de combinar:

MAX_PER_PAGE = 2000

contents = []
for url in urls[:3]:
    content = read_page.invoke({"url": url})
    if len(content) > MAX_PER_PAGE:
        content = content[:MAX_PER_PAGE] + "...[truncado]"
    contents.append(content)

Ejercicios

Ejercicio 1: Tool compuesto básico (Fácil)

Crea un tool search_and_extract que: (1) busque en la web con web_search, (2) use with_structured_output para extraer las 3 entidades principales (nombre, tipo, relevancia). Retorna JSON con las entidades.

from langchain_core.tools import tool
from pydantic import BaseModel, Field
from typing import List, Literal
from langchain.chat_models import init_chat_model
import json

class Entity(BaseModel):
    name: str = Field(description="Nombre de la entidad")
    entity_type: Literal["person", "company", "technology", "concept"] = Field(
        description="Tipo de entidad"
    )
    relevance: Literal["high", "medium", "low"] = Field(description="Relevancia")

class ExtractedEntities(BaseModel):
    entities: List[Entity] = Field(description="Las 3 entidades más relevantes", max_length=3)

@tool
def web_search(query: str) -> str:
    """Busca en la web."""
    return f"OpenAI lanzó GPT-4.1, compitiendo con Anthropic Claude 4 en el mercado de AI..."

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

@tool
def search_and_extract(query: str) -> str:
    """Busca en la web y extrae las entidades principales."""
    raw_results = web_search.invoke({"query": query})
    extractor = model.with_structured_output(ExtractedEntities)
    extracted = extractor.invoke(
        f"Extrae las 3 entidades más relevantes de estos resultados "
        f"sobre '{query}':\n\n{raw_results}"
    )
    return json.dumps([e.model_dump() for e in extracted.entities], indent=2)

print(search_and_extract.invoke({"query": "AI models 2026"}))

Ejercicio 2: Pipeline de 3 pasos con transformaciones (Fácil)

Implementa un pipeline (chaining) que: (1) reciba texto, (2) extraiga personas y empresas con with_structured_output, (3) enriquezca cada empresa con un tool enrich_company. Define funciones de transformación entre cada paso.

from langchain_core.tools import tool
from pydantic import BaseModel, Field
from typing import List
from langchain.chat_models import init_chat_model
import json

class DocumentEntities(BaseModel):
    people: List[dict] = Field(default_factory=list)
    companies: List[dict] = Field(default_factory=list)

@tool
def enrich_company(company_name: str) -> str:
    """Busca info sobre una empresa."""
    return json.dumps({"name": company_name, "employees": 150, "funding": "Series B"})

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

def run_pipeline(text: str) -> dict:
    # Paso 1: Extract
    entities = model.with_structured_output(DocumentEntities).invoke(
        f"Extrae personas y empresas:\n{text}"
    )
    # Transformación + Paso 2: Enrich empresas
    enriched = [json.loads(enrich_company.invoke({"company_name": c["name"]}))
                for c in entities.companies]
    return {"people": entities.people, "companies": enriched}

print(json.dumps(run_pipeline("María García, CTO de TechCorp, alianza con DataAI."), indent=2))

Ejercicio 3: Composition con error handling por capa (Medio)

Crea robust_research que: (1) busque — si falla, retorna error, (2) lea hasta 3 páginas — tolera fallos individuales, (3) resuma — si falla, retorna contenido raw. Cada paso registra su status en una lista steps incluida en el output.

from langchain_core.tools import tool
import json

@tool
def robust_research(query: str) -> str:
    """Investigación robusta con tracking de cada paso."""
    steps = []

    try:
        search_results = web_search.invoke({"query": query})
        steps.append({"step": "search", "status": "success"})
    except Exception as e:
        steps.append({"step": "search", "status": "failed", "error": str(e)})
        return json.dumps({"result": f"No se pudo buscar: {query}", "steps": steps})

    urls = extract_urls(search_results)
    contents = []
    for url in urls[:3]:
        try:
            contents.append(read_page.invoke({"url": url}))
            steps.append({"step": f"read:{url}", "status": "success"})
        except Exception as e:
            steps.append({"step": f"read:{url}", "status": "failed", "error": str(e)})

    if not contents:
        return json.dumps({"result": "Sin contenido legible", "steps": steps})

    combined = "\n\n".join(contents)
    try:
        summary = summarize.invoke({"text": combined, "focus": query})
        steps.append({"step": "summarize", "status": "success"})
    except Exception as e:
        summary = f"Raw content:\n{combined[:500]}"
        steps.append({"step": "summarize", "status": "failed", "error": str(e)})

    return json.dumps({"result": summary, "steps": steps}, indent=2)

Ejercicio 4: Detectar dependencias circulares (Medio)

Escribe detect_circular_deps que reciba un dict de dependencias ({"research": ["search", "summarize"], "summarize": ["extract"], ...}) y retorne los ciclos encontrados usando DFS. Demuestra con un caso con ciclo y uno sin ciclo.

from typing import Dict, List, Set

def detect_circular_deps(deps: Dict[str, List[str]]) -> List[List[str]]:
    """Detecta ciclos en un grafo de dependencias usando DFS."""
    cycles = []
    visited: Set[str] = set()
    path: List[str] = []
    path_set: Set[str] = set()

    def dfs(node: str):
        if node in path_set:
            cycle_start = path.index(node)
            cycles.append(path[cycle_start:] + [node])
            return
        if node in visited:
            return

        path.append(node)
        path_set.add(node)
        for dep in deps.get(node, []):
            dfs(dep)
        path.pop()
        path_set.remove(node)
        visited.add(node)

    for node in deps:
        dfs(node)
    return cycles

# Con ciclo
print(detect_circular_deps({
    "research": ["search", "analyze"],
    "analyze": ["extract", "research"],  # → ciclo
    "search": [], "extract": [],
}))
# [['research', 'analyze', 'research']]

# Sin ciclo (jerarquía correcta)
print(detect_circular_deps({
    "deep_research": ["research", "analyze"],
    "research": ["search", "extract"],
    "analyze": ["extract"],
    "search": [], "extract": [],
}))
# []

Ejercicio 5: Pipeline configurable con estrategias de error (Difícil)

Implementa ConfigurablePipeline que: (1) reciba pasos como objetos PipelineStep(tool, transform_fn, on_error), (2) ejecute secuencialmente pasando output transformado, (3) soporte on_error: "skip" (ignora el paso), "abort" (para todo), "fallback" (usa valor default), (4) retorne resultado + metadata (duración, status por paso). Demuestra con 3 pasos donde uno usa fallback.

from typing import Callable, Literal
import time

class PipelineStep:
    def __init__(self, tool_fn, transform: Callable = None,
                 on_error: Literal["skip", "abort", "fallback"] = "abort",
                 fallback_value: str = ""):
        self.tool_fn = tool_fn
        self.transform = transform or (lambda x: x)
        self.on_error = on_error
        self.fallback_value = fallback_value

class ConfigurablePipeline:
    def __init__(self, steps: list[PipelineStep]):
        self.steps = steps

    def run(self, initial_input: dict) -> dict:
        current, metadata = initial_input, []
        for i, step in enumerate(self.steps):
            name = getattr(step.tool_fn, "name", f"step_{i}")
            start = time.time()
            try:
                current = step.tool_fn.invoke(step.transform(current))
                metadata.append({"step": name, "status": "success",
                                 "ms": round((time.time()-start)*1000)})
            except Exception as e:
                ms = round((time.time()-start)*1000)
                if step.on_error == "abort":
                    metadata.append({"step": name, "status": "aborted", "ms": ms})
                    return {"result": None, "error": str(e), "metadata": metadata}
                elif step.on_error == "skip":
                    metadata.append({"step": name, "status": "skipped", "ms": ms})
                elif step.on_error == "fallback":
                    current = step.fallback_value
                    metadata.append({"step": name, "status": "fallback", "ms": ms})
        return {"result": current, "metadata": metadata}

# Uso: search → analyze (con fallback) → format
pipeline = ConfigurablePipeline([
    PipelineStep(search_tool, transform=lambda inp: {"query": inp["query"]}),
    PipelineStep(analyze_tool, transform=lambda r: {"text": r},
                 on_error="fallback", fallback_value="No analysis"),
    PipelineStep(format_tool, transform=lambda r: {"text": r}),
])
output = pipeline.run({"query": "AI agents 2026"})
print(output["result"])

Resumen

En esta cápsula aprendiste:

• Tool composition es crear un tool que internamente usa otros tools. El modelo ve un solo tool — tú controlas el flujo. Ideal para operaciones que siempre siguen los mismos pasos
• Tool chaining es orquestar tools secuencialmente desde tu código de aplicación, sin crear un tool nuevo. El modelo no participa — es un pipeline de datos puro
• La diferencia clave con el agent loop: en composition/chaining tú decides el flujo; en agent loop, el modelo decide. Composition = predecible y eficiente; agent loop = flexible y costoso
• Los tres se combinan en producción: composition para sub-operaciones predecibles, chaining para procesamiento batch, agent loop para decisiones que requieren juicio del modelo
• Dependencias circulares son el anti-pattern más peligroso: tool A → tool B → tool A = stack overflow. Solución: jerarquía clara de niveles
• Over-composition es igualmente problemático: un fallo interno afecta todo, no puedes reutilizar ni testear pasos individuales
• Las transformaciones entre pasos son el pegamento del chaining — funciones Python que adaptan output de un tool al input del siguiente
• Error handling en composition debe ser por capa: cada sub-tool con su propia estrategia (skip, abort, fallback)

Próxima cápsula: Retry Patterns y Circuit Breakers — qué hacer cuando los sub-tools fallan transitoriamente, cómo implementar exponential backoff, circuit breakers que deshabilitan tools temporalmente, y fallback tools como plan B.

Recursos Adicionales

1. LangChain — Tools — Documentación oficial de tools, incluyendo invocación de tools dentro de otros tools
2. LangGraph — Subgraphs — Composition a nivel de grafos: subgraphs como nodos dentro de grafos más grandes
3. Python — concurrent.futures — Para paralelizar sub-tools dentro de un tool compuesto
4. Martin Fowler — Pipes and Filters — El pattern arquitectónico detrás de tool chaining
5. Pydantic — Model Serialization — Serialización para pasar datos tipados entre pasos del pipeline
6. LangSmith — Tracing — Tracing de tools compuestos para debugging multi-step