Agents, graphs, glue.

• LangChain — a library of building blocks for LLM apps: prompts, model wrappers, output parsers, retrievers, document loaders, and a composition language (LCEL) for chaining them together.
• LangGraph — a runtime for stateful agents modeled as graphs of nodes and edges, with explicit state, persistence, human-in-the-loop, and time travel.

The 2026 reality: many teams are moving back toward direct SDK calls (Anthropic, OpenAI) for simple cases. The Anthropic SDK now supports tool use natively, structured outputs, streaming, prompt caching. For a simple "call the model with a prompt and parse the response" workflow, you don't need LangChain.

Where LangChain still wins: when you're composing things — multiple model providers, multiple retrievers, multiple output formats — and want a uniform abstraction. Where LangGraph wins: when you have actual graph-shaped logic with state, branching, and recovery.

LangGraph builds on LangChain primitives. Most LangGraph nodes wrap LangChain components (prompts, models, retrievers). Understanding the chain layer is prerequisite to using the graph layer well — even though many teams skip directly to LangGraph for new agent projects.

The core types you'll use constantly:

The big idea: everything implements the same five methods so they can be composed.

class Runnable:
    def invoke(input)            # synchronous, single input
    def batch(inputs)            # synchronous, list of inputs
    def stream(input)            # synchronous, yields chunks
    async def ainvoke(input)     # async versions
    async def abatch(inputs)
    async def astream(input)

Models, prompts, parsers, retrievers — all Runnables. So you can pipe them together.

from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("human", "{question}"),
])

model = ChatAnthropic(model="claude-opus-4-7")

chain = prompt | model | StrOutputParser()

result = chain.invoke({"question": "What is the capital of France?"})
# "The capital of France is Paris."

The | operator chains Runnables. Each step's output becomes the next step's input.

LangChain has hundreds of integrations. The ones you'll actually use:

• Models: Anthropic, OpenAI, Google, AWS Bedrock, Azure OpenAI, local Ollama.
• Vector stores: pgvector, Chroma, Pinecone, Weaviate, Qdrant, FAISS, Milvus.
• Document loaders: PDF, DOCX, web pages, GitHub, Notion, Confluence, S3, etc.
• Embeddings: OpenAI, Voyage, Cohere, HuggingFace, local.
• Splitters: recursive character, markdown-aware, code-aware, semantic.
• Tools: web search (Tavily, SerpAPI), Python REPL, SQL, Slack, GitHub.
• Output parsers: Pydantic, JSON, XML, structured data with Zod-like schemas.

The integrations are the main reason to use LangChain. Building 30+ document loaders yourself is a year of work; LangChain has them.

In 2024, LangChain split into multiple packages to stop "install LangChain, get every dependency on earth":

pip install langchain-core         # base abstractions
pip install langchain              # high-level wrappers (chains, agents)
pip install langchain-community    # community integrations
pip install langchain-anthropic    # Anthropic provider
pip install langchain-openai       # OpenAI provider
pip install langgraph              # graph runtime
pip install langsmith              # observability client

For a typical app, you install langchain-core + the provider packages you actually use + optionally langgraph. You don't need the kitchen-sink langchain package for new projects.

# minimal RAG app
pip install langchain-core langchain-anthropic langchain-postgres
pip install langchain-text-splitters

# DON'T do this anymore
pip install langchain  # pulls in too much

Real critiques to weigh:

• Abstraction over-reach. Wrappers around wrappers. For simple use cases, the SDK is clearer.
• API churn. Major refactors in 2023 and 2024. Code from 2 years ago doesn't run.
• Documentation lag. Tutorials often show old patterns; current best practice is in deeply-nested doc pages.
• Debugging difficulty. When a chain fails, the stack trace can be unhelpful — you're inside several layers of LangChain machinery.
• Performance overhead. Not free; for high-throughput services, the overhead matters.

Counter-argument: for prototyping and for non-trivial RAG pipelines, the productivity gain is real. The right call depends on your situation, not on a blanket take.

Chains are linear: step 1 → step 2 → step 3 → done. Real agents loop, branch, retry, delegate. A graph captures this naturally:

The agent node calls the LLM. If the LLM returned tool calls, edge goes to the tools node. After tools execute, edge goes back to the agent. If the LLM returned a final answer, edge goes to END.

1. State — a typed dict that flows through the graph. Each node receives it, returns updates.
2. Nodes — Python functions that read state and return updates to it.
3. Edges — connections between nodes. Can be conditional (branch on state).
4. Checkpointer — persistence layer. Saves state after each node so the graph can pause, resume, time-travel, or resume after a crash.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langchain_anthropic import ChatAnthropic
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    messages: Annotated[list, "append"]  # new messages append to existing

def call_model(state: State):
    model = ChatAnthropic(model="claude-opus-4-7")
    response = model.invoke(state["messages"])
    return {"messages": [response]}

graph = StateGraph(State)
graph.add_node("model", call_model)
graph.add_edge(START, "model")
graph.add_edge("model", END)

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

# run with a thread_id so state persists across invocations
config = {"configurable": {"thread_id": "user-1"}}

result1 = app.invoke({"messages": [HumanMessage("My name is Alex")]}, config)
result2 = app.invoke({"messages": [HumanMessage("What's my name?")]}, config)
# Claude remembers because state persisted via the checkpointer

LangGraph thinks of conversations as threads. Each thread has its own state. The checkpointer stores state per thread. To resume a conversation, pass the same thread_id.

This is the conversation memory model that just works: no manual session management, no manual message-history wrangling. The graph handles it.

def should_continue(state: State) -> str:
    last = state["messages"][-1]
    if last.tool_calls:
        return "tools"  # has tool calls, run them
    return END         # no tool calls, we're done

graph.add_node("agent", call_model)
graph.add_node("tools", run_tools)

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {
    "tools": "tools",
    END: END,
})
graph.add_edge("tools", "agent")  # back to agent after tools run

This is the classic ReAct loop — call the model, run any tool calls, return to the model, repeat until done. With conditional edges, you express it once and the runtime handles the looping.

Most agent bugs are state bugs:

• Tool result not getting back to the model.
• Conversation history corrupted by retries.
• Two paths in the graph both writing to the same field, races.
• Restarting after a crash leaves the agent confused about where it was.

LangGraph forces you to declare state up front. Every node says "I take this state, I return these updates." The runtime applies updates correctly (with reducers like append for lists).

Because state is checkpointed after every node, you get capabilities most agent frameworks don't:

• Pause and resume — start an agent, walk away, resume tomorrow.
• Survive crashes — server restarts, agent picks up where it left off.
• Time travel — list past checkpoints, fork from any of them, try a different path.
• Human-in-the-loop — pause before critical steps for human approval.

# list past states
for state in app.get_state_history(config):
    print(state.config, state.values["messages"][-1].content)

# fork from a previous state with edits
forked_config = app.update_state(
    earlier_state.config,
    {"messages": [HumanMessage("...different question...")]},
)
result = app.invoke(None, forked_config)  # resume from forked state

