6. Tool Composition and Chaining

Overview

So far, every tool you've created is an independent unit: search searches, get_weather checks the weather, create_ticket creates a ticket. But in real systems, tasks aren't atomic. A user says "research this topic and give me a summary" — that requires searching, reading results, filtering what's relevant, and summarizing. Who orchestrates those steps? You have three options: the model decides at each turn (agent loop), you define an explicit pipeline (chaining), or you create a high-level tool that internally coordinates other tools (composition).

Tool composition and chaining are the patterns that let you build complex operations out of simple tools — without depending on the model making the right decision at every step. Think of composition as creating a research tool that internally calls search + read_page + summarize. From the outside, it's one tool. From the inside, it's a coordinated pipeline. The model only sees "research" and calls it once.

The difference from letting the agent loop do everything is control. When you compose tools, you decide the flow: what runs first, what data passes between steps, what to do if a step fails. When you let the model decide, it's more flexible but less predictable — and it burns more tokens. Knowing when to use each is what separates an agent that "works sometimes" from a production-ready one.

The Problem: Tools That Need Other Tools

The real scenario

Imagine a research agent. The user says: "What's the current state of AI regulation in the European Union?"

With independent tools, the agent loop's flow would be:

1. The model decides to call web_search("AI regulation EU 2026")
2. It gets results → decides to call read_page(url_1)
3. It gets content → decides to call read_page(url_2)
4. It gets content → decides to call summarize(all_content)
5. Finally it answers

Four LLM turns. Four decisions. Four opportunities for the model to get distracted, forget a step, or burn unnecessary tokens. On an API with 500ms of latency per call, that's 2+ seconds of pure decision overhead.

What you actually want

A research tool that does all of that internally:

@tool
def research(query: str) -> str:
    """Research a topic: search the web, read the relevant pages, and summarize."""
    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})

The model calls research once. A single turn. Predictable flow. If you want to add caching, retries, or logging — you do it inside the tool, not in the agent's logic.

The trade-off

Composition gives you control and efficiency, but you lose flexibility. If the model could decide at each step, it could do things like "these search results aren't good, I'm going to rephrase the query". The key is understanding when the flow is predictable enough to hard-code and when you need the model to decide.

Tool Composition: High-Level Tools

The fundamental pattern

Composition is creating a tool that internally uses other tools. From the model's perspective, it's a single tool. From your code's perspective, it's an orchestrator.

from langchain_core.tools import tool


@tool
def web_search(query: str) -> str:
    """Search the web for information."""
    return f"Results for: {query}"


@tool
def read_page(url: str) -> str:
    """Read and extract the main content of a URL."""
    return f"Content of: {url}"


@tool
def summarize(text: str, focus: str = "") -> str:
    """Summarize a long text, optionally focusing on a topic."""
    return f"Summary focused on '{focus}': {text[:200]}..."

Now, the composed tool:

@tool
def research(query: str) -> str:
    """Research a topic end to end: search the web, read pages, and generate a summary."""
    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"Could not retrieve content for: {query}"

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

Composition with error handling

A robust composed tool doesn't let one sub-tool's failure take down the whole pipeline:

@tool
def safe_research(query: str) -> str:
    """Research with fallbacks: if the web fails, use the cache. If the summary fails, return raw."""
    # Step 1: Search with a fallback
    try:
        results = web_search.invoke({"query": query})
    except Exception:
        results = check_cache(query)
        if not results:
            return f"Could not search for information about: {query}"

    # Step 2: Reading with partial tolerance
    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"Search succeeded but no readable pages. Raw:\n{results}"

    # Step 3: Summary with a fallback to concatenation
    combined = "\n\n".join(contents)
    try:
        summary = summarize.invoke({"text": combined, "focus": query})
    except Exception:
        summary = f"Summary unavailable. Content:\n{combined[:1000]}"

    if errors:
        summary += f"\n\nNote: {len(errors)} pages could not be read."
    return summary

