langchain-sdk-patterns

What this skill does

# LangChain SDK Patterns (Python)

## Overview

`chain.batch(inputs)` in LangChain 1.0 does **not** parallelize by default. The
`max_concurrency` parameter defaults to **1** in several provider packages
(notably older `langchain-openai`), so a call like `chain.batch(inputs_1000)`
runs 1,000 sequential round-trips — same wall-clock time as a `for` loop, plus
the overhead of the batch machinery. Users file "batch is slow" tickets,
benchmark it against asyncio, and move to a different framework — when the fix
is two lines:

```python
# BAD — silently serializes (P08)
chain.batch(inputs_1000)

# GOOD — 10 in flight at once
chain.batch(inputs_1000, config={"max_concurrency": 10})
```

Then three more traps wait:

- **P07** — `.with_fallbacks([backup])` defaults `exceptions_to_handle=(Exception,)`,
  and on Python <3.12 that tuple includes `KeyboardInterrupt`. A `Ctrl+C` during
  a long run does not stop the process — it silently hands off to the fallback
  chain and keeps billing.
- **P57** — `ChatPromptTemplate.from_messages(..., template_format="f-string")`
  (the default) parses every `{` in every string, including user input. A user
  who pastes `{"error": "..."}` raises `KeyError: 'error'` at invoke time.
- **P53** — Pydantic v2 rejects extra fields by default; models cheerfully add
  `summary` or `confidence` to your `Plan` schema and `with_structured_output`
  crashes with `ValidationError: extra fields not permitted`.

This skill walks through LCEL composition (`RunnableSequence`, `RunnableParallel`,
`RunnableBranch`, `RunnablePassthrough`, `RunnableLambda`); the correct
`exceptions_to_handle` whitelist per provider; `max_concurrency` tuning with
safe ceilings (10 for most providers, 20+ with a semaphore); and prompt
templates that survive untrusted input. Pin: `langchain-core 1.0.x`,
`langchain-anthropic 1.0.x`, `langchain-openai 1.0.x`. Pain-catalog anchors:
P07, P08, P53, P57.

## Prerequisites

- Python 3.10+ (3.12+ fixes the `KeyboardInterrupt` half of P07 — upgrade if you can)
- `langchain-core >= 1.0, < 2.0`
- At least one provider: `pip install langchain-anthropic langchain-openai`
- `pydantic >= 2.0` for schema-aware composition
- Completed `langchain-model-inference` — the chat-model factory from that skill is reused here

## Instructions

### Step 1 — Compose with typed runnables, not lambdas

```python
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough

llm = ChatAnthropic(model="claude-sonnet-4-6", timeout=30, max_retries=2)

prompt = ChatPromptTemplate.from_messages(
    [("system", "You are a summarizer."), ("human", "{text}")],
    template_format="jinja2",  # P57 — see Step 4
)

# Sequence: prompt -> llm -> str
chain = prompt | llm | StrOutputParser()

# Parallel: run two sub-chains and merge
enriched = RunnableParallel(
    summary=chain,
    original=RunnablePassthrough(),
)
```

The `|` operator creates a `RunnableSequence`. Each step has a declared input
and output shape — swap a concrete model for a router and the type contract
holds. See [Runnable Composition Matrix](references/runnable-composition-matrix.md)
for when to reach for `RunnableSequence` vs `RunnableParallel` vs `RunnableBranch`
vs `RunnableLambda`, with input/output shape conventions for each.

### Step 2 — Add fallbacks with a narrow exception whitelist

```python
from anthropic import APIError, APITimeoutError, RateLimitError
from langchain_openai import ChatOpenAI

backup = ChatOpenAI(model="gpt-4o", timeout=30, max_retries=2)
backup_chain = prompt | backup | StrOutputParser()

# GOOD — only retry on transient provider errors
resilient = chain.with_fallbacks(
    [backup_chain],
    exceptions_to_handle=(RateLimitError, APIError, APITimeoutError),
)

# BAD — default `(Exception,)` catches KeyboardInterrupt on Python <3.12 (P07)
# resilient_bad = chain.with_fallbacks([backup_chain])
```

