Production Patterns

Capsule overview

The patterns from capsules 02-06 make your system robust: retry with backoff, parallel branching, reusable subgraphs, map-reduce for collections, and error handling with fallback. But there's one more layer that separates a working prototype from a real production system: per-node timeouts, internal rate limiting, structured logging, and performance metrics.

These are the "last 10%" that make your system operable — so you can understand what's happening when something breaks at 3am, so you can answer "why did this request take 12 seconds?" with concrete data, and so you can scale from 1 user in development to 100 concurrent users without everything falling over.

The problem

Your agent works perfectly in development. You test it with one topic, it runs in 3 seconds, it produces a clean report. Ready for production, right?

No.

In production with 100 concurrent users:

• A search node takes 45 seconds because the external API is slow. The other 99 users wait.
• Your agent makes 200 calls per minute to an API that allows 60. They cut off your access.
• A user reports "my research failed". You have no idea where or why — you only know that it failed.
• The team asks "how many tokens did we spend yesterday?" and you can't answer.

These are operability problems, not functionality problems. Your agent does the right thing — but you can't operate it, monitor it, or scale it. This capsule closes that gap.

Per-node timeouts

The problem it solves

A node in your graph calls an external API. Normally it responds in 2 seconds. But one day the API is slow and takes 60 seconds. Without a timeout, that node blocks the whole pipeline. With 100 concurrent requests, you have 100 threads blocked waiting on an API that isn't going to get better.

Implementation with asyncio

from dotenv import load_dotenv
load_dotenv()

import asyncio
import time
from langgraph.func import entrypoint, task


@task
async def fast_search(query: str) -> dict:
    """A search that responds fast."""
    await asyncio.sleep(0.5)
    return {"source": "fast_api", "content": f"Fast results for '{query}'"}


@task
async def slow_search(query: str) -> dict:
    """A search that simulates a slow API."""
    await asyncio.sleep(10)
    return {"source": "slow_api", "content": f"Slow results for '{query}'"}


@task
async def search_with_timeout(query: str, timeout_seconds: float = 3.0) -> dict:
    """Runs a search with a timeout. If it exceeds the time, it returns an error."""
    try:
        result = await asyncio.wait_for(
            slow_search.acall(query),
            timeout=timeout_seconds,
        )
        return {"status": "success", "result": result}
    except asyncio.TimeoutError:
        return {
            "status": "timeout",
            "error": f"Search exceeded {timeout_seconds}s",
            "query": query,
        }


@entrypoint()
async def research_with_timeouts(topic: str) -> dict:
    start = time.time()

    fast_future = fast_search(topic)
    slow_future = search_with_timeout(topic, timeout_seconds=3.0)

    fast_result = await fast_future
    slow_result = await slow_future

    elapsed = time.time() - start

    return {
        "topic": topic,
        "fast_result": fast_result,
        "slow_result": slow_result,
        "total_time_seconds": round(elapsed, 2),
    }


result = asyncio.run(
    research_with_timeouts.ainvoke("machine learning in medicine")
)
print(f"Total time: {result['total_time_seconds']}s")
print(f"Fast: {result['fast_result']['source']}")
print(f"Slow: {result['slow_result']['status']}")
# Expected output:
# Total time: ~3.0s (not 10s)
# Fast: fast_api
# Slow: timeout

Without the timeout, the pipeline would take 10 seconds waiting on slow_search. With a 3-second timeout, it fails fast and the pipeline moves on with the results it has.

Timeout with threading (for synchronous code)

If your tasks are synchronous, you can use concurrent.futures for timeouts:

from dotenv import load_dotenv
load_dotenv()

import time
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError
from langgraph.func import entrypoint, task


def _slow_api_call(query: str) -> str:
    """Simulates an API that takes a long time."""
    time.sleep(15)
    return f"Result for {query}"


@task
def search_with_thread_timeout(query: str, timeout_seconds: float = 3.0) -> dict:
    """Runs a synchronous search with a timeout using ThreadPoolExecutor."""
    with ThreadPoolExecutor(max_workers=1) as executor:
        future = executor.submit(_slow_api_call, query)
        try:
            result = future.result(timeout=timeout_seconds)
            return {"status": "success", "content": result}
        except FuturesTimeoutError:
            return {
                "status": "timeout",
                "error": f"API exceeded {timeout_seconds}s",
            }


@entrypoint()
def pipeline_with_sync_timeout(topic: str) -> dict:
    result = search_with_thread_timeout(topic, timeout_seconds=2.0).result()
    return result


result = pipeline_with_sync_timeout.invoke("quantum computing")
print(f"Status: {result['status']}")
print(f"Error: {result.get('error', 'none')}")
# Expected output:
# Status: timeout
# Error: API exceeded 2.0s

When to use each approach

Internal rate limiting

The problem it solves

Your agent searches 3 sources for each sub-query, with 4 sub-queries. That's 12 calls to external APIs in parallel. If the API allows 10 requests per minute, the last 2 calls fail with 429 Too Many Requests. Worse: if you have 10 concurrent users, that's 120 simultaneous calls.

Implementation with InMemoryRateLimiter

LangChain ships InMemoryRateLimiter to control the call rate:

from dotenv import load_dotenv
load_dotenv()

import time
from langchain_core.rate_limiters import InMemoryRateLimiter
from langchain.chat_models import init_chat_model
from langgraph.func import entrypoint, task

rate_limiter = InMemoryRateLimiter(
    requests_per_second=2,
    check_every_n_seconds=0.1,
    max_bucket_size=5,
)

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


@task
def analyze_topic(topic: str, index: int) -> dict:
    """Analyzes a sub-topic with automatic rate limiting."""
    start = time.time()
    response = model.invoke(
        f"Summarize this topic in one sentence: {topic}"
    )
    elapsed = time.time() - start
    return {
        "index": index,
        "topic": topic,
        "summary": response.content,
        "elapsed_seconds": round(elapsed, 2),
    }


@entrypoint()
def rate_limited_pipeline(topics: list) -> dict:
    start = time.time()

    futures = [analyze_topic(t, i) for i, t in enumerate(topics)]
    results = [f.result() for f in futures]

    total_elapsed = time.time() - start

    return {
        "results": results,
        "total_time_seconds": round(total_elapsed, 2),
        "requests_made": len(results),
    }


topics = [
    "artificial intelligence",
    "quantum computing",
    "renewable energy",
    "biotechnology",
    "space exploration",
    "cybersecurity",
]