Each step has its own error strategy. It isn't "all or nothing" — it's graceful degradation.

Tool Chaining: Tool Pipelines

Chaining vs Composition

Composition: one tool wraps others. The model only sees the outer tool.

Chaining: you orchestrate the sequence from your application code, passing outputs as inputs to the next step. You don't create a new tool — you define an explicit pipeline.

from langchain_core.tools import tool


@tool
def extract_entities(text: str) -> str:
    """Extract named entities from a text."""
    return "person=Maria Garcia, company=TechCorp, amount=$45000"


@tool
def enrich_entity(entity_name: str) -> str:
    """Look up additional information about an entity."""
    return f"{entity_name}: founded in 2020, 150 employees"


@tool
def generate_report(enriched_data: str) -> str:
    """Generate a structured report."""
    return f"REPORT:\n{enriched_data}\n---\nGenerated automatically."


def entity_enrichment_pipeline(document: str) -> str:
    """Pipeline: extract → enrich → report."""
    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 with intermediate transformations

In real life, one tool's output is rarely exactly the input the next one needs. The transformations between steps are the "glue" — JSON parsing, reformatting, filtering — plain Python code that doesn't need to be a tool:

import json

def news_analysis_pipeline(topic: str) -> str:
    """Pipeline: search → analyze sentiment → briefing."""
    raw_results = search_news.invoke({"topic": topic})
    results = json.loads(raw_results)

    # Transformation: iterate and enrich each result
    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 — your code decides inside a tool. The model sees 1 tool, spends 1 tool call. Predictable flows.

Chaining — your code decides outside of tools. The model doesn't participate. Zero tokens. Batch processing.

Agent loop — the model decides at each turn. It sees every tool, spends N tool calls. Unpredictable tasks where flexibility is worth the cost.

Comparison table

When to combine them

In practice, you mix all three:

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


def process_request(user_query: str) -> dict:
    # Agent loop: the model decides whether to use research, calculator, etc.
    agent = create_react_agent(model, [research, calculator, create_ticket])
    agent_result = agent.invoke({"messages": [("user", user_query)]})

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

Anti-patterns: Circular Dependencies and Over-composition

Anti-pattern 1: Circular dependencies

The most dangerous mistake — a tool A that calls tool B that calls tool A:

# ❌ NEVER DO THIS

@tool
def analyze(text: str) -> str:
    """Analyze text."""
    if needs_more_context(text):
        extra = research(text)  # research calls analyze → INFINITE LOOP
        return f"Analysis with context: {extra}"
    return f"Analysis: {text}"

@tool
def research(query: str) -> str:
    """Research a topic."""
    results = search.invoke({"query": query})
    return analyze(results)  # analyze can call research → BOOM

The solution is a clear hierarchy of levels:

# ✅ Hierarchy with no cycles

# Level 0: Atomic tools (they don't call other tools)
@tool
def search(query: str) -> str:
    """Search the web."""
    return f"Results for: {query}"

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

# Level 1: Only call level 0
@tool
def research(query: str) -> str:
    """Search and extract key points."""
    results = search.invoke({"query": query})
    return extract_key_points.invoke({"text": results})

# Level 2: Call level 0-1
@tool
def deep_analysis(topic: str) -> str:
    """Deep research with multiple searches."""
    initial = research.invoke({"query": topic})
    additional = research.invoke({"query": f"{topic} details"})
    return f"Initial: {initial}\nAdditional: {additional}"

The rule: a level-N tool can only call tools at level N-1 or lower. Never lateral, never upward.

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

# ❌ Too much responsibility

@tool
def do_everything(query: str) -> str:
    """Search, read, analyze, summarize, translate, format, email and archive."""
    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

If translate fails, you lose everything. You can't reuse steps. Impossible to test.

# ✅ Modular composition

@tool
def research_and_summarize(query: str) -> str:
    """Search and summarize."""
    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 = "en") -> str:
    """Translate if needed and format."""
    if language != "en":
        text = translate.invoke({"text": text, "to": language})
    return format_report.invoke({"text": text})