The default `exceptions_to_handle=(Exception,)` on Python <3.12 inherits
`KeyboardInterrupt` and `SystemExit` into the caught set — which means a
`Ctrl+C` during a long `.batch()` run falls through to the backup instead of
stopping. Python 3.12+ moved these under `BaseException` directly, which fixes
the inheritance path, but the default is still too broad: a Pydantic
`ValidationError` or a `ToolException` will trigger a pointless backup call.
See [Fallback Exception List](references/fallback-exception-list.md) for the
curated whitelist per provider with concrete imports.

### Step 3 — Batch with explicit concurrency

```python
import asyncio

inputs = [{"text": doc} for doc in documents]

# Synchronous batch — blocks until done
results = chain.batch(inputs, config={"max_concurrency": 10})

# Async batch — non-blocking
results = await chain.abatch(inputs, config={"max_concurrency": 10})
```

Safe ceilings: **10** for Anthropic and OpenAI at default tier; **20+** only
behind an `asyncio.Semaphore` if you are also tracking rate-limit headers.
Claude TPM/RPM limits vary by tier; OpenAI's TPD (tokens per day) is the
binding limit at scale. See [Batch Concurrency Tuning](references/batch-concurrency-tuning.md)
for per-provider ceilings and the semaphore pattern.

`invoke` vs `batch` vs `stream` — when each is correct:

| Method | Input shape | Concurrency | Error behavior | When to use |
|---|---|---|---|---|
| `.invoke(x)` | Single | 1 | Raises on failure | One-shot call, interactive, tests |
| `.batch(xs, config={"max_concurrency": N})` | List | N parallel | Raises on first failure unless `return_exceptions=True` | Bulk sync workloads, ETL, eval harnesses |
| `.abatch(xs, config={"max_concurrency": N})` | List | N parallel (async) | Same as `.batch` | Event loops, async web servers, LangGraph nodes |
| `.stream(x)` | Single | 1, chunked | Raises on failure | Interactive UI, live token display |
| `.astream(x)` / `.astream_events(x, version="v2")` | Single | 1, chunked (async) | Raises on failure | Async UIs, event-driven pipelines, token metering (see `langchain-model-inference`) |

Pass `return_exceptions=True` in the config to keep a batch from aborting on
the first failure — exceptions come back in the result list instead of raising.

### Step 4 — Escape prompt templates for untrusted input

```python
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# BAD — default f-string format crashes on literal `{` in user input (P57)
bad = ChatPromptTemplate.from_messages(
    [("system", "Reply in JSON"), ("human", "{user_text}")]
)
bad.invoke({"user_text": '{"error": "oops"}'})  # KeyError: 'error'

# GOOD — jinja2 treats `{...}` as literal, uses `{{ var }}` for substitution
good = ChatPromptTemplate.from_messages(
    [("system", "Reply in JSON"), ("human", "{{ user_text }}")],
    template_format="jinja2",
)
good.invoke({"user_text": '{"error": "oops"}'})  # OK

# MIXED — message history is a list, use MessagesPlaceholder
with_history = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    MessagesPlaceholder("history"),
    ("human", "{{ question }}"),
], template_format="jinja2")
```

Rule of thumb: if any variable can contain user-provided free text (a paste,
a transcript, a code block), use `template_format="jinja2"`. The f-string
format is fine for trusted template authors composing fixed instructions, but
it is the wrong tool for user input. See [Prompt Template Escaping](references/prompt-template-escaping.md)
for the full brace-escaping rules and a `MessagesPlaceholder` reference.

### Step 5 — Validate structured output with `extra="ignore"`

```python
from pydantic import BaseModel, ConfigDict, Field

class Plan(BaseModel):
    # P53 — without this, the chain crashes when the model adds extra fields
    model_config = ConfigDict(extra="ignore")
    steps: list[str] = Field(default_factory=list)
    estimated_minutes: int

structured_c