result = rate_limited_pipeline.invoke(topics)
print(f"Total: {result['total_time_seconds']}s for {result['requests_made']} requests")
for r in result["results"]:
    print(f"  [{r['index']}] {r['topic']}: {r['elapsed_seconds']}s")
# Expected output:
# Total: ~3-4s for 6 requests (rate limited to 2/s)
#   [0] artificial intelligence: 0.8s
#   [1] quantum computing: 1.2s
#   [2] renewable energy: 1.5s
#   ...

A rate limiter per service

In production, different APIs have different limits. Create one rate limiter per service:

from langchain_core.rate_limiters import InMemoryRateLimiter

rate_limiters = {
    "openai": InMemoryRateLimiter(
        requests_per_second=5,
        check_every_n_seconds=0.1,
        max_bucket_size=10,
    ),
    "search_api": InMemoryRateLimiter(
        requests_per_second=1,
        check_every_n_seconds=0.1,
        max_bucket_size=3,
    ),
    "news_api": InMemoryRateLimiter(
        requests_per_second=2,
        check_every_n_seconds=0.1,
        max_bucket_size=5,
    ),
}

Each model or client uses its own rate limiter. That way, calls to OpenAI don't block calls to the search API, and vice versa.

Structured logging

The problem it solves

Your agent fails. The log says:

ERROR: Something went wrong

Useless. You don't know which node failed, how long it ran before failing, what input it got, or which user request caused the error. With structured logging, the log says:

{"node": "search_academic", "duration_ms": 4500, "status": "error", "error": "TimeoutError", "request_id": "abc123", "query": "quantum computing", "timestamp": "2026-03-08T14:30:00"}

Now you can filter by request_id, see that search_academic is the problem node, and know that it ran for 4.5 seconds before failing.

Full implementation

from dotenv import load_dotenv
load_dotenv()

import json
import time
import uuid
import logging
from langgraph.func import entrypoint, task

logging.basicConfig(
    level=logging.INFO,
    format="%(message)s",
)
logger = logging.getLogger("research_agent")


class StructuredLogger:
    """Logger that emits structured JSON with request context."""

    def __init__(self, logger_instance: logging.Logger):
        self._logger = logger_instance
        self._request_id = None

    def set_request_id(self, request_id: str):
        self._request_id = request_id

    def _log(self, level: str, node: str, **kwargs):
        entry = {
            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S"),
            "level": level,
            "request_id": self._request_id,
            "node": node,
            **kwargs,
        }
        self._logger.info(json.dumps(entry, ensure_ascii=False))

    def node_start(self, node: str, **kwargs):
        self._log("INFO", node, event="node_start", **kwargs)

    def node_end(self, node: str, duration_ms: float, **kwargs):
        self._log("INFO", node, event="node_end", duration_ms=round(duration_ms, 1), **kwargs)

    def node_error(self, node: str, error: str, duration_ms: float, **kwargs):
        self._log("ERROR", node, event="node_error", error=error, duration_ms=round(duration_ms, 1), **kwargs)


slog = StructuredLogger(logger)


@task
def search_with_logging(query: str, source: str) -> dict:
    """Search with automatic structured logging."""
    slog.node_start("search", source=source, query=query)
    start = time.time()

    try:
        time.sleep(0.3)

        if source == "academic" and "quantum" in query.lower():
            raise ConnectionError("Academic API unavailable")

        result = {
            "source": source,
            "content": f"Results from {source} for '{query}'",
        }
        duration_ms = (time.time() - start) * 1000
        slog.node_end("search", duration_ms=duration_ms, source=source, status="success")
        return result

    except Exception as e:
        duration_ms = (time.time() - start) * 1000
        slog.node_error("search", error=str(e), duration_ms=duration_ms, source=source)
        raise


@task
def synthesize_with_logging(results: list) -> str:
    """Synthesis with structured logging."""
    slog.node_start("synthesize", num_sources=len(results))
    start = time.time()

    summary = f"Synthesis of {len(results)} sources completed."

    duration_ms = (time.time() - start) * 1000
    slog.node_end("synthesize", duration_ms=duration_ms, num_sources=len(results))
    return summary


@entrypoint()
def logged_pipeline(topic: str) -> dict:
    request_id = uuid.uuid4().hex[:8]
    slog.set_request_id(request_id)

    slog.node_start("pipeline", topic=topic)
    pipeline_start = time.time()

    sources = ["web", "academic", "news"]
    futures = [search_with_logging(topic, s) for s in sources]

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

    summary = synthesize_with_logging(results).result()

    pipeline_duration = (time.time() - pipeline_start) * 1000
    slog.node_end("pipeline", duration_ms=pipeline_duration, sources_ok=len(results), sources_failed=len(errors))

    return {
        "request_id": request_id,
        "summary": summary,
        "sources_ok": len(results),
        "errors": errors,
        "pipeline_duration_ms": round(pipeline_duration, 1),
    }


result = logged_pipeline.invoke("quantum computing")
print(f"\nRequest {result['request_id']}: {result['sources_ok']} sources OK, {len(result['errors'])} errors")
print(f"Total duration: {result['pipeline_duration_ms']}ms")
# Expected output (logs on stderr, result on stdout):
# {"timestamp": "2026-03-08T14:30:00", "level": "INFO", "request_id": "a1b2c3d4", "node": "pipeline", "event": "node_start", "topic": "quantum computing"}
# {"timestamp": "2026-03-08T14:30:00", "level": "INFO", "request_id": "a1b2c3d4", "node": "search", "event": "node_start", "source": "web", "query": "quantum computing"}
# {"timestamp": "2026-03-08T14:30:00", "level": "INFO", "request_id": "a1b2c3d4", "node": "search", "event": "node_end", "duration_ms": 302.1, "source": "web", "status": "success"}
# {"timestamp": "2026-03-08T14:30:01", "level": "ERROR", "request_id": "a1b2c3d4", "node": "search", "event": "node_error", "error": "Academic API unavailable", "duration_ms": 300.5, "source": "academic"}
# {"timestamp": "2026-03-08T14:30:01", "level": "INFO", "request_id": "a1b2c3d4", "node": "search", "event": "node_end", "duration_ms": 301.3, "source": "news", "status": "success"}
# {"timestamp": "2026-03-08T14:30:01", "level": "INFO", "request_id": "a1b2c3d4", "node": "synthesize", "event": "node_start", "num_sources": 2}
# {"timestamp": "2026-03-08T14:30:01", "level": "INFO", "request_id": "a1b2c3d4", "node": "pipeline", "event": "node_end", "duration_ms": 920.5, "sources_ok": 2, "sources_failed": 1}
#
# Request a1b2c3d4: 2 sources OK, 1 errors
# Total duration: 920.5ms