@tool
def distribute_report(report: str, email: str) -> str:
    """Send and archive."""
    send_email.invoke({"body": report, "to": email})
    archive.invoke({"report": report})
    return f"Report sent to {email} and archived."

Each composed tool does one coherent thing. You can use them independently.

Anti-pattern 3: Composition without typing

If the intermediate steps return opaque strings with no structure, the pipeline becomes fragile. Validate between steps: check that the output isn't empty, parse JSON explicitly, and handle None. Add type hints and checks like if not raw_results.strip(): return error so every step fails fast with a clear message.

When to Use Composition and When to Let the Agent Decide

General rule: If you can draw the flow without complicated conditional arrows → composition or chaining. If there are many branches and decisions → agent loop (or a combination).

Connection to the Project

This module's project (capsule 08) is an extraction + routing system with function calling. Tool composition shows up in two places:

1. Composed extraction tool: an extract_and_classify tool that internally uses extraction to pull out entities and then classification to assign categories — two steps that always go together.

2. Routing pipeline: after extraction, a pipeline (chaining) routes each entity to the right processor. Extraction → routing → processing is a fixed chain.

In later modules, composition gets more powerful:

• Module 4 (State Machines): the state graph's nodes are essentially composition
• Module 5 (Planning): plan-and-execute uses composition for every step of the plan
• Module 7 (MCP): composition combines local tools with remote tools from MCP servers
• Module 8 (Multi-Agent): each agent is, conceptually, a "composed tool" with autonomy

Troubleshooting

Problem 1: "The composed tool is slow because the sub-tools are sequential"

Symptom: research takes 15 seconds because it reads 5 pages one by one.

Solution: Use concurrent.futures for the independent steps:

import concurrent.futures

@tool
def fast_research(query: str) -> str:
    """Research with parallel reading."""
    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})

Problem 2: "A sub-tool fails and takes the whole composed tool down"

Symptom: If read_page fails for one URL, all of research returns an error.

Solution: Wrap each sub-tool call in try/except and degrade gracefully (see safe_research above). The composed tool should return something useful even if internal steps fail.

Problem 3: "I can't debug which pipeline step failed"

Solution: Add structured logging:

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

@tool
def debuggable_research(query: str) -> str:
    """Research with per-step logging."""
    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

Problem 4: "One tool's output format doesn't match the next one's input"

Solution: Define transformation functions and validate with 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]:
    """Transform search output into the format read needs."""
    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")]

Problem 5: "The composed tool eats too much memory with long texts"

Solution: Truncate each piece before combining:

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] + "...[truncated]"
    contents.append(content)

Exercises

Exercise 1: Basic composed tool (Easy)

Create a search_and_extract tool that: (1) searches the web with web_search, (2) uses with_structured_output to extract the 3 main entities (name, type, relevance). Return JSON with the entities.

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="Entity name")
    entity_type: Literal["person", "company", "technology", "concept"] = Field(
        description="Entity type"
    )
    relevance: Literal["high", "medium", "low"] = Field(description="Relevance")

class ExtractedEntities(BaseModel):
    entities: List[Entity] = Field(description="The 3 most relevant entities", max_length=3)

@tool
def web_search(query: str) -> str:
    """Search the web."""
    return f"OpenAI released GPT-4.1, competing with Anthropic Claude 4 in the AI market..."

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

@tool
def search_and_extract(query: str) -> str:
    """Search the web and extract the main entities."""
    raw_results = web_search.invoke({"query": query})
    extractor = model.with_structured_output(ExtractedEntities)
    extracted = extractor.invoke(
        f"Extract the 3 most relevant entities from these results "
        f"about '{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"}))

Exercise 2: 3-step pipeline with transformations (Easy)

