LangChain Async, Streaming and Batch Execution

Streaming UX

Real LLM applications need more than invoke. Chat UIs stream tokens, data pipelines process many records, and APIs must avoid blocking workers for long-running model calls. LangChain runnables support async, streaming, and batch-style execution so you can match the runtime behavior to the product.

Performance work should be deliberate. Streaming improves perceived latency. Batch execution improves throughput. Async prevents the web server from waiting wastefully. Concurrency limits protect your budget and provider rate limits.

Streaming is useful when answers are long or users need immediate feedback. Your frontend should handle partial text, cancellation, errors after partial output, and final metadata such as sources.

Stream only user-safe text.
Keep citations or source metadata available at the end.
Support cancellation so users can stop expensive responses.

Batch and Concurrency

Batch calls are useful for classification, extraction, evaluation, and offline enrichment. Add concurrency limits so a batch job does not overwhelm rate limits or create surprise costs.

Record failed inputs and retry them separately.
Use structured output for extraction jobs.
Log token usage and latency per item.

Async API Handler

Async invocation fits web APIs that need to keep workers responsive.

Async API Handler

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()
chain = build_rag_chain()

class Question(BaseModel):
    question: str

@app.post("/ask")
async def ask(payload: Question):
    answer = await chain.ainvoke(payload.question)
    return {"answer": answer}

Use async only when the rest of your stack supports it cleanly.
Still set timeouts and rate limits at the service boundary.

Batch Evaluation Inputs

Batch execution is useful for regression tests and offline jobs.

Batch Evaluation Inputs

questions = [
    "How do I configure SSO?",
    "What is the refund window?",
    "Can I delete audit logs?",
]

answers = chain.batch(
    questions,
    config={"max_concurrency": 3},
)

for question, answer in zip(questions, answers):
    print("\nQUESTION:", question)
    print("ANSWER:", answer[:500])

Limit concurrency to avoid provider rate limits.
For large jobs, persist progress so failed runs can resume.

Before you move on