messages = [HumanMessage("...")]
while True:
    resp = model.invoke(messages)
    messages.append(resp)
    if not resp.tool_calls:
        break
    for tc in resp.tool_calls:
        result = run_tool(tc.name, tc.args)
        messages.append(ToolMessage(content=result, tool_call_id=tc.id))

• It's not a model wrapper — you bring your own (LangChain models work, but raw SDK calls work too).
• It's not a prompt library — you write your own prompts.
• It's not magic — agents are still hard. LangGraph just handles the orchestration.

Task: retrieve documents, ask the LLM to answer, return the response.

You can — and often should — use both. LangChain primitives inside LangGraph nodes:

# inside a LangGraph node, use a LangChain chain
def rag_node(state: State):
    rag_chain = retriever | prompt | model | StrOutputParser()
    answer = rag_chain.invoke(state["question"])
    return {"messages": [AIMessage(answer)]}

This is the typical production pattern: LangGraph for the high-level flow, LangChain for the chain-shaped pieces inside each node.

If you have a LangChain AgentExecutor codebase, the official path is to migrate to LangGraph. There's a create_react_agent helper in langgraph.prebuilt that gives you the equivalent ReAct loop with all the LangGraph benefits — persistence, debugging, time travel.

from langgraph.prebuilt import create_react_agent

agent = create_react_agent(
    model=ChatAnthropic(model="claude-opus-4-7"),
    tools=[web_search, calculator],
    checkpointer=MemorySaver(),
)

result = agent.invoke(
    {"messages": [HumanMessage("Research X")]},
    config={"configurable": {"thread_id": "user-1"}},
)

One function. ReAct agent. Persistence built in. This is the 2026 entry point for "I want an agent" — not AgentExecutor.

chain = prompt | model | parser

This creates a single Runnable. When invoked, it pipes the input through each stage. Output of stage N is input to stage N+1.

The chain itself is a Runnable, so it can compose with other chains:

full = preprocess | (chain_a | chain_b) | postprocess

Because everything is a Runnable, you get for free:

• Streaming — call chain.stream(...), get incremental output.
• Async — call chain.ainvoke(...) for non-blocking.
• Batching — chain.batch([input1, input2, ...]) runs in parallel.
• Tracing — every stage logs to LangSmith automatically.
• Retry — chain.with_retry() wraps the whole thing.
• Fallbacks — chain.with_fallbacks([backup_chain]).

for chunk in chain.stream({"question": "Explain transformers"}):
    print(chunk, end="", flush=True)

Each chunk arrives as the model produces it. Streaming works through the whole chain — output parsers can stream too if they support it.

Sometimes you want to swap parts of a chain at runtime — different model, different temperature, different prompt.

from langchain_core.runnables import ConfigurableField

model = ChatAnthropic(
    model="claude-opus-4-7",
    temperature=0.7,
).configurable_fields(
    temperature=ConfigurableField(id="temperature"),
)

chain = prompt | model | parser

# normal call uses default
chain.invoke({"q": "..."})

# override at invocation
chain.invoke(
    {"q": "..."},
    config={"configurable": {"temperature": 0.0}},
)

Useful for deterministic test runs, different envs, or A/B testing prompts.

from operator import itemgetter
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_anthropic import ChatAnthropic

retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

prompt = ChatPromptTemplate.from_messages([
    ("system", "Answer using only the context. If unsure, say so."),
    ("human", "Context:\n{context}\n\nQuestion: {question}"),
])

def format_docs(docs):
    return "\n\n".join(d.page_content for d in docs)

model = ChatAnthropic(model="claude-opus-4-7", temperature=0)

chain = (
    RunnableParallel({
        "context": itemgetter("question") | retriever | format_docs,
        "question": itemgetter("question"),
        "sources": itemgetter("question") | retriever,
    })
    | RunnableParallel({
        "answer": prompt | model | StrOutputParser(),
        "sources": itemgetter("sources"),
    })
)

result = chain.invoke({"question": "What's our PTO policy?"})
# {"answer": "...", "sources": [Document(...), ...]}

When chains fail, the stack trace can be hard to read. Tools that help:

• LangSmith — every chain run shows up as a trace with each stage's input/output. The single biggest help for debugging.
• chain.get_graph().print_ascii() — prints the structure of a chain.
• chain.invoke(input, config={"callbacks": [StdOutCallbackHandler()]}) — logs each step.
• chain.with_config(tags=["debug"]) — adds tags visible in LangSmith.

LCEL is great for the shape "input → transform → output." It's worse for:

• Conditional logic that depends on intermediate results in non-trivial ways.
• Loops (use LangGraph).
• Code that's easier to read as plain Python.

Forced LCEL — using the operator just to use it — is a common antipattern. If RunnableLambda(lambda x: complex_function(x)) is what you'd write, just call the function. The | operator is for composition; it's not magic.

Strip away the hype: an agent is a loop where each iteration:

1. Sends conversation state (including past tool results) to the LLM.
2. LLM responds with either a final answer or one or more tool calls.
3. If tool calls, run them and append results to state.
4. If final answer, exit the loop.

That's it. The intelligence is in the LLM and the tools. The framework just orchestrates the loop.

ReAct = Reason + Act. The model alternates between thinking and taking actions, with each action's result informing the next thought.

User: What's the weather in Paris and should I bring an umbrella?

Agent thought: I need to check the weather forecast for Paris.
Agent action: get_weather("Paris")
Tool result: {"temp": 14, "conditions": "rain expected", "humidity": 80}

Agent thought: Rain is expected. The user should bring an umbrella.
Agent answer: It's 14°C in Paris with rain expected — yes, bring an umbrella.

Modern models do this implicitly via tool calling — they don't need explicit "thought" prompting in the way the original 2022 ReAct paper described.

from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return weather_api.get(city)

@tool
def search_web(query: str) -> str:
    """Search the web for recent information."""
    return tavily.search(query)

model = ChatAnthropic(model="claude-opus-4-7")

agent = create_react_agent(
    model=model,
    tools=[get_weather, search_web],
    checkpointer=MemorySaver(),
)

result = agent.invoke(
    {"messages": [HumanMessage("Should I bring an umbrella to Paris tomorrow?")]},
    config={"configurable": {"thread_id": "user-1"}},
)

create_react_agent is the prebuilt graph. It handles the loop, tool calling, and message flow. For 80% of use cases, this is what you want.

The prebuilt agent is great for ReAct-style loops. You'll need a custom graph when:

• Multiple specialized agents collaborate (multi-agent).
• Specific decision logic between tool calls (e.g., "always validate the result before continuing").
• Human approval gates at certain steps.
• Retry strategies that aren't just "let the LLM try again."
• State that includes more than just messages (workflow progress, tracked entities).

What should be in your agent's state? Common fields:

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]  # conversation
    user_query: str                       # original user question
    plan: list[str]                       # plan for multi-step tasks
    completed_steps: Annotated[list, add] # what's been done
    artifacts: dict                       # documents, data produced
    feedback: list[str]                   # human or self-critique notes

