Deploying LangChain apps and moving to LangGraph

Dependency pinning, streaming from an API, production error handling, and migrating a multi-step agent workflow to LangGraph state machines.

Packaging and pinning

# LangChain moves quickly: pin exact versions and upgrade deliberately
python -m pip freeze | grep -E "langchain|langgraph|pydantic" > requirements.txt

# a minimal pinned set for a RAG service
cat > requirements.txt <<'EOF'
langchain-core==0.3.29
langchain==0.3.14
langchain-openai==0.2.14
langchain-community==0.3.14
langchain-text-splitters==0.3.5
langgraph==0.2.62
pydantic==2.10.4
fastapi==0.115.6
uvicorn==0.34.0
EOF

python -m pip install -r requirements.txt --no-deps

# fail fast at startup if a required environment variable is missing
import os
from dataclasses import dataclass

@dataclass(frozen=True)
class Settings:
    openai_api_key: str
    langchain_project: str
    vector_store_path: str
    max_concurrency: int = 4

    @classmethod
    def load(cls):
        missing = [k for k in ("OPENAI_API_KEY", "LANGCHAIN_PROJECT", "VECTOR_STORE_PATH")
                   if not os.environ.get(k)]
        if missing:
            raise SystemExit("missing environment variables: " + ", ".join(missing))
        return cls(
            openai_api_key=os.environ["OPENAI_API_KEY"],
            langchain_project=os.environ["LANGCHAIN_PROJECT"],
            vector_store_path=os.environ["VECTOR_STORE_PATH"],
            max_concurrency=int(os.environ.get("MAX_CONCURRENCY", "4")),
        )

settings = Settings.load()

• Pin exact versions. LangChain's packages release frequently and a minor bump can change a default that silently alters behaviour.
• Load and validate configuration at startup. A missing API key should stop the container, not surface as a 500 on the first user request.
• Build long-lived objects — the vector store, the model client, the compiled chain — once at startup. Constructing them per request repeats expensive work.
• Keep the chain definition in one module and import it from both the API and the evaluation scripts, so you are always testing the artefact you deploy.

Serving and error handling

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import asyncio

app = FastAPI()

class Query(BaseModel):
    question: str
    session_id: str

# the compiled chain is created once at import time
from app.chain import rag_chain

@app.on_event("startup")
async def warm_up():
    await asyncio.to_thread(rag_chain.invoke, {"question": "warm up"})

@app.post("/ask")
async def ask(query: Query):
    try:
        result = await rag_chain.ainvoke({"question": query.question})
        return {"answer": result["answer"], "sources": result.get("sources", [])}
    except TimeoutError:
        raise HTTPException(status_code=504, detail="the model timed out, please retry")
    except Exception as exc:
        # log the full error, return an opaque one
        print("chain failure:", type(exc).__name__, str(exc)[:200])
        raise HTTPException(status_code=500, detail="the request could not be completed")

@app.post("/stream")
async def stream(query: Query):
    async def generator():
        async for chunk in rag_chain.astream({"question": query.question}):
            if isinstance(chunk, str):
                yield chunk
    return StreamingResponse(generator(), media_type="text/plain")

# always bound the work: an unbounded model call is an unbounded request
async def with_timeout(coro, seconds=30):
    return await asyncio.wait_for(coro, timeout=seconds)

• Put a timeout on every model, retriever and tool call. Without one, a slow provider turns into a hung request and then into a queue of hung requests.
• Return a stable error contract: a status code your client understands and a message that does not leak provider detail or internal paths.
• Streaming endpoints should still set a total timeout. A stream that never ends is as bad as a request that never returns.
• Warm up on startup. The first request otherwise pays for client construction, connection setup and sometimes model weight loading.

Moving to LangGraph

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    messages: Annotated[list, add_messages]     # reducer: append instead of replace
    context: list
    attempts: int

def retrieve(state: State):
    docs = retriever.invoke(state["messages"][-1].content)
    return {"context": [d.page_content for d in docs]}

def generate(state: State):
    answer = model.invoke(
        f"Context:\n{chr(10).join(state['context'])}\n\n"
        f"Question: {state['messages'][-1].content}")
    return {"messages": [answer], "attempts": state.get("attempts", 0) + 1}

def needs_retry(state: State) -> str:
    """A conditional edge: return the name of the next node."""
    last = state["messages"][-1].content.lower()
    if not state["context"] and state.get("attempts", 0) < 2:
        return "retrieve"
    return END

graph = StateGraph(State)
graph.add_node("retrieve", retrieve)
graph.add_node("generate", generate)
graph.add_edge(START, "retrieve")
graph.add_edge("retrieve", "generate")
graph.add_conditional_edges("generate", needs_retry, {"retrieve": "retrieve", END: END})

app_graph = graph.compile(checkpointer=MemorySaver())

config = {"configurable": {"thread_id": "t-1"}}
for event in app_graph.stream({"messages": [("user", "Where is order A-1042?")]},
                              config=config):
    print(event)

FAQ

Related

Runnables and LangChain Expression Language in depth Evaluating chains and RAG

Last refreshed 2026-09-18.