Correlation IDs: tracing a full request

The request_id is the key piece. When a user reports "my research failed", they give you the request_id and you can filter every log line from that request:

# Filter every log line from a specific request
cat logs.jsonl | jq 'select(.request_id == "a1b2c3d4")'

You'll see every node that ran, how long it took, which ones failed, and in what order. Without correlation IDs, the logs from 100 concurrent requests get interleaved and become impossible to analyze.

Performance metrics

The problem it solves

"How long does our pipeline take on average?" "Which node is the bottleneck?" "How many tokens did we spend yesterday?" Without metrics, you answer "I don't know" or with an anecdote from the last time you tested it. With metrics, you answer with numbers.

Implementation: an in-memory metrics collector

from dotenv import load_dotenv
load_dotenv()

import time
import statistics
from collections import defaultdict
from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model


class MetricsCollector:
    """In-memory metrics collector for LangGraph pipelines."""

    def __init__(self):
        self._latencies: dict[str, list[float]] = defaultdict(list)
        self._success_count: dict[str, int] = defaultdict(int)
        self._error_count: dict[str, int] = defaultdict(int)
        self._token_usage: dict[str, int] = defaultdict(int)

    def record_latency(self, node: str, duration_ms: float):
        self._latencies[node].append(duration_ms)

    def record_success(self, node: str):
        self._success_count[node] += 1

    def record_error(self, node: str):
        self._error_count[node] += 1

    def record_tokens(self, node: str, tokens: int):
        self._token_usage[node] += tokens

    def get_summary(self) -> dict:
        summary = {}
        for node in set(
            list(self._latencies.keys())
            + list(self._success_count.keys())
            + list(self._error_count.keys())
        ):
            latencies = self._latencies.get(node, [])
            successes = self._success_count.get(node, 0)
            errors = self._error_count.get(node, 0)
            total = successes + errors

            summary[node] = {
                "total_calls": total,
                "successes": successes,
                "errors": errors,
                "success_rate": round(successes / total, 2) if total > 0 else 0,
                "avg_latency_ms": round(statistics.mean(latencies), 1) if latencies else 0,
                "p50_latency_ms": round(statistics.median(latencies), 1) if latencies else 0,
                "p95_latency_ms": round(
                    sorted(latencies)[int(len(latencies) * 0.95)] if latencies else 0, 1
                ),
                "max_latency_ms": round(max(latencies), 1) if latencies else 0,
                "total_tokens": self._token_usage.get(node, 0),
            }
        return summary

    def reset(self):
        self._latencies.clear()
        self._success_count.clear()
        self._error_count.clear()
        self._token_usage.clear()


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


@task
def search_node(query: str, source: str) -> dict:
    start = time.time()
    try:
        time.sleep(0.2)
        result = {"source": source, "content": f"Result from {source}"}
        duration_ms = (time.time() - start) * 1000
        metrics.record_latency(f"search_{source}", duration_ms)
        metrics.record_success(f"search_{source}")
        return result
    except Exception:
        duration_ms = (time.time() - start) * 1000
        metrics.record_latency(f"search_{source}", duration_ms)
        metrics.record_error(f"search_{source}")
        raise


@task
def synthesize_node(topic: str, results: list) -> str:
    start = time.time()
    context = "\n".join(r["content"] for r in results)
    response = model.invoke(
        f"Summarize these results about '{topic}' in one sentence: {context}"
    )

    duration_ms = (time.time() - start) * 1000
    metrics.record_latency("synthesize", duration_ms)
    metrics.record_success("synthesize")

    token_count = response.usage_metadata.get("total_tokens", 0) if response.usage_metadata else 0
    metrics.record_tokens("synthesize", token_count)

    return response.content


@entrypoint()
def pipeline_with_metrics(topic: str) -> dict:
    start = time.time()

    sources = ["web", "academic", "news"]
    futures = [search_node(topic, s) for s in sources]
    results = [f.result() for f in futures]

    summary = synthesize_node(topic, results).result()

    pipeline_ms = (time.time() - start) * 1000
    metrics.record_latency("pipeline_total", pipeline_ms)
    metrics.record_success("pipeline_total")

    return {"summary": summary}


for topic in ["AI in education", "quantum computing", "renewable energy"]:
    pipeline_with_metrics.invoke(topic)

summary = metrics.get_summary()
print("\n📊 PERFORMANCE METRICS")
print("=" * 60)
for node, data in sorted(summary.items()):
    print(f"\n  {node}:")
    print(f"    Calls: {data['total_calls']} ({data['success_rate']:.0%} success)")
    print(f"    Latency: avg={data['avg_latency_ms']}ms, p50={data['p50_latency_ms']}ms, p95={data['p95_latency_ms']}ms")
    if data['total_tokens'] > 0:
        print(f"    Tokens: {data['total_tokens']}")
# Expected output:
# 📊 PERFORMANCE METRICS
# ============================================================
#
#   pipeline_total:
#     Calls: 3 (100% success)
#     Latency: avg=1200.5ms, p50=1180.3ms, p95=1250.1ms
#
#   search_academic:
#     Calls: 3 (100% success)
#     Latency: avg=201.2ms, p50=200.8ms, p95=202.1ms
#
#   search_news:
#     Calls: 3 (100% success)
#     Latency: avg=200.5ms, p50=200.3ms, p95=201.0ms
#
#   search_web:
#     Calls: 3 (100% success)
#     Latency: avg=200.8ms, p50=200.5ms, p95=201.5ms
#
#   synthesize:
#     Calls: 3 (100% success)
#     Latency: avg=980.3ms, p50=950.1ms, p95=1050.2ms
#     Tokens: 450

The metrics reveal that synthesize is the bottleneck (980ms vs 200ms for search). With that information, you can decide whether to optimize the prompt, use a faster model for synthesis, or cache similar results.

Putting it all together: a production-ready graph

Now combine the 4 patterns into a single pipeline. This is the complete pattern you'd use in production:

from dotenv import load_dotenv
load_dotenv()

import json
import time
import uuid
import logging
import statistics
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError
from langchain_core.rate_limiters import InMemoryRateLimiter
from langchain.chat_models import init_chat_model
from langgraph.func import entrypoint, task