The reducers (add_messages, add) tell LangGraph how to merge updates from concurrent nodes. Without them, two nodes writing to messages would overwrite each other.

Where does the agent's system prompt live?

• In create_react_agent: pass state_modifier to inject system instructions on every model call.
• In a custom graph: add a system message at the start of the messages list, or include it in the prompt template inside the agent node.

agent = create_react_agent(
    model=model,
    tools=tools,
    state_modifier="""You are a helpful research assistant.
    Always cite sources when using web_search results.
    If a tool returns an error, try once more with a refined input."""
)

The model decides which tool to call based on the tool's description. Bad descriptions = bad tool selection.

# bad
@tool
def search(q: str):
    """Search."""
    return ...

# good
@tool
def search_web(query: str) -> str:
    """Search the web for current information about a topic.
    Use this for: current events, recent news, real-time data,
    facts that may have changed since training.
    Don't use this for: math, code generation, or general reasoning
    that doesn't need fresh information."""
    return ...

Treat tool descriptions like documentation for a junior developer. Be explicit about when to use and not use each tool.

class State(TypedDict):
    messages: Annotated[list, add_messages]
    next: str  # which agent to invoke next

def supervisor(state):
    response = supervisor_llm.invoke(state["messages"])
    # supervisor returns structured output with "next" field
    return {"next": response["next"]}

def researcher(state):
    response = research_agent.invoke(state["messages"])
    return {"messages": [AIMessage(response, name="researcher")]}

def writer(state):
    response = writer_agent.invoke(state["messages"])
    return {"messages": [AIMessage(response, name="writer")]}

def fact_checker(state):
    response = fact_check_agent.invoke(state["messages"])
    return {"messages": [AIMessage(response, name="fact_checker")]}

graph = StateGraph(State)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
graph.add_node("fact_checker", fact_checker)

graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", lambda s: s["next"], {
    "researcher": "researcher",
    "writer": "writer",
    "fact_checker": "fact_checker",
    "FINISH": END,
})
graph.add_edge("researcher", "supervisor")
graph.add_edge("writer", "supervisor")
graph.add_edge("fact_checker", "supervisor")

• Infinite loops. Agent keeps calling the same tool forever. Fix: max iterations, or have the agent reflect on whether it's making progress.
• Tool hallucinations. LLM invents tool arguments that don't match the schema. Fix: strict schema validation; some models handle this better than others.
• Hallucinated tool results. Model confidently uses results it didn't actually receive. Fix: include tool results explicitly in messages with proper formatting.
• Premature finish. Agent decides it's done before actually completing the task. Fix: better system prompts; few-shot examples of when to keep going.
• Not finishing. Agent keeps refining when "good enough" was 3 iterations ago. Fix: cap iterations; have agent explicitly evaluate its own output.

from langchain_core.tools import tool

@tool
def lookup_order(order_id: str) -> dict:
    """Look up an order by its ID. Returns order details including
    status, items, and shipping info. Use this when the user asks
    about a specific order."""
    return order_db.get(order_id)

The decorator turns the function into a Tool. The docstring becomes the tool description the LLM sees. Type hints become the input schema.

For complex inputs, use Pydantic:

from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="Search query in natural language")
    max_results: int = Field(default=5, description="Number of results, 1-20")
    recency_days: int = Field(default=30, description="Only results from last N days")

@tool(args_schema=SearchInput)
def search_news(query: str, max_results: int = 5, recency_days: int = 30) -> list[dict]:
    """Search news articles."""
    return news_api.search(query, limit=max_results, recent=recency_days)

The model sees the schema. It learns it can adjust max_results and recency_days when appropriate.

# LangChain native
model_with_tools = ChatAnthropic(...).bind_tools([search_news, lookup_order])

response = model_with_tools.invoke([HumanMessage("Find me recent news about climate policy")])
# response.tool_calls = [{"name": "search_news", "args": {"query": "...", ...}, "id": "..."}]

The model returns tool calls; you (or the framework) execute them.

Tools fail. APIs go down, inputs are invalid, network glitches. The agent needs to recover.

@tool
def fetch_url(url: str) -> str:
    """Fetch the contents of a URL."""
    try:
        response = httpx.get(url, timeout=10)
        response.raise_for_status()
        return response.text[:5000]
    except httpx.TimeoutException:
        return "ERROR: request timed out after 10 seconds"
    except httpx.HTTPStatusError as e:
        return f"ERROR: HTTP {e.response.status_code}"
    except Exception as e:
        return f"ERROR: {type(e).__name__}: {e}"

The pattern: return errors as strings, not raise exceptions. The LLM can read the error and decide what to do (retry with different args, give up gracefully, ask the user).

If you raise, LangGraph's ToolNode by default catches and converts to a tool message anyway, but explicit string errors give better LLM behavior.

• Few well-described tools beat many vaguely-described ones. The model gets confused with 30 tools.
• Tools should be high-leverage. One tool that returns rich results beats five that each return fragments.
• Make signatures explicit. get_user(user_id) beats query(thing, value).
• Return structured data, not freeform text when possible. Easier for the model to use precisely.
• Idempotent when possible. If the same call happens twice, no double-effects.
• Bound the output size. Tool returning 50KB of JSON wastes tokens. Truncate or summarize.

LangChain ships many pre-built tools. Common ones:

• Tavily / SerpAPI / Bing — web search.
• WikipediaQueryRun — wikipedia.
• PythonREPLTool — execute Python (sandboxed for trusted code only).
• SQLDatabaseToolkit — query SQL databases.
• Slack/GitHub/Jira/Notion integrations.
• requests_tools — HTTP requests with restrictions.

For most production tools, you'll write your own — wrappers around your internal APIs.

Some tools shouldn't run without human approval. LangGraph supports this with interrupts:

