Compose LangChain 1.0 Python runnables with the production defaults the docs
do not warn about: parallel batching, narrow fallbacks, and brace-safe prompts.
Use when building an LCEL chain with RunnableSequence / RunnableParallel,
adding resilience via .with_fallbacks(), tuning throughput with .batch()
or .abatch(), or wrapping user input in a prompt template.
Trigger with "langchain runnable", "with_fallbacks", "langchain batch",
"runnable sequence", "lcel", "runnableparallel", "chain composition".
Instalar com Codex ou Claude Copie este prompt, cole no Codex, Claude ou outro assistente e deixe que ele revise a página da skill e instale para você.
Um comando direto ignora o prompt de revisão. Verifique a origem antes de executá-lo.
Instruções da origem · Visualização somente leitura
name
langchain-sdk-patterns
description
Compose LangChain 1.0 Python runnables with the production defaults the docs
do not warn about: parallel batching, narrow fallbacks, and brace-safe prompts.
Use when building an LCEL chain with RunnableSequence / RunnableParallel,
adding resilience via .with_fallbacks(), tuning throughput with .batch()
or .abatch(), or wrapping user input in a prompt template.
Trigger with "langchain runnable", "with_fallbacks", "langchain batch",
"runnable sequence", "lcel", "runnableparallel", "chain composition".
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:
# 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, . Pain-catalog anchors:
P07, P08, P53, P57.
langchain-openai 1.0.x
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", # 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
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()
# 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 for the
curated whitelist per provider with concrete imports.
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 — 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
for the full brace-escaping rules and a MessagesPlaceholder reference.
Step 5 — Validate structured output with extra="ignore"
from pydantic import BaseModel, ConfigDict, Field
classPlan(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_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
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.