# === INFRASTRUCTURE ===

logging.basicConfig(level=logging.INFO, format="%(message)s")
log = logging.getLogger("production_agent")


class ProdLogger:
    def __init__(self):
        self.request_id = None

    def set_request(self, rid: str):
        self.request_id = rid

    def info(self, node: str, event: str, **kw):
        entry = {"ts": time.strftime("%H:%M:%S"), "rid": self.request_id, "node": node, "event": event, **kw}
        log.info(json.dumps(entry, ensure_ascii=False))

    def error(self, node: str, event: str, **kw):
        entry = {"ts": time.strftime("%H:%M:%S"), "rid": self.request_id, "node": node, "event": event, "level": "ERROR", **kw}
        log.info(json.dumps(entry, ensure_ascii=False))


class ProdMetrics:
    def __init__(self):
        self.latencies = defaultdict(list)
        self.counts = defaultdict(lambda: {"ok": 0, "err": 0})
        self.tokens = defaultdict(int)

    def record(self, node: str, duration_ms: float, success: bool, tokens: int = 0):
        self.latencies[node].append(duration_ms)
        self.counts[node]["ok" if success else "err"] += 1
        if tokens:
            self.tokens[node] += tokens

    def summary(self) -> dict:
        out = {}
        for node in self.latencies:
            lats = self.latencies[node]
            c = self.counts[node]
            total = c["ok"] + c["err"]
            out[node] = {
                "calls": total,
                "success_rate": f"{c['ok']/total:.0%}" if total else "N/A",
                "avg_ms": round(statistics.mean(lats), 1),
                "p95_ms": round(sorted(lats)[int(len(lats) * 0.95)], 1) if lats else 0,
                "tokens": self.tokens.get(node, 0),
            }
        return out


plog = ProdLogger()
pmetrics = ProdMetrics()

rate_limiter = InMemoryRateLimiter(
    requests_per_second=3,
    check_every_n_seconds=0.1,
    max_bucket_size=5,
)

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

SEARCH_TIMEOUT_SECONDS = 5.0


# === TASKS ===

def _mock_search(query: str, source: str) -> str:
    """Simulates an external search with variable latency."""
    import random
    delay = random.uniform(0.2, 0.8)
    time.sleep(delay)
    return f"[{source}] Results for '{query}': relevant information found."


@task
def production_search(query: str, source: str) -> dict:
    """Search with timeout + logging + metrics."""
    plog.info("search", "start", source=source, query=query[:50])
    start = time.time()

    with ThreadPoolExecutor(max_workers=1) as executor:
        future = executor.submit(_mock_search, query, source)
        try:
            content = future.result(timeout=SEARCH_TIMEOUT_SECONDS)
            duration_ms = (time.time() - start) * 1000

            plog.info("search", "end", source=source, duration_ms=round(duration_ms))
            pmetrics.record("search", duration_ms, success=True)

            return {"source": source, "content": content, "status": "ok"}

        except FuturesTimeoutError:
            duration_ms = (time.time() - start) * 1000
            plog.error("search", "timeout", source=source, duration_ms=round(duration_ms))
            pmetrics.record("search", duration_ms, success=False)
            return {"source": source, "content": "", "status": "timeout"}


@task
def production_synthesize(topic: str, results: list) -> dict:
    """Synthesis with rate limiting + logging + metrics + tokens."""
    plog.info("synthesize", "start", num_sources=len(results))
    start = time.time()

    context = "\n".join(r["content"] for r in results if r["content"])
    response = model.invoke(
        f"Summarize these results about '{topic}' in 2-3 sentences:\n{context}"
    )

    duration_ms = (time.time() - start) * 1000
    tokens = response.usage_metadata.get("total_tokens", 0) if response.usage_metadata else 0

    plog.info("synthesize", "end", duration_ms=round(duration_ms), tokens=tokens)
    pmetrics.record("synthesize", duration_ms, success=True, tokens=tokens)

    return {"summary": response.content, "tokens_used": tokens}


# === PIPELINE ===

@entrypoint()
def production_pipeline(topic: str) -> dict:
    request_id = uuid.uuid4().hex[:8]
    plog.set_request(request_id)
    plog.info("pipeline", "start", topic=topic)

    pipeline_start = time.time()

    sources = ["web", "academic", "news"]
    search_futures = [production_search(topic, s) for s in sources]
    search_results = [f.result() for f in search_futures]

    successful = [r for r in search_results if r["status"] == "ok"]
    failed = [r for r in search_results if r["status"] != "ok"]

    if not successful:
        plog.error("pipeline", "all_sources_failed")
        pmetrics.record("pipeline", (time.time() - pipeline_start) * 1000, success=False)
        return {"error": "All sources failed", "request_id": request_id}

    synthesis = production_synthesize(topic, successful).result()

    pipeline_ms = (time.time() - pipeline_start) * 1000
    plog.info("pipeline", "end", duration_ms=round(pipeline_ms), sources_ok=len(successful), sources_failed=len(failed))
    pmetrics.record("pipeline", pipeline_ms, success=True)

    return {
        "request_id": request_id,
        "summary": synthesis["summary"],
        "sources_ok": len(successful),
        "sources_failed": len(failed),
        "tokens_used": synthesis["tokens_used"],
        "duration_ms": round(pipeline_ms),
    }


result = production_pipeline.invoke("the impact of LLMs on software production")
print(f"\n{'=' * 50}")
print(f"Request: {result['request_id']}")
print(f"Sources: {result['sources_ok']} OK, {result['sources_failed']} failed")
print(f"Tokens: {result['tokens_used']}")
print(f"Duration: {result['duration_ms']}ms")
print(f"Summary: {result['summary'][:120]}...")

print(f"\n{'=' * 50}")
print("ACCUMULATED METRICS:")
for node, data in pmetrics.summary().items():
    print(f"  {node}: {data['calls']} calls, {data['success_rate']} success, avg {data['avg_ms']}ms")
# Expected output:
# {"ts": "14:30:00", "rid": "a1b2c3d4", "node": "pipeline", "event": "start", "topic": "the impact of LLMs on software production"}
# {"ts": "14:30:00", "rid": "a1b2c3d4", "node": "search", "event": "start", "source": "web", ...}
# {"ts": "14:30:00", "rid": "a1b2c3d4", "node": "search", "event": "end", "source": "web", "duration_ms": 450}
# ... (more logs)
#
# ==================================================
# Request: a1b2c3d4
# Sources: 3 OK, 0 failed
# Tokens: 150
# Duration: 1850ms
# Summary: LLMs are transforming software production...
#
# ==================================================
# ACCUMULATED METRICS:
#   search: 3 calls, 100% success, avg 420.5ms
#   synthesize: 1 calls, 100% success, avg 980.3ms
#   pipeline: 1 calls, 100% success, avg 1850.2ms

