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:
chain.batch(inputs_1000)
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
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",
)
chain = prompt | llm | StrOutputParser()
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
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
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()
resilient = chain.with_fallbacks(
[backup_chain],
exceptions_to_handle=(RateLimitError, APIError, APITimeoutError),
)
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 for the
curated whitelist per provider with concrete imports.
Step 3 — Batch with explicit concurrency
import asyncio
inputs = [{"text": doc} for doc in documents]
results = chain.batch(inputs, config={"max_concurrency": 10})
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
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
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
bad = ChatPromptTemplate.from_messages(
[("system", "Reply in JSON"), ("human", "{user_text}")]
)
bad.invoke({"user_text": '{"error": "oops"}'})
good = ChatPromptTemplate.from_messages(
[("system", "Reply in JSON"), ("human", "{{ user_text }}")],
template_format="jinja2",
)
good.invoke({"user_text": '{"error": "oops"}'})
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
for the full brace-escaping rules and a MessagesPlaceholder reference.
Step 5 — Validate structured output with extra="ignore"
from pydantic import BaseModel, ConfigDict, Field
class Plan(BaseModel):
model_config = ConfigDict(extra="ignore")
steps: list[str] = Field(default_factory=list)
estimated_minutes: int
structured_chain = prompt | llm.with_structured_output(Plan, method="json_schema")
Pydantic v2 rejects unknown fields by default. Models trained on "be helpful"
add summary, confidence, rationale — the schema crashes instead of
dropping them. extra="ignore" is the right default for model outputs.
Output
RunnableSequence / RunnableParallel composition with declared input/output shapes.with_fallbacks(exceptions_to_handle=(...)) with a narrow, provider-specific whitelist.batch() / .abatch() with explicit max_concurrency (10 default, 20+ behind semaphore)ChatPromptTemplate.from_messages(..., template_format="jinja2") for any template touching user input- Pydantic schemas with
ConfigDict(extra="ignore") for structured output
- A clear
invoke / batch / abatch / stream / astream decision matrix for each chain stage
Error Handling
| Error | Cause | Fix |
|---|
Ctrl+C does not stop a long .batch(); fallback keeps running | exceptions_to_handle=(Exception,) swallows KeyboardInterrupt on Python <3.12 (P07) | Pass a narrow tuple: exceptions_to_handle=(RateLimitError, APIError, APITimeoutError) |
.batch(inputs) takes same time as sequential loop | max_concurrency defaults to 1 (P08) | config={"max_concurrency": 10}; raise to 20+ only with a semaphore |
KeyError: '<some-token>' when invoking a ChatPromptTemplate | f-string parser reads user input's { as a variable (P57) | template_format="jinja2"; escape literals as {{/}} in f-string mode |
ValidationError: extra fields not permitted on structured output | Pydantic v2 strict-by-default (P53) | model_config = ConfigDict(extra="ignore") on the schema |
ValidationError caught by fallback and treated as transient | Fallback whitelist too broad | Remove ValidationError from exceptions_to_handle so it surfaces |
.batch aborts on the first failure, losing all results | Default raises on first error | Pass config={"max_concurrency": 10, "return_exceptions": True} and filter |
Fallback chain never fires even on genuine RateLimitError | Provider's own max_retries consumes the error first | Lower max_retries=0 on the primary when a fallback chain is the retry strategy |
Examples
Fan-out enrichment with RunnableParallel
A common pattern — given a document, produce a summary, extracted entities,
and sentiment in parallel. RunnableParallel runs sub-chains concurrently and
merges results into a dict. Combined with .batch() at the outer level, you
get N documents times 3 sub-chains in flight up to max_concurrency.
See Runnable Composition Matrix
for the fan-out/fan-in pattern and the input/output shape of each runnable type.
Resilient chain with per-provider fallback
Primary: Claude Sonnet 4.6. Fallback: GPT-4o. Catch only RateLimitError,
APIError, and APITimeoutError from each SDK — let AuthenticationError
and ValidationError crash the process so they get debugged, not masked.
See Fallback Exception List for the
concrete imports per provider and a note on why BadRequestError should not
be in the whitelist.
High-throughput batch with semaphore-bounded concurrency
At N >= 20 concurrent in-flight calls, provider rate-limit headers become the
bottleneck. Wrap .abatch() in an asyncio.Semaphore and honor the
retry-after header on 429 responses.
See Batch Concurrency Tuning for the
semaphore pattern and a table of provider TPM/RPM limits per tier.
Prompt template over user-pasted JSON payload
Support ticket triage where users paste arbitrary JSON from their app's error
log. Without template_format="jinja2", every single ticket with a JSON body
crashes the chain at template-render time.
See Prompt Template Escaping for the
worked example and the MessagesPlaceholder pattern for chat history.
Resources
