What is .batch()?
.batch() processes a list of inputs through a Runnable in parallel and returns a list of results in the same order as the input. Under the hood, LangChain uses a ThreadPoolExecutor for synchronous runnables (subject to a default concurrency cap of 5), or asyncio.gather for async ones. You can override the cap by passing max_concurrency inside the config argument.
The key advantage over a for-loop is throughput: running 20 embeddings sequentially might take 20 seconds; .batch() with max_concurrency=10 can cut that to roughly 2 seconds, limited by the provider's rate limits. Results preserve input order even when worker threads complete out of order — LangChain handles the re-ordering internally. For per-input tracing, you can pass a list of RunnableConfig dicts (one per input) instead of a single config, which lets you attach different callbacks, tags, or metadata to each item — useful in multi-tenant workloads.
Error handling: by default, one failing input raises and the entire batch aborts. Pass return_exceptions=True inside config to collect errors as Exception objects at the failed positions instead of raising, which lets you inspect which inputs failed and retry selectively without re-running the successful ones. In async contexts, prefer .abatch() — the sync .batch() creates a new event loop internally and blocks if called from within an existing async context.
Use Cases
- • Bulk LLM processing
- • Batch embeddings
- • Parallel API calls
- • Data pipelines
- • Efficiency
- • Cost reduction
Key Features
- ✓ Parallel execution
- ✓ Order preserved
- ✓ Concurrency control
- ✓ Error isolation
- ✓ Shared config
- ✓ Speed improvement
When NOT to Use
For single items—use invoke(). In async contexts—use abatch(). For streaming output—use stream() or astream().
Notes
Default concurrency is 5 — tune per workload
The default max_concurrency=5 is conservative. Embedding models handle 50+ concurrent calls comfortably; GPT-4o at peak hours may need 2–3. Profile your provider's rate limit headers and adjust accordingly rather than leaving the default.
Rate limits are your real throughput ceiling
.batch() will trigger provider rate limits (tokens per minute, requests per minute) much faster than sequential calls. Pair it with an exponential-backoff retry callback or a rate-limiter wrapper. The langchain_community.callbacks.RateLimiter class provides token-bucket rate limiting.
Conversation memory is not thread-safe with batch
ConversationBufferMemory and similar stateful memory classes are not safe for concurrent write from multiple threads. Use .batch() for stateless runnables (embeddings, classification, extraction). For per-user chat history, keep memory objects separate per user and use sequential invoke per user.
Use abatch() in async contexts
In an async framework like FastAPI, calling .batch() blocks the event loop because it creates a new thread pool internally. Use await runnable.abatch(inputs) instead — it uses asyncio.gather and keeps the event loop free for other requests.
Method Signature
results = runnable.batch(inputs, config=None, return_exceptions=False)Parameters
| Parameter | Type | Required | Purpose |
|---|---|---|---|
| inputs | List[Any] | Yes | List of inputs to process in parallel |
| config | RunnableConfig | List[RunnableConfig] | None | No | Shared config or list of per-input configs |
| return_exceptions | bool | No | Return errors as values instead of raising |
Return Value
Type:
List[Any]
Description:
Results in same order as inputs
Example Output:
[result1, result2, ...]Code Examples
Batch LLM questions
questions = ['What is AI?', 'What is ML?', 'What is RAG?']
results = model.batch(questions)
for q, r in zip(questions, results):
print(f'{q}: {r.content[:60]}')Batch with max_concurrency
from langchain_openai import OpenAIEmbeddings
embedder = OpenAIEmbeddings(model="text-embedding-3-small")
texts = ["Document one content", "Document two content", "Document three"]
# max_concurrency=10 for embeddings (higher rate limits than chat)
vectors = embedder.embed_documents(texts) # already uses batching internally
# Or with a runnable pipeline:
from langchain_core.runnables import RunnableLambda
classify = RunnableLambda(lambda text: {"label": "positive" if "good" in text else "negative"})
results = classify.batch(texts, config={"max_concurrency": 10})return_exceptions for partial failure isolation
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
model = ChatOpenAI(model="gpt-4o")
inputs = [
[HumanMessage(content="Summarise: Python is great")],
[HumanMessage(content="Summarise: " + "x" * 200000)], # too long
[HumanMessage(content="Summarise: LangChain simplifies LLM apps")],
]
results = model.batch(
inputs,
config={"max_concurrency": 3},
return_exceptions=True,
)
for i, r in enumerate(results):
if isinstance(r, Exception):
print(f"Input {i} failed: {r}")
else:
print(f"Input {i}: {r.content[:60]}")Common Mistakes
❌ for input in inputs: model.invoke(input) # Slow
✅ model.batch(inputs) # Fast and efficient
Related LangChain References
Browse the full LangChain API reference index to explore more classes, methods, and decorators, or start with the LangChain introduction tutorial for end-to-end context on building with .batch() and the wider framework.
.batch() FAQ
What does .batch() do in LangChain?
Process multiple inputs in parallel. .batch() processes a list of inputs through a Runnable in parallel and returns a list of results in the same order as the input. Under the hood, LangChain uses a ThreadPoolExecutor for synchronous runnables (subject to a default concurrency cap of 5), or asyncio.gather for async ones. You can override the cap by passing max_concurrency inside the config argument. The key advantage over a for-loop is throughput: running 20 embeddings sequentially might take 20 seconds; .batch…
Which LangChain classes support .batch()?
.batch() is available on All Runnables. Pin your installed LangChain version and verify the method exists in that release before deploying.
When should I use .batch()?
Use .batch() when your LangChain chains, agents, or pipelines need the behavior described in this guide.
What does .batch() return?
.batch() returns a List[Any]. Results in same order as inputs
Does .batch() have an async equivalent?
Yes — use abatch() for async contexts such as FastAPI handlers or asyncio-based pipelines. It is the non-blocking counterpart to .batch().
Where can I explore more LangChain API reference pages?
Open the LangChain API reference index on DevShelfHub to browse classes, methods, and decorators, each with runnable examples, parameters, common mistakes, and cross-links.