from langgraph.types import interrupt

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email."""
    # In a graph node, before sending:
    approved = interrupt({
        "action": "send_email",
        "to": to,
        "subject": subject,
        "body": body,
    })
    if not approved:
        return "User declined to send email."
    return email_service.send(to, subject, body)

The interrupt pauses graph execution. Your application surfaces the proposed action to a human, who approves or rejects. The graph resumes with the human's decision.

In LangGraph, the conversation is just the messages field of state. The checkpointer persists state per thread_id. Same thread → same history.

config = {"configurable": {"thread_id": "user-42"}}

# turn 1
agent.invoke(
    {"messages": [HumanMessage("My name is Alex")]},
    config,
)

# turn 2 — checkpointer loaded the previous state
agent.invoke(
    {"messages": [HumanMessage("What's my name?")]},
    config,
)
# Claude responds "Your name is Alex" — because thread state has the previous turn

This is the entire mechanism. No separate memory abstraction needed. The state IS the memory.

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string(DB_URL) as checkpointer:
    checkpointer.setup()  # creates tables on first run
    agent = graph.compile(checkpointer=checkpointer)

For production, Postgres is the default choice. It's durable, queryable, backs up like any DB, and integrates with the rest of your data.

Threads can grow unboundedly. After 50 turns, your context is too big and inference is slow. Options:

Things you want to remember about a user across all their conversations:

• Their preferences ("prefers concise responses").
• Facts about them ("vegetarian", "lives in NYC").
• Past topics ("we discussed their migraine triggers last month").

This is NOT what LangGraph state handles. Thread state is per-thread, not per-user-across-threads.

LangGraph added a separate BaseStore for cross-thread, cross-session memory:

from langgraph.store.memory import InMemoryStore
from langgraph.store.postgres import PostgresStore

store = PostgresStore.from_conn_string(DB_URL)

# in a node
def update_facts(state, *, store):
    user_id = state["user_id"]
    new_fact = extract_fact(state["messages"][-1])
    store.put(
        ("user", user_id, "facts"),  # namespace
        f"fact-{uuid4()}",            # key
        {"fact": new_fact},           # value
    )
    return {}

def recall_facts(state, *, store):
    user_id = state["user_id"]
    facts = store.search(("user", user_id, "facts"))
    return {"context": [f.value["fact"] for f in facts]}

The store is namespace-keyed and supports semantic search (when configured with embeddings). It's the right place for "remember this about this user forever."

Most agents use a combination: thread state for the current conversation, store for user facts, vector store for "things this user said in past conversations."

Layer 1 — current thread (LangGraph state, Postgres checkpointer)
  - Last ~20 messages of the active conversation
  - Working memory for the current task

Layer 2 — recent history (last 30 days, vector store)
  - Embeddings of past conversations
  - Retrieved by similarity to current query
  - Surfaced in context when topic recurs

Layer 3 — long-term facts (LangGraph Store)
  - Structured facts: "Alex is vegetarian"
  - Preferences: "prefers metric units"
  - Updated by an "extract facts" node after each turn
  - Loaded on session start

Layer 4 — knowledge base (separate vector store)
  - Documents the assistant has access to
  - Retrieved per-query as in regular RAG

Memory creates privacy obligations. Plan for:

• User data export. GDPR right to access. Be able to dump all stored facts and conversations for a user.
• User data deletion. "Right to be forgotten." Delete checkpoints, store entries, and any vector embeddings tied to the user.
• Sensitive content handling. If a user shares health info, financial info, etc., have policies around storage and access.
• Encryption at rest. The store IS user data — protect accordingly.

1. Load documents from sources (PDFs, web, DBs, APIs).
2. Split into chunks (semantic, fixed-size, recursive).
3. Embed each chunk with an embedding model.
4. Store in a vector database with metadata.
5. Query — embed user query, find similar chunks.
6. Augment — insert retrieved chunks into prompt.
7. Generate — LLM answers using the chunks as context.

from langchain_community.document_loaders import (
    PyPDFLoader,
    WebBaseLoader,
    GitHubIssuesLoader,
    NotionDBLoader,
    S3DirectoryLoader,
)

docs = PyPDFLoader("handbook.pdf").load()
# returns list[Document]; Document has page_content + metadata

Hundreds of loaders in langchain_community. The metadata they attach (page numbers, source URLs, headings) is critical for citations.

Chunks need to be small enough to fit in context but large enough to be self-contained. Common strategies:

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=150,
    separators=["\n\n", "\n", ". ", " ", ""],
)
chunks = splitter.split_documents(docs)

Overlap (~10-20% of chunk size) helps when relevant info spans a chunk boundary.

from langchain_openai import OpenAIEmbeddings
from langchain_voyageai import VoyageAIEmbeddings

embeddings = VoyageAIEmbeddings(model="voyage-3")

Where embeddings live and get searched.

2026 default for most teams: pgvector. You probably already have Postgres. Adding the extension is one line. Vectors live alongside other data, queryable with SQL filters.

from langchain_postgres import PGVector

vectorstore = PGVector(
    embeddings=embeddings,
    collection_name="docs",
    connection=DB_URL,
)
vectorstore.add_documents(chunks)

retriever = vectorstore.as_retriever(
    search_kwargs={"k": 5, "filter": {"team": "engineering"}},
)

RAG quality is multi-dimensional. Eval each piece:

• Retrieval quality — does the retriever return relevant chunks? Use a labeled dataset; measure recall@k.
• Answer faithfulness — does the answer stay grounded in the context, no hallucinations?
• Answer relevance — does the answer actually address the question?
• Context precision — what fraction of retrieved chunks are actually relevant?

Tools: Ragas for RAG-specific metrics, LangSmith for end-to-end evals with custom evaluators.

chain = prompt | model | StrOutputParser()

for chunk in chain.stream({"question": "Explain transformers"}):
    print(chunk, end="", flush=True)

Each chunk is a piece of the final string. Output parsers stream too if they support it (StrOutputParser does; PydanticParser usually doesn't, since you need the full output to validate).

Two kinds of streaming:

• chain.stream(input) — yields output values progressively (token chunks for a string output).
• chain.astream_events(input, version="v2") — yields lifecycle events: start, chunk, end, retrieval results, etc. Fine-grained.

async for event in chain.astream_events({"question": "..."}, version="v2"):
    kind = event["event"]
    if kind == "on_chat_model_stream":
        chunk = event["data"]["chunk"]
        yield chunk.content  # forward token to client
    elif kind == "on_retriever_end":
        docs = event["data"]["output"]
        yield {"sources": [d.metadata for d in docs]}

Use events when you want to surface intermediate state to the user — "looking up sources..." → "found 5 documents" → token-by-token answer.

LangGraph has multiple stream modes that show different views of execution:

# stream node updates as they happen
async for chunk in agent.astream(input, config, stream_mode="updates"):
    print(chunk)
# {"agent": {"messages": [AIMessage(...)]}}
# {"tools": {"messages": [ToolMessage(...)]}}
# {"agent": {"messages": [AIMessage(final answer)]}}

# stream LLM tokens directly
async for token, metadata in agent.astream(input, config, stream_mode="messages"):
    if metadata["langgraph_node"] == "agent":
        print(token.content, end="", flush=True)

# stream multiple modes at once
async for mode, chunk in agent.astream(input, config, stream_mode=["updates", "messages"]):
    if mode == "messages":
        token, meta = chunk
        ...
    elif mode == "updates":
        ...

Users want feedback. The full streaming UX has multiple layers:

1. "Working..." — show immediately when the request starts.
2. Stage indicators — "Searching documents", "Reading sources", "Drafting response".
3. Token streaming — show the answer as it generates.
4. Sources/citations — appear when retrieval completes, before answer.
5. Tool invocations — visible in agent UIs ("calling search_news...").

LangGraph's multi-mode streaming makes this achievable. Subscribe to updates for stage transitions, messages for tokens, surface both to the UI.

Every Runnable has an ainvoke / abatch / astream async variant. Use these in async web frameworks (FastAPI, etc.) to avoid blocking the event loop.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.post("/chat")
async def chat(request: ChatRequest):
    async def generate():
        async for chunk in chain.astream({"question": request.question}):
            yield chunk
    return StreamingResponse(generate(), media_type="text/event-stream")

One thing worth knowing: when an agent decides to call a tool, you usually don't want to stream the model's "I will use the X tool" reasoning to the user. Stream only the final answer.

async for token, metadata in agent.astream(
    input, config, stream_mode="messages"
):
    # only stream tokens from the agent node when the message
    # has no tool calls (i.e., it's the final answer)
    if metadata["langgraph_node"] == "agent":
        if not getattr(token, "tool_calls", None):
            yield token.content

Otherwise you'd surface intermediate "thinking" that may include tool call JSON or partial reasoning the user shouldn't see.

from pydantic import BaseModel, Field
from langchain_anthropic import ChatAnthropic

class Sentiment(BaseModel):
    sentiment: Literal["positive", "negative", "neutral"]
    confidence: float = Field(ge=0, le=1)
    keywords: list[str] = Field(description="Key emotional words detected")

model = ChatAnthropic(model="claude-opus-4-7")
structured = model.with_structured_output(Sentiment)

result = structured.invoke("This product is amazing, I love it!")
# Sentiment(sentiment='positive', confidence=0.95, keywords=['amazing', 'love'])

Under the hood, this uses the model's native tool-calling API to enforce the schema. The output is a real Pydantic instance — validated, typed, ready to use.

• The model knows the schema. Tool calling APIs include the schema in the request; the model is constrained to it.
• Validation is automatic. Pydantic raises if fields are missing or types wrong.
• No prompt engineering for format. No "respond ONLY in JSON" pleas in the prompt.
• It works. Modern models have ~99% schema compliance with this approach.

The model sees field descriptions. Use them to disambiguate.

class Order(BaseModel):
    order_id: str = Field(description="The order number, format: ORD-XXXXX")
    items: list[str] = Field(description="List of product SKUs in the order")
    total: float = Field(description="Total cost in USD before tax")
    customer_email: EmailStr = Field(description="Customer's email; must be valid format")
    notes: Optional[str] = Field(
        default=None,
        description="Special instructions if any; null if none"
    )

Without descriptions, the model has to guess what each field means. With descriptions, it produces correct output reliably.

You can stream tokens of a structured output:

for chunk in structured.stream("..."):
    print(chunk)
# Sentiment(sentiment='positive', confidence=None, keywords=[])
# Sentiment(sentiment='positive', confidence=0.9, keywords=[])
# Sentiment(sentiment='positive', confidence=0.95, keywords=['amazing'])
# Sentiment(sentiment='positive', confidence=0.95, keywords=['amazing', 'love'])

Each chunk is a partial-but-valid Pydantic instance. Useful for showing form fields filling in as the model generates.

Use with_structured_output with modern models. Falls back gracefully on the others.

When the output type depends on the input:

from typing import Union, Literal

class WeatherResponse(BaseModel):
    type: Literal["weather"] = "weather"
    location: str
    temperature: float
    conditions: str

class ErrorResponse(BaseModel):
    type: Literal["error"] = "error"
    message: str

class ClarifyResponse(BaseModel):
    type: Literal["clarify"] = "clarify"
    question: str

class Output(BaseModel):
    response: Union[WeatherResponse, ErrorResponse, ClarifyResponse] = Field(
        discriminator="type"
    )

structured = model.with_structured_output(Output)
result = structured.invoke("What's the weather?")
# Could be any of the three based on the input

The model picks the right variant; Pydantic validates accordingly.

class ActionItem(BaseModel):
    description: str = Field(description="What needs to be done")
    owner: Optional[str] = Field(description="Person responsible; null if not specified")
    due_date: Optional[str] = Field(description="Due date if mentioned, format YYYY-MM-DD")
    priority: Literal["low", "medium", "high"] = Field(default="medium")

class MeetingNotes(BaseModel):
    summary: str = Field(description="One-paragraph summary of the meeting")
    decisions: list[str] = Field(description="Decisions made during the meeting")
    action_items: list[ActionItem]
    follow_up_topics: list[str] = Field(description="Topics tabled for follow-up")

structured = model.with_structured_output(MeetingNotes)

raw_notes = """
Discussed the launch — decided to push to Nov 15.
Sarah will handle the marketing copy by next Friday.
Need to follow up with legal on the contract.
Bug triage going well; Mike has it under control.
"""

result = structured.invoke(f"Extract structured info from: {raw_notes}")

Before with_structured_output, you'd see:

• StrOutputParser — extracts the string content from a chat message.
• JsonOutputParser — parses JSON, returns dict.
• PydanticOutputParser — parses JSON into a Pydantic model.
• StructuredOutputParser — schema-defined output.
• XMLOutputParser — XML-formatted responses.

These still exist and work. For new code, prefer with_structured_output on the model itself.

from langchain_core.prompts import PromptTemplate, ChatPromptTemplate

# string template (legacy completion style)
prompt = PromptTemplate.from_template("Translate to French: {text}")

# chat template (modern, message-based)
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a translator. Translate the user's text to {language}."),
    ("human", "{text}"),
])

result = chat_prompt.format_messages(language="French", text="Hello")
# [SystemMessage("You are a translator. Translate the user's text to French."),
#  HumanMessage("Hello")]

Modern models all use chat APIs. ChatPromptTemplate is the right default.

For prompts that include conversation history:

from langchain_core.prompts import MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    MessagesPlaceholder("history"),  # injected list of past messages
    ("human", "{question}"),
])

result = prompt.format_messages(
    history=[
        HumanMessage("What's 2+2?"),
        AIMessage("4"),
    ],
    question="What about 3+3?",
)

from langchain_core.prompts import FewShotChatMessagePromptTemplate

examples = [
    {"input": "happy", "output": "glad"},
    {"input": "sad", "output": "unhappy"},
    {"input": "fast", "output": "quick"},
]

example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}"),
])

few_shot = FewShotChatMessagePromptTemplate(
    examples=examples,
    example_prompt=example_prompt,
)

final_prompt = ChatPromptTemplate.from_messages([
    ("system", "Find a synonym for the user's word."),
    few_shot,
    ("human", "{input}"),
])

For large example pools, select the most relevant ones at runtime via embedding similarity:

from langchain_core.example_selectors import SemanticSimilarityExampleSelector

example_selector = SemanticSimilarityExampleSelector.from_examples(
    examples,
    embeddings,
    vectorstore_cls=Chroma,
    k=3,  # show 3 most similar
)

few_shot = FewShotChatMessagePromptTemplate(
    example_selector=example_selector,
    example_prompt=example_prompt,
    input_variables=["input"],
)

Pre-fill some variables, leave others for runtime:

prompt = PromptTemplate.from_template("{role}: {input}")
admin_prompt = prompt.partial(role="admin")

result = admin_prompt.format(input="delete user 5")
# "admin: delete user 5"

from langchain import hub

# fetch a community prompt
prompt = hub.pull("hwchase17/react")

# push your own (with auth)
hub.push("yourorg/customer-support-agent", prompt)

# pull specific version
prompt = hub.pull("yourorg/customer-support-agent:1.2.3")

Useful for prompt versioning, sharing across projects, and decoupling prompt iteration from code deploys.

Where do production prompts live?

Most teams start inline, move to external files when prompts get long, then to LangSmith/Hub when they need versioning + A/B testing.

prompt = "You are a customer support agent. Answer the user's question: {q}"

SYSTEM = """You are a customer support agent for AcmeCorp.
Tone: friendly, concise. Always offer to escalate complex issues.
If you don't know, say so honestly.
Available products: {products}
Current promotions: {promotions}
"""
prompt = ChatPromptTemplate.from_messages([
    ("system", SYSTEM),
    MessagesPlaceholder("history"),
    ("human", "{question}"),
])

prompts/customer_support.md  # markdown file with the system prompt
+ Python loader that reads the file at startup
+ tests that verify the prompt produces expected output on a fixture set

prompt = hub.pull("acme/customer-support:current")
# product team can update the prompt without engineering deploy
# A/B testing managed via Hub
# LangSmith tracks performance per prompt version

• Role + task + context + format. Tell the model who it is, what to do, what's relevant, how to respond. The four-section structure handles most prompts.
• Examples beat instructions. One good example beats five paragraphs of "make sure to..."
• Constrain output explicitly. "Respond in 1-2 sentences" works better than hoping for brevity.
• Use XML tags for structure. Anthropic's models respond very well to <context>...</context> style structuring.
• Mention what NOT to do. Negative constraints help. "Don't include preamble" / "Don't ask follow-up questions."

• Traces — every chain/graph run, every component, with full inputs and outputs.
• Datasets — labeled examples for evaluation.
• Evaluators — score outputs (LLM-as-judge, custom code, classical metrics).
• Experiments — run a chain over a dataset, compare variants.
• Annotation — humans review and label runs.
• Prompt Hub — versioned prompts with linked traces.
• Feedback — capture user thumbs up/down on production runs.

# env vars
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=ls__your_key
LANGCHAIN_PROJECT=my-app
LANGCHAIN_ENDPOINT=https://api.smith.langchain.com

That's it. Every LangChain/LangGraph run is automatically traced. Open the LangSmith UI, see every invocation with inputs, outputs, timing, costs.

RAG chain run                                              2.4s, $0.012
├── retriever                                              250ms
│   ├── embed query                                         80ms
│   └── vector search                                       170ms     5 docs
├── format docs                                              5ms
├── prompt template                                          1ms
└── chat model                                            2.1s, $0.011
    └── streaming output                                              312 tokens

Click any node to see its input and output. For chat models, see the full prompt sent and the full response received.

from langsmith import Client

client = Client()

# create a dataset from existing production traces
dataset = client.create_dataset(name="rag-eval-set")
client.create_examples(
    inputs=[{"question": "..."}, {"question": "..."}],
    outputs=[{"answer": "..."}, {"answer": "..."}],
    dataset_id=dataset.id,
)

# define evaluators
from langsmith.evaluation import evaluate

def correctness(run, example):
    predicted = run.outputs["answer"]
    expected = example.outputs["answer"]
    # use an LLM judge or custom logic
    score = llm_judge(predicted, expected)
    return {"key": "correctness", "score": score}

def has_citations(run, example):
    return {
        "key": "has_citations",
        "score": 1 if "[1]" in run.outputs["answer"] else 0,
    }

# run the experiment
results = evaluate(
    chain.invoke,
    data="rag-eval-set",
    evaluators=[correctness, has_citations],
)

Outcome: a linked, comparable run of your chain over the dataset, with scores per example, viewable in the UI.

Built-in evaluators that use an LLM to score outputs:

from langsmith.evaluation import LangChainStringEvaluator

# correctness eval
correctness_evaluator = LangChainStringEvaluator(
    "labeled_criteria",
    config={
        "criteria": "correctness",
        "llm": ChatAnthropic(model="claude-opus-4-7"),
    },
)

# helpfulness eval
helpfulness_evaluator = LangChainStringEvaluator(
    "criteria",
    config={"criteria": "helpfulness"},
)

Capture user feedback alongside traces:

from langsmith import traceable
import langsmith

@traceable
def chat(question: str):
    answer = chain.invoke({"question": question})
    return {"answer": answer, "run_id": langsmith.get_current_run_tree().id}

# user clicks thumbs up
client.create_feedback(
    run_id=run_id,
    key="user_rating",
    score=1,
)

Now you can filter traces by user feedback. "Show me all runs where users rated negatively" → identify failure modes.

LangSmith is free for individual developers (small quotas). Production usage is paid — scales with traces.

Alternatives:

• Self-hosted LangSmith — Docker image, runs on your infra (enterprise feature).
• OpenTelemetry export — LangChain instruments OTel; export to Datadog, Honeycomb, Grafana Tempo.
• Helicone, Phoenix, Langfuse — competing platforms with similar features.

For most teams: LangSmith if you can afford it, OTel + Grafana if you can't or you're already invested in your existing observability stack.

Don't just trace; instrument well:

• Tag runs by user, tenant, environment — filter by these in the UI.
• Add metadata for relevant identifiers — order_id, project_id, etc.
• Capture user feedback — thumbs up/down, edit-after, regeneration count.
• Track key business metrics — for support agent: did the user resolve the issue without escalating?

from langsmith import traceable

@traceable(metadata={"team": "support"}, tags=["customer-facing"])
def handle_inquiry(user_id: str, question: str):
    with langsmith.trace(metadata={"user_id": user_id}):
        return chain.invoke({"question": question})

Three levels of granularity:

1. Component evals — does this retriever return relevant docs? Does this prompt produce valid JSON?
2. End-to-end evals — given a user question, does the full pipeline produce a good answer?
3. Production evals — sampling and scoring real production traffic.

Mature systems do all three. Component evals catch regressions in the building blocks. E2E evals measure user-experienced quality. Production evals catch issues that don't appear in test sets.

The eval is only as good as the data. A useful golden dataset:

• 50-500 examples (start small, grow).
• Real user questions, not synthetic ones (where possible).
• Diverse — easy, hard, ambiguous, adversarial.
• Labeled — expected answer and/or pass criteria.
• Versioned — keep the dataset stable so scores are comparable over time.

Sources:

• Sample production traces, manually label outputs.
• Have domain experts write canonical Q&A pairs.
• Use bug reports as failure cases (negative examples).

JUDGE_PROMPT = """
You are evaluating whether an answer is correct.