Implement a pipeline (chaining) that: (1) takes text, (2) extracts people and companies with with_structured_output, (3) enriches each company with an enrich_company tool. Define transformation functions between each step.

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:
    """Look up info about a company."""
    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:
    # Step 1: Extract
    entities = model.with_structured_output(DocumentEntities).invoke(
        f"Extract people and companies:\n{text}"
    )
    # Transformation + Step 2: Enrich companies
    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("Maria Garcia, CTO of TechCorp, partnership with DataAI."), indent=2))

Exercise 3: Composition with per-layer error handling (Medium)

Create robust_research that: (1) searches — if it fails, returns an error, (2) reads up to 3 pages — tolerating individual failures, (3) summarizes — if it fails, returns the raw content. Each step records its status in a steps list included in the output.

from langchain_core.tools import tool
import json

@tool
def robust_research(query: str) -> str:
    """Robust research with per-step tracking."""
    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"Could not search: {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": "No readable content", "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)

Exercise 4: Detect circular dependencies (Medium)

Write detect_circular_deps that takes a dependency dict ({"research": ["search", "summarize"], "summarize": ["extract"], ...}) and returns the cycles it finds using DFS. Demonstrate with one case that has a cycle and one that doesn't.

from typing import Dict, List, Set

def detect_circular_deps(deps: Dict[str, List[str]]) -> List[List[str]]:
    """Detect cycles in a dependency graph using 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

# With a cycle
print(detect_circular_deps({
    "research": ["search", "analyze"],
    "analyze": ["extract", "research"],  # → cycle
    "search": [], "extract": [],
}))
# [['research', 'analyze', 'research']]

# No cycle (correct hierarchy)
print(detect_circular_deps({
    "deep_research": ["research", "analyze"],
    "research": ["search", "extract"],
    "analyze": ["extract"],
    "search": [], "extract": [],
}))
# []

Exercise 5: Configurable pipeline with error strategies (Hard)

Implement ConfigurablePipeline that: (1) takes steps as PipelineStep(tool, transform_fn, on_error) objects, (2) runs them sequentially, passing the transformed output along, (3) supports on_error: "skip" (ignore the step), "abort" (stop everything), "fallback" (use a default value), (4) returns the result + metadata (duration, status per step). Demonstrate with 3 steps where one uses a 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}

# Usage: search → analyze (with 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"])

Summary

In this capsule you learned:

• Tool composition is creating a tool that internally uses other tools. The model sees a single tool — you control the flow. Ideal for operations that always follow the same steps
• Tool chaining is orchestrating tools sequentially from your application code, without creating a new tool. The model doesn't participate — it's a pure data pipeline
• The key difference from the agent loop: in composition/chaining you decide the flow; in the agent loop, the model decides. Composition = predictable and efficient; agent loop = flexible and expensive
• All three combine in production: composition for predictable sub-operations, chaining for batch processing, agent loop for decisions that require the model's judgment
• Circular dependencies are the most dangerous anti-pattern: tool A → tool B → tool A = stack overflow. Solution: a clear hierarchy of levels
• Over-composition is just as problematic: one internal failure affects everything, and you can't reuse or test individual steps
• Transformations between steps are the glue of chaining — Python functions that adapt one tool's output into the next one's input
• Error handling in composition should be per layer: each sub-tool with its own strategy (skip, abort, fallback)

Next capsule: Retry Patterns and Circuit Breakers — what to do when sub-tools fail transiently, how to implement exponential backoff, circuit breakers that temporarily disable tools, and fallback tools as plan B.

Additional Resources

1. LangChain — Tools — Official tools documentation, including invoking tools inside other tools
2. LangGraph — Subgraphs — Composition at the graph level: subgraphs as nodes inside bigger graphs
3. Python — concurrent.futures — For parallelizing sub-tools inside a composed tool
4. Martin Fowler — Pipes and Filters — The architectural pattern behind tool chaining
5. Pydantic — Model Serialization — Serialization for passing typed data between pipeline steps
6. LangSmith — Tracing — Tracing composed tools for multi-step debugging