This pipeline has everything you need for production:

• ✅ Timeouts per search (5s maximum, doesn't block the pipeline)
• ✅ Rate limiting on model calls (3 req/s, respects OpenAI's limits)
• ✅ Structured logging with a correlation ID (filterable by request)
• ✅ Metrics for latency, success rate, and tokens (answers business questions)
• ✅ Graceful degradation (if every source fails, it doesn't crash)

What you'd need without LangGraph

Without LangGraph, implementing this pipeline requires:

The boilerplate without LangGraph is 3-5x more code for the same functionality, and it doesn't include checkpointing (which is the hardest part to implement correctly).

Troubleshooting

Problem 1: "The rate limiter blocks everything and the pipeline is very slow"

Symptom: With requests_per_second=1 and 10 parallel tasks, the pipeline takes 10 seconds.

Cause: The rate limiter serializes the requests. If you have 10 tasks and only 1 req/s allowed, each task waits its turn.

Fix: Tune requests_per_second to your API's actual limit. If the API allows 60 req/min, use requests_per_second=1 (correct). If it allows 600 req/min, use requests_per_second=10. The max_bucket_size allows short bursts:

rate_limiter = InMemoryRateLimiter(
    requests_per_second=10,
    check_every_n_seconds=0.05,
    max_bucket_size=20,
)

Problem 2: "Logs from different requests get mixed together and I can't trace one"

Symptom: With 10 concurrent requests, the JSON logs are interleaved and you don't know which ones belong to which request.

Cause: The request_id gets lost between nodes, or you're using a global logger without correlation.

Fix: The StructuredLogger in this capsule uses set_request_id() at the start of the pipeline. For real concurrency, use Python's contextvars so each thread has its own request_id:

import contextvars

_request_id_var = contextvars.ContextVar("request_id", default="unknown")

class ThreadSafeLogger:
    def set_request_id(self, rid: str):
        _request_id_var.set(rid)

    def _log(self, node: str, **kwargs):
        entry = {"request_id": _request_id_var.get(), "node": node, **kwargs}
        print(json.dumps(entry))

Problem 3: "The p95 metrics aren't reliable"

Symptom: The reported p95 doesn't match reality — sometimes it's lower than the average.

Cause: With few samples (fewer than 20), statistical percentiles aren't representative. With 3 samples, the p95 is basically the highest value.

Fix: Only report percentiles when you have enough samples:

def get_p95(self, node: str) -> str:
    lats = self.latencies.get(node, [])
    if len(lats) < 20:
        return f"~{max(lats):.1f}ms (only {len(lats)} samples)"
    return f"{sorted(lats)[int(len(lats) * 0.95)]:.1f}ms"

Problem 4: "The timeout doesn't cancel the task, it just ignores the result"

Symptom: You set a 3-second timeout, but the function keeps running in the background eating resources.

Cause: ThreadPoolExecutor.submit().result(timeout=3) raises TimeoutError in the caller, but the thread keeps running. Python can't kill threads cleanly.

Fix: For real timeouts that free up resources, use asyncio with cancellable tasks:

async def search_with_real_cancel(query: str, timeout: float):
    task = asyncio.create_task(async_search(query))
    try:
        return await asyncio.wait_for(task, timeout=timeout)
    except asyncio.TimeoutError:
        task.cancel()
        return {"status": "timeout"}

With asyncio, task.cancel() really does cancel the coroutine. With threads, there's no clean equivalent.

Exercises

Exercise 1: Configurable timeout per source (Easy)

Modify the timeout pipeline so each source has its own timeout: web = 2s, academic = 5s, news = 3s. Simulate academic taking 4s (within its timeout) and web taking 3s (outside its timeout).

from dotenv import load_dotenv
load_dotenv()

import time
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError
from langgraph.func import entrypoint, task

SOURCE_TIMEOUTS = {
    "web": 2.0,
    "academic": 5.0,
    "news": 3.0,
}

SOURCE_DELAYS = {
    "web": 3.0,
    "academic": 4.0,
    "news": 1.0,
}


def _mock_search(query: str, source: str) -> str:
    time.sleep(SOURCE_DELAYS[source])
    return f"[{source}] Results for '{query}'"


@task
def search_with_custom_timeout(query: str, source: str) -> dict:
    timeout = SOURCE_TIMEOUTS[source]
    with ThreadPoolExecutor(max_workers=1) as executor:
        future = executor.submit(_mock_search, query, source)
        try:
            content = future.result(timeout=timeout)
            return {"source": source, "status": "ok", "content": content}
        except FuturesTimeoutError:
            return {"source": source, "status": "timeout", "timeout_seconds": timeout}


@entrypoint()
def pipeline(topic: str) -> dict:
    sources = ["web", "academic", "news"]
    futures = [search_with_custom_timeout(topic, s) for s in sources]
    results = [f.result() for f in futures]

    ok = [r for r in results if r["status"] == "ok"]
    failed = [r for r in results if r["status"] != "ok"]

    return {
        "ok_sources": [r["source"] for r in ok],
        "failed_sources": [f"{r['source']} (timeout {r.get('timeout_seconds', '?')}s)" for r in failed],
    }


result = pipeline.invoke("deep learning")
print(f"OK: {result['ok_sources']}")
print(f"Failed: {result['failed_sources']}")
# Expected output:
# OK: ['academic', 'news']  (academic takes 4s but its timeout is 5s; news takes 1s)
# Failed: ['web (timeout 2.0s)']  (web takes 3s but its timeout is 2s)

Exercise 2: Rate limiter with visual tracking (Easy)

Create a pipeline that makes 8 calls to a model with a rate limit of 2/second. Print the timestamp of each call so you can visually verify that the rate limit is respected.

from dotenv import load_dotenv
load_dotenv()

import time
from langchain_core.rate_limiters import InMemoryRateLimiter
from langchain.chat_models import init_chat_model
from langgraph.func import entrypoint, task

rate_limiter = InMemoryRateLimiter(
    requests_per_second=2,
    check_every_n_seconds=0.1,
    max_bucket_size=2,
)

model = init_chat_model("openai:gpt-4.1-mini", rate_limiter=rate_limiter)
start_time = time.time()


@task
def timed_call(index: int) -> dict:
    elapsed = time.time() - start_time
    response = model.invoke(f"Say 'hello {index}' and nothing else.")
    after = time.time() - start_time
    return {
        "index": index,
        "request_at": round(elapsed, 2),
        "response_at": round(after, 2),
        "content": response.content.strip(),
    }


@entrypoint()
def rate_limit_demo(count: int) -> list:
    futures = [timed_call(i) for i in range(count)]
    return [f.result() for f in futures]


results = rate_limit_demo.invoke(8)
for r in sorted(results, key=lambda x: x["request_at"]):
    print(f"  Call {r['index']}: request @{r['request_at']}s → response @{r['response_at']}s | {r['content']}")
# Expected output (approximate times — rate limited to 2/s):
#   Call 0: request @0.01s → response @0.85s | hello 0
#   Call 1: request @0.01s → response @0.90s | hello 1
#   Call 2: request @0.51s → response @1.35s | hello 2
#   Call 3: request @0.51s → response @1.40s | hello 3
#   Call 4: request @1.01s → response @1.85s | hello 4
#   ... (each pair of calls separated by ~0.5s)

Exercise 3: Logger with levels and filtering (Medium)

Extend StructuredLogger to support levels (DEBUG, INFO, WARN, ERROR) and a configurable minimum level. If the minimum level is WARN, INFO and DEBUG logs aren't emitted. Test it with a pipeline that generates logs at every level.

from dotenv import load_dotenv
load_dotenv()

import json
import time
import logging
from langgraph.func import entrypoint, task

logging.basicConfig(level=logging.INFO, format="%(message)s")


class LeveledLogger:
    LEVELS = {"DEBUG": 0, "INFO": 1, "WARN": 2, "ERROR": 3}

    def __init__(self, min_level: str = "INFO"):
        self._min = self.LEVELS.get(min_level, 1)
        self._logger = logging.getLogger("leveled")
        self._rid = None

    def set_request_id(self, rid: str):
        self._rid = rid

    def _emit(self, level: str, node: str, message: str, **kw):
        if self.LEVELS.get(level, 0) < self._min:
            return
        entry = {
            "ts": time.strftime("%H:%M:%S"),
            "level": level,
            "rid": self._rid,
            "node": node,
            "msg": message,
            **kw,
        }
        self._logger.info(json.dumps(entry, ensure_ascii=False))

    def debug(self, node, msg, **kw): self._emit("DEBUG", node, msg, **kw)
    def info(self, node, msg, **kw): self._emit("INFO", node, msg, **kw)
    def warn(self, node, msg, **kw): self._emit("WARN", node, msg, **kw)
    def error(self, node, msg, **kw): self._emit("ERROR", node, msg, **kw)


log_info = LeveledLogger(min_level="INFO")
log_warn = LeveledLogger(min_level="WARN")


@task
def demo_task(value: int, logger: LeveledLogger) -> dict:
    logger.debug("demo", f"Processing value {value}", value=value)
    logger.info("demo", f"Value received: {value}")

    if value > 5:
        logger.warn("demo", f"High value: {value}", threshold=5)
    if value > 8:
        logger.error("demo", f"Critical value: {value}", threshold=8)

    return {"value": value, "processed": True}


@entrypoint()
def logging_demo(config: dict) -> dict:
    logger = log_info if config.get("verbose") else log_warn
    logger.set_request_id("demo-001")

    values = config["values"]
    futures = [demo_task(v, logger) for v in values]
    results = [f.result() for f in futures]

    return {"processed": len(results)}


print("=== With min_level=INFO (verbose) ===")
logging_demo.invoke({"values": [2, 6, 9], "verbose": True})

print("\n=== With min_level=WARN (warnings and up only) ===")
logging_demo.invoke({"values": [2, 6, 9], "verbose": False})
# Expected output:
# === With min_level=INFO (verbose) ===
# {"ts": "14:30:00", "level": "INFO", "rid": "demo-001", "node": "demo", "msg": "Value received: 2"}
# {"ts": "14:30:00", "level": "INFO", "rid": "demo-001", "node": "demo", "msg": "Value received: 6"}
# {"ts": "14:30:00", "level": "WARN", "rid": "demo-001", "node": "demo", "msg": "High value: 6", "threshold": 5}
# {"ts": "14:30:00", "level": "INFO", "rid": "demo-001", "node": "demo", "msg": "Value received: 9"}
# {"ts": "14:30:00", "level": "WARN", "rid": "demo-001", "node": "demo", "msg": "High value: 9", "threshold": 5}
# {"ts": "14:30:00", "level": "ERROR", "rid": "demo-001", "node": "demo", "msg": "Critical value: 9", "threshold": 8}
#
# === With min_level=WARN (warnings and up only) ===
# {"ts": "14:30:00", "level": "WARN", "rid": "demo-001", "node": "demo", "msg": "High value: 6", "threshold": 5}
# {"ts": "14:30:00", "level": "WARN", "rid": "demo-001", "node": "demo", "msg": "High value: 9", "threshold": 5}
# {"ts": "14:30:00", "level": "ERROR", "rid": "demo-001", "node": "demo", "msg": "Critical value: 9", "threshold": 8}

Exercise 4: Metrics dashboard with a summary (Medium)

Extend MetricsCollector to generate a text dashboard that includes: the top 3 slowest nodes, the overall success rate, total tokens consumed, and alerts for nodes with a success rate below 90%.

from dotenv import load_dotenv
load_dotenv()

import time
import statistics
from collections import defaultdict
from langgraph.func import entrypoint, task


class DashboardMetrics:
    def __init__(self):
        self.latencies = defaultdict(list)
        self.counts = defaultdict(lambda: {"ok": 0, "err": 0})
        self.tokens = defaultdict(int)

    def record(self, node: str, ms: float, ok: bool, tokens: int = 0):
        self.latencies[node].append(ms)
        self.counts[node]["ok" if ok else "err"] += 1
        if tokens:
            self.tokens[node] += tokens

    def dashboard(self) -> str:
        lines = []
        lines.append("╔══════════════════════════════════════════════════╗")
        lines.append("║          📊 PRODUCTION DASHBOARD                ║")
        lines.append("╚══════════════════════════════════════════════════╝")

        total_ok = sum(c["ok"] for c in self.counts.values())
        total_err = sum(c["err"] for c in self.counts.values())
        total_all = total_ok + total_err
        total_tokens = sum(self.tokens.values())

        lines.append(f"\n  🎯 Global success rate: {total_ok}/{total_all} ({total_ok/total_all:.0%})" if total_all else "")
        lines.append(f"  🪙 Total tokens: {total_tokens}")
        lines.append(f"  📦 Nodes tracked: {len(self.latencies)}")

        avg_by_node = {
            n: statistics.mean(lats) for n, lats in self.latencies.items()
        }
        slowest = sorted(avg_by_node.items(), key=lambda x: x[1], reverse=True)[:3]

        lines.append(f"\n  🐌 TOP 3 SLOWEST NODES:")
        for i, (node, avg) in enumerate(slowest, 1):
            lines.append(f"     {i}. {node}: {avg:.1f}ms avg")

        alerts = []
        for node, c in self.counts.items():
            total = c["ok"] + c["err"]
            if total > 0 and c["ok"] / total < 0.9:
                rate = c["ok"] / total
                alerts.append(f"     ⚠️ {node}: {rate:.0%} success rate ({c['err']} errors)")

        if alerts:
            lines.append(f"\n  🚨 ALERTS:")
            lines.extend(alerts)
        else:
            lines.append(f"\n  ✅ No alerts — every node above 90% success rate")

        return "\n".join(lines)


dm = DashboardMetrics()


@task
def reliable_node(name: str, delay: float) -> str:
    time.sleep(delay)
    dm.record(name, delay * 1000, ok=True, tokens=50)
    return f"{name} OK"


@task
def flaky_node(name: str, delay: float, fail_rate: float) -> str:
    import random
    time.sleep(delay)
    if random.random() < fail_rate:
        dm.record(name, delay * 1000, ok=False)
        raise RuntimeError(f"{name} failed")
    dm.record(name, delay * 1000, ok=True, tokens=30)
    return f"{name} OK"


@entrypoint()
def demo_dashboard(runs: int) -> str:
    for _ in range(runs):
        reliable_node("search_web", 0.2).result()
        reliable_node("synthesize", 0.8).result()
        try:
            flaky_node("search_academic", 0.3, 0.3).result()
        except Exception:
            pass

    return dm.dashboard()


print(demo_dashboard.invoke(10))
# Expected output:
# ╔══════════════════════════════════════════════════╗
# ║          📊 PRODUCTION DASHBOARD                ║
# ╚══════════════════════════════════════════════════╝
#
#   🎯 Global success rate: 27/30 (90%)
#   🪙 Total tokens: 1210
#   📦 Nodes tracked: 3
#
#   🐌 TOP 3 SLOWEST NODES:
#      1. synthesize: 800.0ms avg
#      2. search_academic: 300.0ms avg
#      3. search_web: 200.0ms avg
#
#   🚨 ALERTS:
#      ⚠️ search_academic: 70% success rate (3 errors)

Exercise 5: Pipeline with a circuit breaker (Advanced)

Implement a circuit breaker: if a node fails more than 3 times in a row, the next calls to that node return an error immediately (without trying) for 30 seconds. After 30 seconds, it allows one test call ("half-open"). If that call succeeds, the circuit closes (normal). If it fails, the circuit opens again.

from dotenv import load_dotenv
load_dotenv()

import time
from langgraph.func import entrypoint, task


class CircuitBreaker:
    """Circuit breaker: CLOSED → OPEN → HALF_OPEN → CLOSED/OPEN."""

    def __init__(self, failure_threshold: int = 3, recovery_timeout: float = 30.0):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self._states: dict[str, dict] = {}

    def _get_state(self, name: str) -> dict:
        if name not in self._states:
            self._states[name] = {
                "status": "CLOSED",
                "consecutive_failures": 0,
                "opened_at": 0,
            }
        return self._states[name]

    def can_execute(self, name: str) -> tuple[bool, str]:
        state = self._get_state(name)

        if state["status"] == "CLOSED":
            return True, "CLOSED"

        if state["status"] == "OPEN":
            elapsed = time.time() - state["opened_at"]
            if elapsed >= self.recovery_timeout:
                state["status"] = "HALF_OPEN"
                return True, "HALF_OPEN"
            return False, f"OPEN ({self.recovery_timeout - elapsed:.0f}s remaining)"

        if state["status"] == "HALF_OPEN":
            return True, "HALF_OPEN"

        return False, state["status"]

    def record_success(self, name: str):
        state = self._get_state(name)
        state["consecutive_failures"] = 0
        state["status"] = "CLOSED"

    def record_failure(self, name: str):
        state = self._get_state(name)
        state["consecutive_failures"] += 1

        if state["consecutive_failures"] >= self.failure_threshold:
            state["status"] = "OPEN"
            state["opened_at"] = time.time()


cb = CircuitBreaker(failure_threshold=3, recovery_timeout=5.0)

call_count = 0


@task
def protected_search(query: str, source: str) -> dict:
    global call_count
    can_exec, status = cb.can_execute(source)

    if not can_exec:
        return {"source": source, "status": "circuit_open", "circuit": status}

    try:
        call_count += 1
        if source == "flaky_api" and call_count <= 4:
            raise ConnectionError(f"flaky_api failure #{call_count}")

        cb.record_success(source)
        return {"source": source, "status": "ok", "content": f"Result from {source}"}

    except Exception as e:
        cb.record_failure(source)
        return {"source": source, "status": "error", "error": str(e)}


@entrypoint()
def circuit_demo(config: dict) -> list:
    results = []

    for i in range(config["num_calls"]):
        r = protected_search(config["query"], config["source"]).result()
        results.append({"call": i + 1, **r})

        if config.get("sleep_between"):
            time.sleep(config["sleep_between"])

    return results


results = circuit_demo.invoke({
    "query": "test",
    "source": "flaky_api",
    "num_calls": 8,
    "sleep_between": 1.0,
})

for r in results:
    print(f"  Call {r['call']}: {r['status']} | {r.get('circuit', r.get('error', r.get('content', '')))}")
# Expected output:
#   Call 1: error | flaky_api failure #1
#   Call 2: error | flaky_api failure #2
#   Call 3: error | flaky_api failure #3         ← circuit OPENS here
#   Call 4: circuit_open | OPEN (2s remaining)  ← calls blocked
#   Call 5: circuit_open | OPEN (1s remaining)  ← calls blocked
#   Call 6: error | flaky_api failure #4          ← HALF_OPEN test call fails → re-opens
#   Call 7: circuit_open | OPEN (4s remaining)  ← blocked again
#   Call 8: circuit_open | OPEN (3s remaining)  ← blocked again

Exercise 6: Complete production-ready pipeline (Advanced)

Combine timeouts, rate limiting, logging, metrics, and a circuit breaker in a pipeline that searches 3 sources. One source must be "flaky" (it fails 50% of the time). The pipeline must: respect timeouts, log every operation, record metrics, and use a circuit breaker for the flaky source.

from dotenv import load_dotenv
load_dotenv()

import json
import time
import random
import logging
import statistics
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError
from langgraph.func import entrypoint, task

logging.basicConfig(level=logging.INFO, format="%(message)s")


class MiniCircuitBreaker:
    def __init__(self, threshold=3, timeout=10.0):
        self.threshold = threshold
        self.timeout = timeout
        self._fails = defaultdict(int)
        self._opened = {}

    def allow(self, name):
        if name in self._opened:
            if time.time() - self._opened[name] > self.timeout:
                del self._opened[name]
                return True
            return False
        return True

    def success(self, name):
        self._fails[name] = 0
        self._opened.pop(name, None)

    def failure(self, name):
        self._fails[name] += 1
        if self._fails[name] >= self.threshold:
            self._opened[name] = time.time()


class MiniMetrics:
    def __init__(self):
        self.data = defaultdict(lambda: {"lats": [], "ok": 0, "err": 0})

    def record(self, node, ms, ok):
        self.data[node]["lats"].append(ms)
        self.data[node]["ok" if ok else "err"] += 1

    def report(self):
        for node, d in sorted(self.data.items()):
            total = d["ok"] + d["err"]
            rate = d["ok"] / total if total else 0
            avg = statistics.mean(d["lats"]) if d["lats"] else 0
            print(f"  {node}: {total} calls, {rate:.0%} ok, avg {avg:.0f}ms")


cb = MiniCircuitBreaker(threshold=2, timeout=5.0)
met = MiniMetrics()


def _search_impl(query, source):
    delays = {"reliable_a": 0.3, "flaky_b": 0.4, "reliable_c": 0.2}
    time.sleep(delays.get(source, 0.3))
    if source == "flaky_b" and random.random() < 0.5:
        raise ConnectionError(f"{source} failed")
    return f"[{source}] Results for '{query}'"


@task
def guarded_search(query: str, source: str) -> dict:
    if not cb.allow(source):
        met.record(source, 0, False)
        return {"source": source, "status": "circuit_open"}

    start = time.time()
    with ThreadPoolExecutor(max_workers=1) as ex:
        fut = ex.submit(_search_impl, query, source)
        try:
            content = fut.result(timeout=2.0)
            ms = (time.time() - start) * 1000
            cb.success(source)
            met.record(source, ms, True)
            return {"source": source, "status": "ok", "content": content}
        except FuturesTimeoutError:
            ms = (time.time() - start) * 1000
            cb.failure(source)
            met.record(source, ms, False)
            return {"source": source, "status": "timeout"}
        except Exception as e:
            ms = (time.time() - start) * 1000
            cb.failure(source)
            met.record(source, ms, False)
            return {"source": source, "status": "error", "error": str(e)}


@entrypoint()
def full_production_pipeline(config: dict) -> dict:
    topic = config["topic"]
    sources = config["sources"]
    all_run_results = []

    for run in range(config.get("runs", 1)):
        futures = [guarded_search(topic, s) for s in sources]
        results = [f.result() for f in futures]
        ok = [r for r in results if r["status"] == "ok"]
        all_run_results.append({"run": run + 1, "ok": len(ok), "total": len(results)})

    return {"runs": all_run_results}


result = full_production_pipeline.invoke({
    "topic": "AI production patterns",
    "sources": ["reliable_a", "flaky_b", "reliable_c"],
    "runs": 6,
})

print("\n📊 Results per run:")
for r in result["runs"]:
    print(f"  Run {r['run']}: {r['ok']}/{r['total']} sources OK")

print("\n📈 Accumulated metrics:")
met.report()
# Expected output (varies with flaky_b's random):
# 📊 Results per run:
#   Run 1: 3/3 sources OK      (or 2/3 if flaky_b failed)
#   Run 2: 2/3 sources OK
#   Run 3: 2/3 sources OK      (circuit opens for flaky_b)
#   Run 4: 2/3 sources OK      (circuit_open — doesn't even try)
#   Run 5: 2/3 sources OK      (circuit_open)
#   Run 6: 2/3 sources OK      (circuit_open or half_open)
#
# 📈 Accumulated metrics:
#   flaky_b: 6 calls, 33% ok, avg 180ms
#   reliable_a: 6 calls, 100% ok, avg 302ms
#   reliable_c: 6 calls, 100% ok, avg 201ms

Summary

In this capsule you learned the 4 patterns that separate a working prototype from a production system:

• Per-node timeouts — They keep a slow API from blocking the whole pipeline. Use asyncio.wait_for for async code or ThreadPoolExecutor for synchronous code. Without a timeout, one slow node freezes every user
• Internal rate limiting — LangChain's InMemoryRateLimiter controls how many requests per second you send to each service. Without rate limiting, you blow past the API's limits and get cut off
• Structured logging — JSON with a request_id, the node name, duration, and status. Correlation IDs to trace a request across every node. Without structured logging, "something failed" is your best diagnosis
• Performance metrics — Latency per node (avg, p50, p95), success rate, tokens consumed. They identify bottlenecks and answer business questions. Without metrics, you optimize blind

Together, these patterns make your agent operable: you can diagnose problems, answer business questions, and scale with confidence. They're the equivalent of instrumenting a web server with logs, metrics, and health checks — standard in the industry, but frequently forgotten in AI engineering.

Next capsule: Evolving Project — you're going to apply retry logic, parallel branching, and these production patterns to Research Agent v1 to turn it into v2.

Additional resources

1. LangChain Rate Limiters — Documentation for InMemoryRateLimiter and its integration with models
2. Python asyncio.wait_for — Timeouts with asyncio for asynchronous code
3. Python logging JSON — Official guide to logging with custom formatters
4. Circuit Breaker Pattern — Martin Fowler on circuit breakers (the original article)
5. LangGraph Functional API — Reference for @entrypoint and @task
6. Python concurrent.futures — ThreadPoolExecutor for synchronous timeouts

Module 7 — LangChain & LangGraph: From Chains to Agents