Question: {question}
Reference answer: {reference}
Predicted answer: {predicted}

Score 1 if the predicted answer conveys the same information as the reference.
Score 0 otherwise.

Provide just the score (0 or 1) and a one-sentence reason.
"""

class JudgeOutput(BaseModel):
    score: int = Field(ge=0, le=1)
    reason: str

judge = ChatAnthropic(model="claude-opus-4-7", temperature=0)
structured_judge = judge.with_structured_output(JudgeOutput)

def evaluate_answer(question, reference, predicted):
    return structured_judge.invoke(
        JUDGE_PROMPT.format(
            question=question, reference=reference, predicted=predicted
        )
    )

LLM judges have biases:

• Position bias — when comparing two answers, often prefers the first one shown.
• Length bias — sometimes prefers longer answers regardless of quality.
• Self-preference — judges using the same model that generated may be biased.
• Verbosity bias — confident wrong answers can score higher than uncertain right ones.

Mitigations:

• Validate judge against human ratings on a subset.
• Use a different model as judge than as generator.
• Randomize position when comparing two outputs.
• Include explicit grading criteria in the judge prompt.
• Use multiple judges and average scores for high-stakes evaluation.

Often more reliable than absolute scoring. "Is A better than B?" is easier than "Is A good?"

JUDGE_PROMPT = """
Compare two answers to the question. Pick the better one.

