LangChain in Production: Escape Tutorial Hell with Battle-Tested Patterns for Enterprise RAG and Agents

LangChain tutorials make everything look simple. Production LangChain is a different beast — requiring streaming response handling, robust error recovery, observability instrumentation, and careful chain composition.

1. Production Chain Pattern

from langchain_core.runnables import RunnableParallel, RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# Production chain with error handling and fallbacks
retrieval_chain = (
    RunnableParallel(
        context=retriever | RunnableLambda(format_docs),
        question=RunnableLambda(lambda x: x["question"])
    )
    | prompt_template
    | ChatOpenAI(model="gpt-4o", temperature=0).with_retry(stop_after_attempt=3)
    | StrOutputParser()
).with_fallbacks([
    simple_llm_chain  # Fallback to direct LLM if retrieval fails
])

2. Streaming with Callbacks

from langchain_core.callbacks import BaseCallbackHandler

class ProductionCallback(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        logger.info(f"LLM call started: {len(prompts)} prompts")
        metrics.increment("llm_calls_total")

    def on_llm_error(self, error, **kwargs):
        logger.error(f"LLM error: {error}")
        metrics.increment("llm_errors_total")
        alerting.notify(f"LLM error in production: {error}")

# Stream with instrumentation
async for chunk in chain.astream(input, config={"callbacks": [ProductionCallback()]}):
    yield chunk

The gap between LangChain tutorials and production LangChain is the gap between a prototype and a product — bridged by error handling, observability, and defensive design.