Question: {question}

Answer A: {answer_a}
Answer B: {answer_b}

Which is better? Reply 'A', 'B', or 'tie'.
"""

# evaluate prompt v1 vs prompt v2
for example in dataset:
    a = chain_v1.invoke(example.input)
    b = chain_v2.invoke(example.input)
    # randomize order to avoid position bias
    if random() < 0.5:
        a, b = b, a
        winner = "B" if judge.compare(question, a, b) == "A" else "A"
    else:
        winner = judge.compare(question, a, b)
    record(winner)

For a RAG system, evaluate each piece:

• Retriever — given a question and known-relevant docs, does the retriever return them? Measure recall@k.
• Reranker — does it move relevant docs to the top? Measure MRR or NDCG.
• Generator — given retrieved context, is the answer correct and grounded?

Tools: Ragas has built-in metrics (faithfulness, answer relevancy, context precision, context recall).

For an agent, evaluate trajectories:

• Tool selection — when a known tool is needed, does the agent call it?
• Tool argument quality — are the arguments well-formed?
• Recovery from errors — when a tool fails, does the agent handle it?
• Termination — does the agent stop at the right point, not too early or late?
• Final answer quality — same as RAG generator eval.

from langsmith.evaluation import evaluate

# 1. Component eval — retriever
def retrieval_eval(run, example):
    retrieved = run.outputs["sources"]
    expected = example.outputs["expected_sources"]
    recall = len(set(r.id for r in retrieved) & set(expected)) / len(expected)
    return {"key": "retrieval_recall", "score": recall}

# 2. End-to-end eval — answer correctness via LLM judge
def correctness_eval(run, example):
    predicted = run.outputs["answer"]
    expected = example.outputs["answer"]
    score = llm_judge(predicted, expected)  # returns 0 or 1
    return {"key": "correctness", "score": score}

# 3. Format check — code-based
def has_citations(run, example):
    answer = run.outputs["answer"]
    return {"key": "has_citations", "score": 1 if "[1]" in answer else 0}

# Run the eval
results = evaluate(
    target=lambda inputs: chain.invoke(inputs),
    data="my-eval-dataset",
    evaluators=[retrieval_eval, correctness_eval, has_citations],
    experiment_prefix="rag-v2",
)

# In CI:
# fail if any metric drops > 5% from baseline
# (LangSmith stores baseline; comparison is built-in)

• Tiny dataset. 5-10 examples. Random noise dominates the signal.
• Synthetic-only data. Real users ask weirder questions than you imagine. Sample production.
• Single metric. "Correctness" alone misses grounding, brevity, format compliance. Use multiple.
• Judge using same model as generator. Self-preference bias. Use a different model.
• Forgetting baseline. Reporting "85% correctness" without comparison is meaningless.
• No drift check. Models change underneath you. Re-run baseline periodically.
• Eval-only optimization. Optimizing for the eval set; production drifts from it.

Most LangGraph apps end up as one of:

• HTTP API — typical REST/streaming service in front of the agent.
• Worker / queue consumer — long-running tasks pulled from a queue.
• Scheduled job — agents that run on a cron.
• Webhook handler — triggered by external events.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.post("/agent")
async def run_agent(req: Request):
    config = {"configurable": {"thread_id": req.thread_id}}

    async def generate():
        async for chunk in agent.astream(
            {"messages": [HumanMessage(req.message)]},
            config=config,
            stream_mode="messages",
        ):
            token, metadata = chunk
            if metadata["langgraph_node"] == "agent" and not getattr(token, "tool_calls", None):
                yield f"data: {json.dumps({'token': token.content})}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

LangChain offers LangGraph Cloud — a managed runtime that handles deployment, persistence, scaling, and APIs for your graphs.

Pros:

• Persistence handled (managed Postgres).
• Auto-scaling.
• Built-in HTTP API for invocation, streaming, thread management.
• Integrated with LangSmith.
• Long-running task support out of the box.

Cons:

• Lock-in to LangChain Cloud.
• Pricing scales with usage.
• Less control over infra than rolling your own.

For teams that want to focus on the agent, not the deployment, it's reasonable. For teams with existing infrastructure, deploying as a regular service is simpler.

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

Standard container. Deploy to ECS Fargate, Cloud Run, Fly, Railway, wherever. The agent code is just Python.

Don't use MemorySaver in production — state is lost on restart. Use Postgres:

from langgraph.checkpoint.postgres import PostgresSaver

# at startup
checkpointer = PostgresSaver.from_conn_string(DATABASE_URL)
checkpointer.setup()  # idempotent; creates tables on first run

agent = graph.compile(checkpointer=checkpointer)

For multi-instance deployments, all instances connect to the same Postgres. Conversations work across replicas.

LLM APIs rate-limit. Your agent will hit them. Configure retries at the model level:

from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(
    model="claude-opus-4-7",
    max_retries=3,           # built-in retry on retryable errors
    timeout=60,              # per-call timeout
)

For more control, wrap with with_retry:

from langchain_core.runnables import Runnable

resilient_chain = (prompt | model | parser).with_retry(
    retry_if_exception_type=(httpx.HTTPStatusError,),
    wait_exponential_jitter=True,
    stop_after_attempt=5,
)

Agents can spiral. Without limits, a buggy loop can spend $1000 in a few minutes.

• Max iterations on agents. Hard cap at 10-20 loops. Most legitimate tasks finish in < 10.
• Max tokens per response — set on the model.
• Total token budget per invocation — track in state, exit when exceeded.
• Per-user / per-tenant rate limits at the API layer.
• Daily spend alerts — billing alarms in your LLM provider.

class State(TypedDict):
    messages: Annotated[list, add_messages]
    iterations: int

def should_continue(state):
    if state["iterations"] > 15:
        return END  # bail out
    if state["messages"][-1].tool_calls:
        return "tools"
    return END

def agent(state):
    response = model.invoke(state["messages"])
    return {"messages": [response], "iterations": state["iterations"] + 1}

Identical prompts produce identical outputs. Cache to save money:

from langchain.globals import set_llm_cache
from langchain_community.cache import RedisCache
import redis

set_llm_cache(RedisCache(redis_=redis.Redis.from_url(REDIS_URL)))

For Anthropic specifically: prompt caching at the API level is dramatically cheaper than implementing your own. Mark large stable prefixes (system prompt, retrieved docs) as cacheable; subsequent calls with the same prefix cost ~10% as much.

from langchain_anthropic import ChatAnthropic

# enable prompt caching for stable system prompt
model = ChatAnthropic(
    model="claude-opus-4-7",
    extra_headers={"anthropic-beta": "prompt-caching-2024-07-31"},
)

# in messages, mark cache breakpoints
messages = [
    SystemMessage(
        content=[
            {"type": "text", "text": LONG_SYSTEM_PROMPT,
             "cache_control": {"type": "ephemeral"}},
        ]
    ),
    HumanMessage("..."),
]

Most B2B agents serve multiple tenants. Considerations:

• Tenant ID in thread_id namespace — {"thread_id": f"{tenant}/{user}/{conversation}"}. Prevents cross-tenant leaks.
• Per-tenant configuration — different prompts, tools, models per customer. Use ConfigurableField.
• Per-tenant data isolation — RAG indexes scoped by tenant. Postgres row-level security if shared tables.
• Per-tenant cost tracking — tag every LLM call with tenant_id.
• Per-tenant rate limits — prevent one customer from starving others.

A growing 2026 perspective: for many use cases, you don't need a framework at all.

# Anthropic SDK directly — no LangChain
from anthropic import Anthropic

client = Anthropic()

def chat(messages: list, tools: list = None):
    response = client.messages.create(
        model="claude-opus-4-7",
        max_tokens=4096,
        system="You are a helpful assistant.",
        messages=messages,
        tools=tools,
    )
    return response

# tool calling loop — about 30 lines
def run_agent(user_message: str, tools: list):
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-opus-4-7",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        )
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "end_turn":
            return response.content

        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = TOOLS[block.name](**block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": str(result),
                    })
            messages.append({"role": "user", "content": tool_results})

That's a complete agent. ~30 lines. No abstractions, no version pinning, no "it broke when LangChain updated." For simple cases, this is the right answer.

LangChain/LangGraph earn their complexity when you need:

• Document loaders for many formats (PDF, web, SharePoint, etc.) — building these yourself is months of work.
• Retrieval orchestration (hybrid search, reranking, query rewriting).
• Multi-provider model abstraction (swap OpenAI / Anthropic / Bedrock).
• Persistence and time travel for agents (LangGraph specifically).
• Multi-agent coordination at scale.
• Prompt versioning + observability (with LangSmith).

If you don't need most of these, the SDK is fine. If you need many of them, building them yourself is harder than learning the framework.

Where the ecosystem is going:

• Frameworks getting thinner. LangChain split into smaller packages. The "everything in one import" era is over.
• Direct SDK use rising. Provider SDKs added structured outputs, tool use, prompt caching natively — closing the gap with frameworks.
• Agent frameworks consolidating. LangGraph, AutoGen, CrewAI converging on similar patterns (state, nodes, edges, tools).
• Observability becoming table stakes. LangSmith, Helicone, Phoenix, Langfuse — all converging.
• Eval becoming first-class. Frameworks integrating eval datasets, judges, regression checks into the dev loop.

One prompt, one response, done. The SDK is faster to write, easier to debug, has zero dependency overhead.

# LangChain
from langchain_anthropic import ChatAnthropic
model = ChatAnthropic(model="claude-opus-4-7")
response = model.invoke([HumanMessage("...")])

# SDK
from anthropic import Anthropic
client = Anthropic()
response = client.messages.create(
    model="claude-opus-4-7",
    max_tokens=1024,
    messages=[{"role": "user", "content": "..."}],
)

The SDK version is two more lines. In exchange, you get: full IDE autocomplete on the response shape, better error messages, no framework abstraction layers when something goes wrong, no surprise breaking changes.

For a junior engineer or someone new to LLMs, LangChain hides the actual API behind multiple layers. You learn LangChain instead of learning how LLMs work.

The SDK forces you to understand: messages, tool calls, tokens, structured output. All concepts that transfer regardless of which framework you eventually use.

LangChain has overhead. Multiple wrapper layers, callback machinery, Runnable abstractions. For high-throughput services (thousands of requests per second), this overhead is non-trivial.

Direct SDK calls are leaner. If the difference matters at your scale, skip the framework.

Be honest. If the actual scope is "extract structured data from invoices using an LLM," that's a single function with with_structured_output on the SDK. It doesn't need a framework.

The framework's value is in handling complexity. If your problem isn't complex, the framework adds complexity instead of removing it.

LangChain's API changed substantially in 2023 and 2024. Code from earlier versions doesn't run. Tutorials online are mixed across versions.

If you're shipping something that needs to work for years with minimal maintenance, the SDK is more stable. Provider APIs change too, but they change slower and with longer deprecation windows.

Some engineers actively dislike LangChain. The criticisms are well-known: heavy abstraction, opaque errors, churning API, kitchen-sink scope. If your team feels this way, fighting them on framework choice isn't worth it. They'll write better code in the tool they prefer.

If your "LLM use case" is actually "I have data in a known schema and I want to do X with it," consider whether you need an LLM at all. Many problems labeled "AI" are well-served by:

• Regex / parsing
• Classical ML (sklearn, XGBoost)
• SQL queries
• Rule engines

LLMs are great for fuzzy, language-heavy work. They're slow and expensive for structured data manipulation. Pick the right tool.

For systems where downtime is unacceptable (medical, financial), LangChain adds dependencies that can break. Each integration is a potential failure point. Vendor APIs change. Versions conflict.

Direct SDK + minimal dependencies is more auditable, easier to lock down, and easier to certify.

Many production systems land here: LangChain for the parts that are commodity (loaders, splitters, retrievers, vector stores), direct SDK for the parts that matter (the actual model calls and orchestration).

# Use LangChain for ingestion
from langchain_community.document_loaders import PyPDFLoader, GitHubIssuesLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_postgres import PGVector

docs = PyPDFLoader("handbook.pdf").load()
chunks = RecursiveCharacterTextSplitter(...).split_documents(docs)
vectorstore.add_documents(chunks)

# Use SDK directly for the agent loop
from anthropic import Anthropic
client = Anthropic()

def agent(user_msg):
    # custom retrieval
    query_emb = embed(user_msg)
    docs = vectorstore.similarity_search_by_vector(query_emb, k=5)

    # direct API call
    response = client.messages.create(
        model="claude-opus-4-7",
        system=f"Answer using context: {format(docs)}",
        messages=[{"role": "user", "content": user_msg}],
        max_tokens=2048,
    )
    return response.content[0].text

You get the integration ecosystem without the orchestration overhead. This is increasingly the production-mature pattern.

Most LangChain/LangGraph production issues come from the same handful of root causes:

• Unbounded loops (no iteration caps).
• Unbounded context (no message trimming).
• Missing reducers on state fields.
• Wrong scope on memory (thread_id, store namespaces).
• Sync code in async context.
• Eval signals diverging from real quality.
• Treating ingestion as one-time setup.

Each is preventable with the right pattern. The patterns are well-known; what makes them production knowledge is having seen them break before. Run your agent against adversarial inputs, long conversations, and concurrent users before shipping. The bugs that don't appear in dev are the ones waiting in prod.