DEV Community: langchain

Your Agent Harness Is Paying LLM Prices for Classifier Decisions

Jayesh Tankariya — Wed, 14 Oct 2026 03:49:18 +0000

Your Agent Harness Is Paying LLM Prices for Classifier Decisions

Keep your LangGraph router. Let a decision model take the calls it's sure about, and prove the savings on your own traffic before you change anything.

The 60-Second Version

The Problem: Modern AI agents make dozens of small decisions on every run: which specialist agent handles this task, which tool to invoke, which model tier to use, or where to route next. Most teams make each of those decisions with a full frontier LLM call—paying LLM latency (800ms–3,000ms+) and LLM token prices.

The Reality: These decisions almost never generate new text. They select one option from a known, discrete list. They are classification problems masquerading as generative prompts.

The Approach: Put a fast System 1 decision model (TypeSafe's Jev) in front of the router you already have. When Jev is confident, it decides in ~100ms for a fraction of a cent. When it isn't, or if the API is ever unreachable, your existing LLM router decides exactly as it does today.

The Universal Scope: This isn't built for a single niche. It applies to any agentic architecture in any industry that uses LLM calls for control flow.

The Proof: A non-invasive shadow mode runs alongside your router without touching production behavior, measures agreement against your existing router, and recommends an optimal confidence threshold on your own live traffic.

To try it: pip install langgraph-jev-router and add one decorator (@jev_first) to your router function.

1. The Decisions Hiding Inside Your Agent Harness

The AI engineering community has settled on a fundamental architectural principle: a raw foundation model is not an agent; an agent is a model plus a harness.

The harness is everything surrounding the model: memory, state schemas, guardrails, tool bindings, and the control-flow graph that orchestrates what happens next.

Look closely at the control flow in a typical LangGraph workflow. Between an incoming user request and the final synthesized response, the graph evaluates multiple decision points:

Which specialist sub-agent should handle this intent? (Billing, Technical Support, Triage, DevOps, Sales)
Does this task require a heavyweight frontier model (Gemini 1.5 Pro / GPT-4o) or a lightweight worker (Flash / Haiku / 4o-mini)?
Which tool or MCP server should be called?
Does this input violate safety policies or require escalation?

Almost none of these choices produce free-form text. They evaluate input context and pick one discrete item from a known set of routes.

As analyzed in The Router Is the Harness, the decision layer of an agent is fundamentally a classification problem, yet production systems routinely pay generative LLM prices for it: roughly $0.000006 per decision for a typed classifier versus ~$0.02 for a frontier LLM call.

                           [ Incoming Request ]
                                    │
                                    ▼
                      ┌───────────────────────────┐
                      │   TypeSafe Jev Fast-Path  │
                      │   (~100ms System 1 reflex)│
                      └─────────────┬─────────────┘
                                    │
                 ┌──────────────────┴──────────────────┐
                 │                                     │
      Confidence ≥ Threshold                Confidence < Threshold
         (Clear intent)                     (Ambiguous / Edge case)
                 │                                     │
                 ▼                                     ▼
      ┌──────────────────────┐              ┌──────────────────────┐
      │ Direct Route Dispatch│              │ Existing LLM Router  │
      │ ⚡ ~100ms / No LLM   │              │ 🤖 Trusted Fallback  │
      └──────────────────────┘              └──────────────────────┘

Even when context caching shrinks token costs, the primary operational penalties remain: latency, jitter, and failure modes. A generative model requires multiple token-generation steps and is vulnerable to prompt drift, formatting hallucinations, rate limits (429), and upstream API outages. A typed classifier evaluates the input in a single pass, responds deterministically in ~100ms, and cannot be prompt-injected into an invalid route.

2. Why Not Just Replace Your LLM Router with a Classifier?

Because you would be trading a router you have painstakingly tuned, tested, and trust for a model you haven't validated on your data.

TypeSafe's Jev is a purpose-built decision model, but enterprise engineering teams should not blindly trust third-party confidence scores or unverified benchmarks with mission-critical routing.

The battle-tested architectural pattern is confidence-gated routing:

The classifier handles the high-confidence, unambiguous queries (the 60%–80% of routine traffic).
The existing LLM router acts as the authoritative fallback whenever the classifier is uncertain or ambiguous.

That pattern is exactly what langgraph-jev-router implements as a drop-in library for LangGraph.

Adopting it cannot make routing worse than it is today. If Jev has any doubt, or if the network blips, your trusted router executes exactly as it does right now.

3. Universal Applicability: An Agent Primitive, Not an Industry Tool

A common misunderstanding is asking: "Which industry is this built for?"

langgraph-jev-router is not an industry-specific application. It is an infrastructure primitive for agentic harnesses, akin to an HTTP connection pool or an edge-caching layer.

It applies to any agentic system in any domain where an LLM is currently used to decide control flow:

Customer Experience & Support: Routing conversations to billing, account management, tier-2 technical triage, or human agents.
Enterprise DevOps & SRE Agents: Dispatching incident alerts to database, networking, infrastructure, or application specialists.
Developer Tools & Coding Agents: Choosing between code generation, unit test creation, linting, or documentation modules.
Financial Services & FinTech: Categorizing transactions, compliance disputes, fraud verification, or credit requests.
Healthcare & Clinical Workflows: Triaging intake requests to cardiology, billing, appointment scheduling, or emergency escalation.
Industrial & Manufacturing Operations: Routing plant floor safety reports to mechanical, electrical, chemical, or safety officers.

The industry benchmarks we publish (manufacturing, supply chain, healthcare, telecom) are not boundaries of the tool—they are empirical stress tests on messy, real-world domain vocabularies to prove that the architecture holds up under enterprise conditions.

4. One Decorator, and Your Router Becomes the Safety Net

Integrating langgraph-jev-router requires adding a single decorator to your existing routing function.

from typing import Literal
from langchain_core.messages import HumanMessage, AIMessage
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph_jev_router import jev_first, attach_routes

# 1. Define your routing options with clear docstrings/descriptions
ROUTES = {
    "billing": "Invoices, credit card charges, payment failures, and refund requests.",
    "technical": "System errors, stack traces, API timeouts, and login failures.",
    "account": "Password resets, team invites, SSO, and user profile management.",
}

# 2. Add one decorator to your existing router
@jev_first(
    routes=ROUTES,
    threshold=0.75,  # Only take over when 75%+ confident
)
def route_ticket(state: MessagesState) -> Literal["billing", "technical", "account"]:
    # ── YOUR EXISTING LLM ROUTER (Unchanged, acts as 100% safety fallback) ──
    print("🤖 Ambiguous or low confidence: Falling back to LLM router...")
    # Your standard OpenAI / Gemini / Claude call runs here
    return "technical"

# 3. Build your LangGraph as usual
builder = StateGraph(MessagesState)

builder.add_node("billing", lambda s: {"messages": [AIMessage("Handling Billing")]})
builder.add_node("technical", lambda s: {"messages": [AIMessage("Handling Tech Issue")]})
builder.add_node("account", lambda s: {"messages": [AIMessage("Handling Account")]})

builder.add_conditional_edges(START, route_ticket)
for node in ["billing", "technical", "account"]:
    builder.add_edge(node, END)

# 4. Attach routes and compile
# attach_routes reads node docstrings or route dicts as descriptions
graph = attach_routes(builder).compile()

When an incoming message arrives:

Jev analyzes the request against the route descriptions.
If confidence >= 0.75$, Jev dispatches directly to the destination node in ~100ms without invoking your LLM router.
If confidence < 0.75$, your existing LLM router function executes normally.

What if you route using prebuilt supervisors?

Many LangGraph developers use prebuilt abstractions like create_supervisor from langgraph-supervisor, where the routing function is compiled internally and cannot be decorated directly.

For these graphs, langgraph-jev-router provides install_shortcut:

from langgraph_jev_router import install_shortcut

# Intercept before the prebuilt supervisor node
handle = install_shortcut(workflow, before="supervisor")
app = workflow.compile()

When Jev is confident, the request skips the supervisor's LLM call entirely and routes directly to the specialist worker node. When uncertain, control passes to the supervisor node unaltered.

5. Engineering Safety Defaults & The "Loop Trap"

Most of the engineering in langgraph-jev-router went into handling non-ideal edge cases gracefully:

Situation	Runtime Behavior
Jev is confident ($\ge \text{threshold}$)	Jev dispatches route; LLM router call is bypassed
Jev is below threshold	Existing LLM router decides; zero disruption
Network timeout, 500 error, or no API key	Existing LLM router decides; one warning logged; graph never crashes
Jev predicts route not in valid list	Fallback to LLM router
Intermediate agent/tool spoke last	LLM router decides (preserves multi-turn reasoning)
Terminal state (`END`)	Never chosen by Jev unless explicitly opted in
Identical input arrives again	In-memory cache returns route instantly (0ms)
Kill switch: `JEV_ROUTER_DISABLED=1`	Jev is completely disabled globally; 100% LLM execution

The "Loop Trap" Bug in Cyclic Graphs

One of the most critical bugs uncovered during testing occurs in cyclic multi-agent graphs:

[Router] ──> [Worker Agent] ──> [Tool Execution] ──> [Router] ──> ...

In a loop-shaped graph, a worker agent finishes an action and hands control back to the router. A naive fast-path router inspecting only the original user request will see the initial prompt again and re-route right back to the same worker agent—creating an infinite execution loop.

At that stage in the conversation, the real question is no longer "Who should handle this request?" but rather "Is the task complete, or does it require another iteration?" Answering that requires analyzing the full conversation history—a task suited for the reasoning LLM.

To prevent this, langgraph-jev-router automatically inspects message sender history: Jev only activates on fresh user requests. Any internal loop or agent-to-agent handoff automatically routes to your full LLM.

6. Prove It on Your Traffic Before You Switch: Shadow Mode

Engineering leaders should never adopt routing changes based on vendor claims. You need to prove the behavior on your own production traffic.

langgraph-jev-router includes built-in Shadow Mode. In shadow mode, your existing LLM router makes 100% of the live routing decisions, while Jev executes asynchronously in parallel to observe, record, and evaluate agreement:

# Enable zero-risk shadow observation
export JEV_ROUTER_SHADOW=1
export JEV_ROUTER_LOG=jev_decisions.jsonl

# Run your normal traffic for a few days...

Then generate an automated agreement and savings report using the CLI:

jev-router report jev_decisions.jsonl --llm-cost-per-call 0.002

Output:

======================================================================
SHADOW MODE REPORT: 425 decisions compared
(Your existing router decided all traffic; Jev observed in parallel)
======================================================================
  Threshold | Jev Autonomous % | Agreement with LLM Router
     0.60   |   82.4% (n=350)  |   94.1%
     0.70   |   74.2% (n=315)  |   96.8%  <-- Recommended Threshold
     0.80   |   68.5% (n=291)  |   98.2%
     0.90   |   58.1% (n=247)  |   99.1%
----------------------------------------------------------------------
Recommendation: Set threshold=0.70 to eliminate ~74% of routing LLM calls
while maintaining 96.8% agreement with your current router.

(Note: Agreement is measured against your current LLM router. When disagreement occurs, manual inspection often reveals the classifier chose the correct category while the LLM drifted.)

7. Empirical Validation: The Industrial Safety Benchmark

To measure langgraph-jev-router against real-world data containing specialized technical jargon, we built an open-source evaluation suite: jev-router-benchmark.

The Dataset

We evaluated 425 real industrial plant accident reports from the IHM Stefanini Industrial Safety and Health Analytics Database (CC0 1.0 Universal Public Domain). Incidents involve complex operational hazards: molten metal splashes, hydraulic press pinch points, high-voltage flashovers, and chemical inhalation.

We benchmarked a standard Google Gemini Flash router against langgraph-jev-router across 50 real manufacturing incidents.

Results at Default Threshold (0.70)

Metric	Traditional LLM Router	Jev-First (`langgraph-jev-router`)	System Impact
LLM Calls Required	50 (100%)	13 (26%)	-74.0% LLM calls eliminated
Autonomous Jev Decisions	0 (0%)	37 (74%)	37 calls routed via fast reflex
Decision Latency	305 ms (baseline) / 25s+ (API queue)	116 ms (TypeSafe Jev)	Sub-120ms deterministic speed
Total Routing Cost	$0.00750	$0.00290	61.3% net financial reduction

The Executive Trade-off: Threshold Calibration Curve

How does threshold selection affect safety and LLM call reduction? We executed a multi-threshold sweep from 0.60 to 0.95 across the incident dataset:

Confidence Threshold	Calls Avoided (%)	Autonomous Jev Decisions	LLM Fallback Calls	Agreement with Ground Truth	Avg Latency
`0.60`	78.0%	39 / 50	11 / 50	56.0%	183 ms
`0.70`	74.0%	37 / 50	13 / 50	58.0%	82 ms
`0.80`	72.0%	36 / 50	14 / 50	60.0%	88 ms
`0.85`	64.0%	32 / 50	18 / 50	66.0%	112 ms
`0.90`	60.0%	30 / 50	20 / 50	70.0%	125 ms
`0.95`	56.0%	28 / 50	22 / 50	72.0%	137 ms

Architectural Takeaway:
Even if you configure an ultra-conservative threshold of 0.95, Jev still eliminates 56% of routing LLM calls at an average latency of 137ms. Engineering teams have a predictable knob to tune cost savings against accuracy preservation.

8. Where This Fits in the Ecosystem

Different tools address model routing at different layers of the AI stack:

Gateway Layer (e.g. LiteLLM Proxy): Dispatches requests across model providers or cost tiers before requests reach the application code.
Middleware Layer (e.g. LangChain create_agent middleware): Filters tool calls or model tiers during agent initialization.
Graph Orchestration Layer (langgraph-jev-router): Operates inside the state graph at conditional edges, supervisors, and handoff nodes, with zero-risk fallback to your application's custom LLM logic.

9. For Engineering Leaders: 5 Questions Before Approving a Pilot

If you are a CTO, VP of Engineering, or Principal Architect evaluating this architecture, here is the operational reality:

What changes on Day 1? Nothing. With JEV_ROUTER_SHADOW=1, Jev runs asynchronously in parallel. Your current LLM router continues to make 100% of routing decisions.
What is the rollback mechanism? Set JEV_ROUTER_DISABLED=1 in your environment variables and restart. The library completely disengages with zero code changes.
What data leaves our perimeter? Only the newest user prompt and your route candidate names/descriptions are sent to TypeSafe's API endpoint. Message history is never logged or stored by default.
Can vendor model updates silently drift our routing? No. The model version (jev-1-preview) is explicitly pinned by default.
How do we quantify success? The CLI report provides exact numbers: LLM calls bypassed, fallback rates, agreement percentage, and net cost reduction.

10. What's Next in This Series

Empirical claims require open benchmarks. In the upcoming articles of this series, we examine full benchmark evaluations across diverse operational domains:

Manufacturing (Published): Industrial plant safety incident classification (GitHub Benchmark).
Supply Chain & Logistics: Real-time carrier exception and shipment routing.
Healthcare Operations: Patient clinical intake and urgency triage.
Telecommunications: Customer service intent dispatch and billing escalations.

All benchmark code, raw dataset references, and calibration scripts are publicly available and reproducible.

🛠️ Get Started

Install the library:

pip install langgraph-jev-router

Verify your setup:

export TYPESAFE_API_KEY="your_api_key"
jev-router check

GitHub Repository: https://github.com/JTankariya/langgraph-jev-router
Benchmark Suite & Data: https://github.com/JTankariya/jev-router-benchmark

Note: langgraph-jev-router is an independent open-source project released under the MIT license. Not affiliated with LangChain, Inc. or TypeSafe AI.

0 of 12 AI agent frameworks can cap what an agent spends in dollars (we checked)

Aidakh — Tue, 13 Oct 2026 17:53:37 +0000

Disclosure: I work on Pink Agentic AI Payments at PinkWallet, a payment-control layer for AI agents. This survey covers general-purpose agent frameworks, not payment products, and Pink is not in the table. Every cell links to the framework's own docs: github.com/Pink-Agentic-Payments/agent-framework-spend-guardrails.

Short answer: none of the 12 most popular agent frameworks documents a spending budget in currency for tool calls. They can make an agent stop after N steps (9 of 12) or pause for a human (10 of 12), but none of them knows what a single tool call costs in dollars. If your agent can spend money, the cap has to come from somewhere else.

The 12 frameworks and 8 controls

We checked each framework's own documentation or source code on 2026-10-09 for eight controls:

C1 human approval before a tool runs
C2 a programmatic per-tool hook that can block a call
C3 max steps / iterations / turns
C4 a token or LLM-cost budget for a run
C5 a money budget for tools, in currency
C6 a tool-call rate limit
C7 idempotency / dedup of tool calls
C8 a built-in audit log of tool calls with arguments

Framework	C1	C2	C3	C4	C5	C6	C7	C8
LangChain/LangGraph	yes	yes	yes	partial	no	yes	no	n/d
CrewAI	yes	yes	yes	no	no	yes	no	partial
OpenAI Agents SDK	yes	yes	yes	n/d	no	n/d	n/d	partial
Vercel AI SDK	partial	yes	yes	partial	no	n/d	n/d	partial
Google ADK	yes	yes	yes	no	no	n/d	n/d	partial
Microsoft AutoGen	partial	n/d	yes	yes	no	n/d	n/d	n/d
Semantic Kernel	n/d	yes	n/d	n/d	no	n/d	n/d	partial
LlamaIndex	yes	n/d	n/d	n/d	no	n/d	n/d	n/d
Pydantic AI	yes	yes	yes	yes	no	yes	n/d	n/d
Mastra	yes	yes	n/d	partial	no	n/d	n/d	n/d
Claude Agent SDK	yes	yes	yes	n/d	no	n/d	n/d	n/d
smolagents	no	no	yes	no	no	no	no	partial

n/d = not documented.

What the table says

0 of 12 document a currency budget for tool calls (C5). That is fair to say plainly: these are general-purpose frameworks, so a dollar cap was never their job.
10 of 12 offer some form of human approval (C1 yes or partial), but not all of them can gate one specific tool call; some only stop the whole run.
9 of 12 can block a specific tool call in code (C2), not just by prompt instruction. That hook is where a spending check can be plugged in.
9 of 12 document a max-steps limit (C3), the most common control in the survey.
Only 2 of 12 document a clean token or cost budget for a run (C4 = yes), with 3 more partial. And a token budget limits what the model costs, not what the agent buys.

So where should the spending cap live?

Not in the prompt. "Never spend more than $50" is text in the model's context; it can be misread, forgotten, or overridden by a prompt injection.

Use the framework hook (C2) to call something outside the model that knows the money: the payment API or a control layer in front of it. That layer should hold the budget, the per-payment cap, the allowed payees and the approval rule, and it should answer before the payment executes. Three outcomes cover almost every case: allow, ask a human (with silence meaning no), or block.

A companion census of 49 payment MCP servers found the same gap from the other side: of 37 servers that can move money, only 4 document a spending budget and only 6 document human approval (payment-mcp-census).

Try it

Pink's public sandbox gives an agent seven MCP tools (pink.get_budget, pink.check_policy, pink.request_payment and friends) and answers every request with allowed / pending_human / blocked against per-agent budgets and rules. Test money only: agentic-sandbox.pinkwallet.com. There are runnable LangGraph, OpenAI Agents SDK, CrewAI and Vercel AI SDK examples in sandbox-examples.

If we got a cell wrong for your framework, open an issue with a docs link and we will fix it.

NestJS + LangChain: Build an AI Agent in 5 Minutes

Martin-Bruel — Tue, 13 Oct 2026 14:49:07 +0000

To add an AI agent to a NestJS app, register it as a module with nestjs-langchain, inject it like any provider, and call run(). This tutorial goes from nest new to a working POST /ask endpoint in about 30 lines of TypeScript.
Tested with NestJS 12, LangChain 1.x, nestjs-langchain 2.0 and Node 22.12.

How do I add LangChain to a NestJS project?

Create the app, then install the library and a model provider:

npx @nestjs/cli new my-app --package-manager npm
cd my-app
npm install nestjs-langchain @langchain/openai

@langchain/openai is the provider: @langchain/anthropic, @langchain/ollama, @langchain/mistralai or any other LangChain chat integration works the same way.

The provider reads its key from the environment:
export OPENAI_API_KEY=sk-...

How do I register the agent?

Import LangChainModule in app.module.ts:

import { Module } from '@nestjs/common';
import { LangChainModule } from 'nestjs-langchain';
import { AppController } from './app.controller.js';
import { AppService } from './app.service.js';
import { AskController } from './ask.controller.js';

@Module({
  imports: [
    LangChainModule.register({
      // `provider:model`; the provider reads OPENAI_API_KEY.
      model: { model: 'openai:gpt-5-mini' },
      systemPrompt: 'You are a helpful assistant. Answer in one paragraph.',
    }),
  ],
  controllers: [AppController, AskController],
  providers: [AppService],
})
export class AppModule {}

ts
model names the provider and the model: switching to Claude is anthropic:claude-sonnet-5-5, with @langchain/anthropic installed. The agent is built when the app starts, so a wrong configuration shows up there, not on the first request.

How do I call the agent from a controller?

Inject Agent like any provider, in a new ask.controller.ts:

import { Body, Controller, Post } from '@nestjs/common';
import { Agent } from 'nestjs-langchain';

@Controller('ask')
export class AskController {
  constructor(private readonly agent: Agent) {}

  @Post()
  async ask(@Body('question') question: string): Promise<string> {
    const { output } = await this.agent.run(question);
    return output;
  }
}

Besides output, the answer, run() returns the tools the model called, the tokens it used and how long the run took.
Start the app and ask it something:

npm run start:dev

curl -X POST http://localhost:3000/ask \
  -H 'Content-Type: application/json' \
  -d '{"question": "What is dependency injection?"}'

The response is the model's answer, as plain text.

Next: give the agent tools

An agent becomes useful when it can act on your app. With @tool(), a service method becomes a tool the model can call:

@Injectable()
export class MathService {
  @Tool({ description: 'Adds two numbers.' })
  add(
    @ToolParam({ name: 'a' }) a: number,
    @ToolParam({ name: 'b' }) b: number,
  ): number {
    return a + b;
  }
}

Then tools: [MathModule] in register() hands it to the agent. The next article in this series covers tools in detail.

• Full sample: samples/chat
• Package: nestjs-langchain on npm
• Source, issues and stars: GitHub

AI Agent Frameworks Compared: LangGraph vs CrewAI vs AutoGen

Avenotech Team — Tue, 13 Oct 2026 03:08:41 +0000

Why is the most-starred framework in this comparison the one you shouldn't start a new project on in 2026?

AutoGen holds roughly 60,400 GitHub stars, more than CrewAI's 57,100 and LangGraph's 39,800. It hasn't shipped a Python release in 320 days. Its last tagged version (python-v0.7.5) shipped on September 30, 2025. The repo hasn't had a commit pushed since April 15, 2026. And yet every "LangGraph vs CrewAI vs AutoGen" post being written right now is still treating it as a live option.

That gap, between stars and production reality, is what this post is actually about. Not a feature matrix. Not a benchmark that can't be audited. The evidence of where each of these frameworks breaks when you run it at scale, who maintains it, and what it actually costs when things go wrong.

Every fact here comes from the GitHub REST API, PyPI release history, official READMEs, vendor migration guides, and public issue trackers, all verified on 2026-08-16. Where evidence is one engineer's account rather than a measurement, it's labeled that way.

Key Takeaways

AutoGen is in maintenance mode per its own README, and the Python package hasn't shipped a stable release in about 320 days. The successor is Microsoft Agent Framework, GA since 2026-04-03. New projects should start there. AG2 is a real option for legacy codebases, and it's a fork, not a rebrand.
Stars lie; installs are closer to the truth. AutoGen leads on stars (about 60K) and trails on monthly PyPI installs (about 1M). LangGraph has fewer stars (about 40K) and runs at about 71M installs, roughly 65x AutoGen. The gap survives every honest caveat.
Treat the star-to-install ratio as a freshness signal in reverse. Stars accumulate at peak hype; installs track present-tense production dependence. A high ratio means a project was once exciting and now isn't load-bearing anywhere.
Verified 2026-08-16 against LangGraph 1.2.11, CrewAI 1.15.16, AutoGen 0.7.5 and Microsoft Agent Framework 1.0. This category moves within weeks, so check those versions on PyPI before you act on any of it.
Don't pick on star counts. Pick on what shipped, what broke, and what's still being maintained.

If you're scoping this for a real system rather than a spike, the framework choice is one input into building a production agent system, and it's rarely the input that decides whether the thing ships.

Which agent framework should you use in 2026?

The short answer, before the evidence.

Pick LangGraph if you need explicit, auditable control flow and durable state, and you can staff the persistence work it brings. Pick CrewAI if you're prototyping multi-agent workflows fast and you accept that delegation is model-mediated. Don't start on AutoGen: as of August 2026 its README carries a maintenance-mode banner directing new users elsewhere. If you're already on AutoGen, your target is Microsoft Agent Framework, GA since April 2026, and the move is a rewrite rather than an upgrade. And if your workflow is three LLM calls in a loop, use no framework at all.

Why that answer isn't the usual one

Every framework here can call a tool and pass a message, so a feature matrix can't separate them. What separates them is what fails at scale: LangGraph breaks on checkpoint growth, CrewAI breaks on prompt-mediated delegation, AutoGen breaks on maintenance. The rest of this post is the evidence for each, with the tracker issues and post-mortems that document them.

The metric that should matter (but doesn't show up in comparison posts)

Stars are a lagging indicator. They record what got bookmarked at peak hype, which for AutoGen was the v0.2 launch era in 2024. Stars accumulate indefinitely and are never removed, even when a project goes quiet. Installs are the opposite: a recurring production cost, paid every time CI runs or an image rebuilds.

The numbers have completely decoupled.

Monthly PyPI installs, last 30 days

Category	Value
LangGraph	71.1M
CrewAI	20.7M
MS Agent Framework	1.6M
AutoGen	1.1M

LangGraph ships roughly 71 million monthly PyPI installs. AutoGen ships about 1.08 million. That's 65x more installs for the framework with fewer stars. CrewAI sits at roughly 20.7 million monthly installs, real production usage by any reading, but a fraction of LangGraph's volume.

A 65x gap in installs against a lead in stars is a strong signal about where the production load actually is. It's not a popularity contest. It's a load map.

GitHub footprint: stars vs forks

Category	Stars	Forks
AutoGen	60K	9K
CrewAI	57K	8K
LangGraph	40K	7K
MS Agent Fw.	13K	2K

Here's the thing that makes it a procurement problem: PyPI counts include CI pipelines and container rebuilds, so they're not user counts. Treat them as a relative signal, not an absolute. And LangGraph benefits from being pulled in transitively by the wider LangChain ecosystem, which inflates its number relative to a standalone install. AutoGen has no such tailwind, and the gap is still enormous. A 65x gap survives both caveats. A 1.2x gap wouldn't.

Release cadence tells the same story from a different angle. Days since the last release, as of 2026-08-16:

AG2: 1 day
CrewAI and MS Agent Framework: 2 days
LangGraph: 5 days
AutoGen: 320 days

One more cadence note before we get to the frameworks. CrewAI shipped 1.15.16 on August 14, 1.15.15 on August 12, 1.15.14 on August 8, and two releases on August 5. That's roughly ten releases in nineteen days. Whether that signals fast fixes or fast churn depends on how much you trust their release process, but either way, lock your version before you start.

The mental models (and why the model matters)

	LangGraph	CrewAI	AutoGen
Best for	Stateful, controllable graphs	Fast multi-agent prototypes	Nothing new in 2026
Latest stable	1.2.11 (2026-08-11)	1.15.16 (2026-08-14)	0.7.5 (2025-09-30)
Monthly PyPI installs	71,095,727	20,672,935	1,082,331
Licence	MIT	MIT	Ambiguous, see below
Orchestration model	Explicit graph, developer-defined edges	Manager agent, prompt-mediated	Group chat, speaker selection
Control flow	Enforced by the graph	Suggested by the prompt	Suggested by the prompt
State	Checkpointed, persistent, JSON-shaped	Task outputs plus optional memory	Agent-held conversation history
Default loop ceiling	1,000 supersteps	20 iterations per agent	String-matched termination
Maintenance status	Active (5 days since release)	Active (2 days since release)	Maintenance mode (320 days)
Our verdict	Default choice for production	Good for speed, verify delegation	Migrate off

Framework comparison as of 2026-08-16. Sources: GitHub REST API, PyPI, official documentation.

LangGraph is a directed graph with a central state object. Agents are nodes. Edges define flow. Conditional edges handle branching. Reducers merge concurrent state updates. You draw the control flow; the LLM executes within it. The mental model is a state machine, and the framework enforces it in code.

Built-in primitives: durable execution (checkpointing, resume from interruption), human-in-the-loop (interrupt the graph, collect input, resume), short-term and long-term memory. It's the most explicit of the three about where control lives.

CrewAI is a role-based team. Every agent carries a role, a goal, a backstory, and tools. A Crew executes tasks sequentially or hierarchically. In hierarchical mode, an auto-generated manager agent delegates and reviews. The runtime provides the hierarchy; the delegation decisions inside it are model-mediated rather than encoded as developer-defined conditional edges. Flows, a newer primitive, are event-driven pipelines that offer more predictable production control: one LLM call per step, no autonomous delegation.

AutoGen, in its current maintained mode, is a conversation-driven multi-agent runtime. Two-agent chats, group chats with an LLM-powered speaker selector, sequential and nested conversations. The successor, Microsoft Agent Framework, has the same patterns but a different architecture. AutoGen itself is not being developed.

AutoGen's licence deserves a flag of its own, because it's a procurement problem rather than a technical one: GitHub detects the repo as CC-BY-4.0, the README describes a dual CC-BY-4.0 and MIT arrangement, and PyPI reports MIT. Your legal team will not enjoy that conversation.

Where LangGraph breaks in production

The failure point isn't the API. It isn't the tooling either.

It's the checkpointer, and specifically, what happens to your Postgres instance when nobody has planned for checkpoint growth.

Tadeo Donegana Braunschweig, an engineer at Tiendanube, published the clearest public account of this in April 2026. At more than 120,000 conversations per week, his team's checkpoint tables grew without bound. The open-source checkpointer ships no native TTL. An average conversation produced roughly 93 rows across four checkpoint tables after just two messages and three tool calls.

Two messages, and your database is ninety-three rows heavier.

In staging, checkpoint_blobs alone hit 56 MB after a single week of traffic. His summary: operational databases should store operational state, not historical exhaust.

This isn't one team's unlucky configuration. On the LangGraphJS tracker, issue #1138, "How do I keep data in Postgres checkpointer database from growing unbounded?", was opened in April 2025 and is still open. LangChain's own documentation suggests running a cron job to delete old checkpoints. That's a clear statement about where lifecycle management lives.

The vendor shipped a fix, and it's worth reading the numbers carefully because they get misquoted. PR #7586 introduced DeltaChannel, which stores a small sentinel value plus replay instructions instead of re-serializing the full accumulated state at every step.

Checkpoint storage for a 500-turn thread

Category	Value
add_messages (default)	221 MB
DeltaChannel, snapshots every 50	10.53 MB
DeltaChannel, no snapshots	1.98 MB

The default reducer writes more than a hundred times what the tightest DeltaChannel configuration does, and the middle row is the setting most teams actually want.

Three caveats before you bank on it. It landed in LangGraph 1.2.0 on 2026-05-12, it's marked beta, and it's opt-in: upgrading does nothing until you actually annotate the channels you want it on. Bloat already written to your tables stays written.

Here's where it gets expensive for teams that didn't budget the operational work. LangGraph's checkpoint packages (checkpoint, checkpoint-postgres, checkpoint-sqlite) version separately from core. If you pin core and float the checkpointers, you've secured the wrong half of the stack. Budget the persistence work before launch, not after

Braunschweig's three fixes generalise. Move checkpoint table creation into Alembic migrations rather than calling LangGraph's setup(), so your schema is version-controlled like everything else. Run recurring TTL deletion, paired with a weekly job that archives history to object storage if you need it for analysis. And checkpoint selectively: persist only the supervisor graph and keep sub-graphs stateless. None of this is exotic work, which is exactly why it gets skipped when an agent prototype gets bolted onto a product that already exists.

The typed-state complaint

Separate from persistence, there's a recurring ergonomics objection about state typing, and it needs stating carefully because the strong version of it is wrong. StateGraph accepts a real schema: TypedDict, a dataclass, or a Pydantic BaseModel if you want recursive validation. LangGraph is not untyped.

The gap is one level up. LangChain's own documentation notes that the higher-level create_agent factory does not support Pydantic state schemas, and the factory is where most teams start. So the typing is available on the API you graduate to, not the one you begin on. One engineer on Hacker News put the frustration this way:

"My experience with langgraph is you spend so much time just fixing stupid runtime type errors because the state of every graph is a stupid JSON blob with very minimal typing, and it's so hard figuring out how data moves through the system."

That's one account. But it's not floating free of evidence, and the evidence points at the same seam. Issue #5891 (TypeError: Type is not msgpack serializable: Send) is a create_react_agent bug, which is exactly the factory path that can't take a Pydantic schema. Issue #740 (InvalidUpdateError on state writing) names the same root cause: state shape discovered at runtime rather than at definition time.

The practical read: LangGraph gives you the tools to keep state contracts honest, and it doesn't force you to use them. The risk isn't missing typing, it's contracts drifting across nodes because nothing made you declare them. Define the schema explicitly at the boundaries that matter, and most of this class of bug disappears.

Where LangGraph is actually strong

Explicit node structure makes execution topology predictable: the graph decides which nodes run and in what order, rather than the model deciding for itself. Token usage is a separate question, and it stays dependent on prompt size, retrieved context, tool outputs, retries and model behaviour. Predictable structure is not a predictable bill, which is why the cost section below argues what it does.

What the structure does buy you is auditability. LangGraph's control flow is enforced in code, not suggested in a prompt, so when a compliance reviewer asks which agent can call which tool under what condition, you can point at an edge instead of a paragraph of instructions.

Where CrewAI breaks in production

Delegation.

Not as a theoretical concern, but as a documented, reproducible failure mode that's been sitting in the issue tracker for over a year.

The 15,759-token incident. Partha Sarkar documented this in detail: a three-agent support crew (triage, technical, billing) running in hierarchical mode received a purely technical query. The manager invoked all three agents, including billing. The billing agent's response then overwrote the correct technical answer. The run consumed 38 seconds and 15,759 tokens to produce roughly 200 tokens of output. Nothing errored.

Root cause: hierarchical mode doesn't enforce conditional branching or true delegation. The manager decides, and when the model's judgment wobbles, the control flow wobbles with it. Look at what the failing workload actually was: routing a support question to the right specialist is ordinary chatbot routing, and it's the case a multi-agent crew is supposed to handle best. His fix was to replace the built-in manager with a custom manager agent carrying explicit step-wise instructions. After that change, technical-only runs took 24 seconds and about 10,000 tokens. You recover roughly a third of the cost by hand-rolling the orchestration the framework was supposed to handle.

Three delegation bugs in eleven months

They all show the same theme. Issue #2606 (April 2025): manager passes dicts while the schema expects strings. Issue #4783 (March 2026): manager executes tasks with its own tools and never delegates, collapsing hierarchical into sequential. Closed as not_planned. Issue #3925 (November 2025): with English task descriptions the manager delegates correctly; with Japanese descriptions the manager answers alone without delegating. Reproducible in both directions.

Sit with what that last one implies. The control flow of a production system changed because the natural language of its task descriptions changed. No framework that enforces orchestration in code can exhibit that behaviour, which makes #3925 the cleanest available proof that CrewAI's hierarchy is a request written into the prompt, not a rule the runtime enforces. If you run CrewAI in a non-English market, make this your regression test before you ship. Comment counts are not an engagement signal on this repo

One methodology note from reading the tracker directly, because it changes how you should read any CrewAI issue thread. CrewAI's highest-comment issues (#4877 at 271 comments, #5802 at 98, #4560 at 97, #5888 at 82) all carry zero reactions, which is an odd shape for genuinely contested threads. Reading #5802 explains it: the issue author repeatedly pitches their own product, and one commenter opens with "Hey! I'm Hermes Agent, an autonomous coding AI agent. I saw this issue and thought I could help." Use reactions and maintainer replies instead. In 2026 that's a general lesson about evaluating open source, not a knock on CrewAI specifically.

The open security issue

Issue #5057, "Memory content injected into system prompt without sanitization enables indirect prompt injection," has been open since March 2026 and was last updated in July. Mechanically, lite_agent.py concatenates retrieved memories straight into the system prompt. Anything written to memory inherits system-level instruction authority. It's rated Medium and cites the OWASP Agentic Security Index. Nearly five months on, it's still open. If you use CrewAI memory with any user-influenced content, apply the recommended mitigation (moving memory out of the system prompt and into user messages) before you ship.

The release cadence as a risk

CrewAI shipped its 1.0.0 rewrite on October 20, 2025, with breaking changes. Since then, the pace has been relentless, roughly a release every two days. That means your pinned version is end-of-life within weeks. It also means bugs get fixed fast. Weigh both.

Where CrewAI is actually strong

For teams that need to ship a multi-agent prototype quickly, CrewAI's higher-level abstractions are genuine. The role/goal/backstory model means you can get a working crew running in hours. The 20.7 million monthly installs reflect real production usage. And if something breaks, there's a good chance a fix is already in the next release.

Where AutoGen breaks (and why "migrate off" is the right starting point)

AutoGen is in maintenance mode. That's the sentence that should end the discussion for anyone starting a new project. But it's worth understanding what that means in practice, because there are teams already running on it who need to know the shape of their exit.

The README carries a maintenance-mode banner verbatim: "AutoGen is now in maintenance mode. It will not receive new features or enhancements and is community managed going forward. New users should start with Microsoft Agent Framework." Last tagged release: python-v0.7.5, September 30, 2025. Last repo push: April 15, 2026. That's roughly ten and a half months without a release, and no archive banner marks the repo, so at a glance the project doesn't look dead. It just is.

Migrating off AutoGen is an architecture review, not a dependency bump. The behaviour that changes underneath you is invisible until a user reports it.

The lineage is worth knowing. Original authors Chi Wang and Qingyun Wu left Microsoft in November 2024 and forked the project to AG2 (ag2ai/ag2), now at v1.0.x under Apache-2.0. Microsoft shipped Microsoft Agent Framework (MAF) 1.0 GA on April 3, 2026, developed by the core AutoGen and Semantic Kernel teams. MAF ships for Python and .NET under MIT, supports A2A and MCP interop, and has stable APIs with an LTS commitment. Official migration guides exist for AutoGen to MAF and Semantic Kernel to MAF.

The migration is a rewrite, and one change is silent

Microsoft's own migration guide reads as a from-scratch guide, not an upgrade guide. AssistantAgent becomes Agent, and the state semantics invert underneath the rename. AssistantAgent maintained conversation history as part of its state. Agent is stateless and does not maintain history between invocations. Nothing errors when you port the code. Your agent doesn't throw an error, it just stops remembering, and you discover it from a customer complaint.

There's a second inversion in the same rename. AssistantAgent is single-turn unless you raise max_tool_iterations. Agent is multi-turn by default. So the migration turns one component from "remembers, but acts once" into "acts many times, but remembers nothing."

Both inversions hide inside a port that looks like a rename (the_silent_migration_bug.py):

# AutoGen: single-turn by default, and the agent holds the history
agent = AssistantAgent(name="support", model_client=client, tools=[lookup])
await agent.run(task="What did I order?")
await agent.run(task="And when does it ship?")   # remembers the first turn

# Agent Framework: the port compiles, runs, and errors on nothing
agent = Agent(name="support", chat_client=client, tools=[lookup])
await agent.run("What did I order?")
await agent.run("And when does it ship?")        # no memory of turn one

# Nothing raised. The agent just answers the second question blind,
# and you find out when a customer complains. Thread state is now
# yours to carry explicitly between invocations.

One more gap to scope before you commit: AutoGen supports local code execution tools, and Microsoft's guide states that this feature is planned for future Agent Framework versions. Today it's a hole.

Termination by string matching

AutoGen terminates when an LLM generates the exact token TERMINATE. Because models are sloppy with exact tokens, they emit terminate, Terminate., Task terminated, or wrap it in markdown. None of those match, so the loop continues and you pay for every turn. Speaker selection fails the same way: if the selector's output doesn't exactly match an agent name, the same agent gets re-selected repeatedly, producing a single-agent infinite loop.

Issue #12 ("force TERMINATE") dates to July 2023. On classic AutoGen, pair max_consecutive_auto_reply with a token ceiling at your gateway before you deploy anything. Note that the knob doesn't carry over to Agent Framework, which is one more reason to treat the migration as a rewrite. And a framework in maintenance mode isn't shipping fixes for any of it.

One honest caveat

AG2 is a real option for teams on classic AutoGen with heavy investment and no appetite for a Microsoft-shaped migration. The original authors maintain it. It's at v1.0.x under Apache-2.0, actively released, with the classic API preserved in a separate ag2-classic repo. It's a smaller ecosystem and a fork rather than a corporate-backed project, so weigh that accordingly.

What multi-agent orchestration actually costs

Roughly 2.3x a sequential baseline, if you pick the most accurate pattern.

Kulkarni and Kulkarni benchmarked multi-agent architectures on SEC filings (arXiv:2603.22651, March 2026). Reflexive self-correcting loops achieved the highest field-level F1 (accuracy per extracted field) at 0.943, at 2.3x the cost of a sequential baseline. Hierarchical supervisor-worker reached F1 0.921 at 1.4x cost, the best cost-accuracy point in the study. A hybrid recovered 89% of the reflexive gains at 1.15x baseline cost.

Relative cost of four orchestration patterns

Category	Value
Sequential	1x
Hybrid	1.15x
Hierarchical	1.4x
Reflexive	2.3x

Those patterns map onto the frameworks directly. CrewAI's hierarchical process is the supervisor-worker row. LangGraph's cyclic graphs are how you build the reflexive loop. AutoGen's group chat is a third shape, less disciplined than either. Practically: if you're paying 2.3x for reflexion across every field in a document, you're overpaying, because the hybrid recovers most of the accuracy for a 15% premium.

None of these three will stop a run on spend

One finding that should shape how you budget this: across LangGraph, CrewAI and Microsoft Agent Framework, every available control knob bounds structure. None bounds money.

Framework	Knob	Default	What it actually bounds
CrewAI	max_iter	20	Iterations per agent
CrewAI	max_retry_limit	2	Retries, and it multiplies spend
CrewAI	respect_context_window	True	Summarizes to stay under context
LangGraph	recursion_limit	1,000	Supersteps per execution
Agent Framework	max_iterations	100	Workflow iterations

Every knob bounds structure. None bounds money. Sources: official CrewAI, LangGraph and Microsoft Agent Framework documentation, retrieved 2026-08-16.

Two problems hide in that table.

Default ceilings differ by 50x and nobody compares them. LangGraph allows 1,000 supersteps. MAF allows 100 workflow iterations. CrewAI allows 20 iterations per agent. CrewAI's looks tightest until you notice it's per-agent, so a five-agent crew's effective ceiling multiplies, and max_retry_limit=2 multiplies it again.

A structural limit also treats a 200-token call and a 200,000-token long-context call as identical events. That's why teams file "we set max_iter, why was the bill still huge?" tickets. You capped the number of steps, but your bill is counting tokens, and the two move independently. Watching it happen isn't an option either: CrewAI's usage metrics are strictly post-hoc, available after kickoff() completes. The bill only shows up once the run has already finished and been paid.

If spend is a hard constraint, put the cap at your gateway or proxy layer, where you can actually count tokens as they're spent.

When to skip the framework entirely

Sometimes the right agent framework is none.

Gartner predicted that over 40% of agentic AI projects will be canceled by the end of 2027 (press release, June 2025). Cost, unclear business value, and inadequate risk controls are the leading reasons. Cost and risk controls are exactly the two things these frameworks don't give you natively.

Anthropic Engineering's "Building Effective AI Agents" makes the case most directly: the most successful implementations weren't using complex frameworks but simple, composable patterns. If your workflow is a fixed sequence of three LLM calls with a retry, a framework adds an abstraction layer, a dependency tree, and a migration risk for very little. Plenty of production AI features never need orchestration at all, and finding that out early saves the migration you'd otherwise pay for later. Useful test: write the workflow as a plain function first. If that function reads cleanly, you don't need a framework. Once it starts accumulating retries, branches, and shared state, you do, and that's when agent development becomes an architecture problem rather than a library choice.

The verdict

Dimension	Winner	Why
Production reliability	LangGraph	Control flow enforced in code, not prompts
Time to first working crew	CrewAI	Higher-level abstractions; fastest to prototype
Real production usage	LangGraph	71M monthly PyPI installs; LangGraph is load-bearing
Release responsiveness	CrewAI	Ten releases in nineteen days; bugs get fixed fast
Licence clarity	LangGraph / CrewAI (tie)	Both MIT; AutoGen reports three different answers
Cost controls	None	None of the three halts a run on spend
Operational burden	CrewAI	LangGraph's checkpointer needs lifecycle management upfront
Enterprise support path	Microsoft Agent Framework	GA April 2026, stable APIs, LTS commitment
Overall for new projects	LangGraph	Best control-to-risk ratio for production work

Verdict by dimension, as of 2026-08-16.

Choose LangGraph if you need explicit, auditable control flow; you need durable execution or human-in-the-loop steps; your workflow has real branching and shared state; and your team will own the persistence work. Budget that work upfront: checkpoint TTL deletion, selective persistence on the supervisor graph only, and pinning the checkpoint packages separately from core. Its persistence is your problem, not the vendor's.

Choose CrewAI if you're shipping a multi-agent prototype this quarter, you want higher-level abstractions, and you can absorb the operational risk. Pin your version. Write delegation assertions into your test suite. Apply the #5057 memory mitigation before user content touches agent memory. Ten releases in nineteen days means your pinned version is end-of-life within weeks, so plan the upgrade cadence rather than discovering it.

Don't start on AutoGen if you're picking a framework today and have no legacy investment holding you there. If you're already on it, scope the MAF migration as a rewrite, put the stateless Agent behavior at the top of your test plan, and know that local code execution is a feature gap in MAF today. Its termination depends on string-matching a literal token, so pair max_consecutive_auto_reply with a token ceiling until you're off it.

Choose Microsoft Agent Framework if you're a Microsoft shop, you're migrating off AutoGen or Semantic Kernel, or a vendor-backed support path with an LTS commitment is what gets the project approved.

Choose no framework if your workflow is a few deterministic LLM calls and a plain function with retries is still readable. Write the function. Add a retry. Only reach for a framework when the function stops being readable.

The AutoGen lesson isn't that Microsoft abandoned a project. It's that a framework stuck at v0.x was never really a product: it's a research prototype that happened to get famous. Check the maintenance status, the release cadence, and the maintainer's roadmap before you commit. Everything in this post was accurate on 2026-08-16. For this kind of tool, that's not a long shelf life.

Frequently Asked Questions

Is AutoGen deprecated?

Not formally archived, but effectively yes. The README carries a maintenance-mode banner. Last stable Python release: autogen-agentchat 0.7.5 on 2025-09-30, about 320 days before 2026-08-16. The repo remains publicly readable and community-managed. Microsoft recommends Agent Framework for new projects.

What's the difference between AutoGen and AG2?

AG2 is a fork, not a rebrand. Original AutoGen authors Chi Wang and Qingyun Wu left Microsoft in November 2024 and forked the project. AG2 reached v1.0.0 in late July 2026. The classic autogen.* API lives in a separate ag2-classic repo, so 'move to AG2' is a migration. AutoGen stays with Microsoft under a mixed CC-BY-4.0/MIT licence situation; AG2 is Apache-2.0.

Does LangGraph handle its own database cleanup?

No. The open-source Postgres checkpointer ships no native TTL. LangChain's docs recommend a cron job to delete old checkpoints. LangGraph issue #1138 (open since 2025-04-29) reports around 100 rows per graph traversal across three tables with no maintainer resolution. Plan TTL deletion and archival before launch.

Can I cap how much an agent run costs?

Not within LangGraph, CrewAI or Microsoft Agent Framework. All three expose structural limits (recursion_limit at 1000, max_iter at 20 per agent, max_iterations at 100) but none halts execution on spend. CrewAI's usage metrics are post-hoc, available only after kickoff() returns. Enforce budgets at your API gateway or proxy.

Is multi-agent worth the cost over a single agent?

Depends on the accuracy you need. On SEC filings, reflexive loops hit F1 0.943 at 2.3x sequential cost; hierarchical hit 0.921 at 1.4x; a hybrid recovered 89% of the reflexive gains at 1.15x (arXiv:2603.22651, March 2026). Start hybrid. Escalate to reflexive only where accuracy pays for itself.

Should I trust framework star counts?

Treat them as a freshness signal in reverse. Stars accumulate at peak hype. Installs track present-tense production dependence. A high star-to-install ratio means a project was once exciting and now isn't load-bearing anywhere. AutoGen's ratio is the worst of the three by an order of magnitude, and the release cadence (320 days since last release) agrees.

Originally published on the Avenotech blog, where we write about shipping AI and mobile products to production. Figures were verified on August 16, 2026; check current versions before you act on them.

Production RAG with LangChain & LangGraph: Retrieval, Reranking & Agentic Workflows

Phil IPS — Mon, 12 Oct 2026 21:49:29 +0000

Production RAG with LangChain & LangGraph: End-to-End Implementation

Build a production-ready RAG system with document ingestion, hybrid retrieval, reranking, citations, guardrails, and monitoring.

Introduction

RAG (Retrieval-Augmented Generation) is the standard for grounding LLMs in proprietary data. This guide builds a production system — not a prototype.

Production RAG Requirements

Requirement	Prototype	Production
Ingestion	Manual, one-off	Automated, incremental, monitored
Retrieval	Vector only	Hybrid (dense + sparse + rerank)
Generation	Basic prompt	Structured output + citations + guardrails
Evaluation	Manual	Automated metrics + LLM judge + A/B
Monitoring	None	Latency, quality, cost, drift alerts
Deployment	Local notebook	Containerized, scalable, resilient

System Architecture

┌─────────────┐     ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  Ingest     │────▶│  Embedding  │────▶│  Retrieval  │────▶│ Generation  │
│  Pipeline   │     │  & Index    │     │  Pipeline   │     │  + Guardrails
└─────────────┘     └─────────────┘     └─────────────┘     └─────────────┘
       │                   │                   │                   │
       ▼                   ▼                   ▼                   ▼
  Chunking            Vector DB           Hybrid Search      Structured Output
  Metadata            (Qdrant)            (Dense+Sparse)     Citations
  Deduplication       Quantization        Reranking          Guardrails
  Incremental         Replication         Filtering          Evaluation

Part 1: Document Ingestion Pipeline

# ingestion/pipeline.py
from dataclasses import dataclass
from typing import List, Optional
from pathlib import Path
import hashlib
import asyncio
from datetime import datetime

@dataclass
class Document:
    id: str
    content: str
    metadata: dict
    source: str
    created_at: str
    chunk_hash: str = ""

@dataclass
class Chunk:
    id: str
    document_id: str
    content: str
    metadata: dict
    token_count: int
    embedding: List[float] = None

class IngestionPipeline:
    def __init__(
        self,
        chunker,
        embedder,
        vector_store,
        doc_store,
        deduplicator
    ):
        self.chunker = chunker
        self.embedder = embedder
        self.vector_store = vector_store
        self.doc_store = doc_store
        self.deduplicator = deduplicator

    async def ingest_file(self, file_path: Path, metadata: dict = None) -> dict:
        """Ingest a single file."""
        # 1. Load document
        doc = await self._load_document(file_path, metadata)

        # 2. Check for duplicates
        if await self.deduplicator.is_duplicate(doc.chunk_hash):
            return {"status": "duplicate", "document_id": doc.id}

        # 3. Chunk
        chunks = self.chunker.chunk(doc)

        # 4. Embed
        texts = [c.content for c in chunks]
        embeddings = await self.embedder.embed_batch(texts)
        for chunk, emb in zip(chunks, embeddings):
            chunk.embedding = emb

        # 5. Store
        await self.doc_store.save_document(doc)
        await self.vector_store.upsert_chunks(chunks)
        await self.deduplicator.mark_seen(doc.chunk_hash)

        return {"status": "success", "document_id": doc.id, "chunks": len(chunks)}

    async def ingest_batch(self, file_paths: List[Path]) -> dict:
        """Parallel batch ingestion."""
        semaphore = asyncio.Semaphore(10)

        async def limited_ingest(path):
            async with semaphore:
                return await self.ingest_file(path)

        results = await asyncio.gather(*[limited_ingest(p) for p in file_paths])
        return {
            "total": len(results),
            "successful": sum(1 for r in results if r["status"] == "success"),
            "duplicates": sum(1 for r in results if r["status"] == "duplicate"),
            "details": results
        }

    async def _load_document(self, path: Path, metadata: dict) -> Document:
        # Implement based on file type (PDF, MD, HTML, etc.)
        content = path.read_text(encoding="utf-8")
        chunk_hash = hashlib.sha256(content.encode()).hexdigest()[:16]

        return Document(
            id=hashlib.md5(str(path).encode()).hexdigest()[:12],
            content=content,
            metadata=metadata or {},
            source=str(path),
            created_at=datetime.utcnow().isoformat(),
            chunk_hash=chunk_hash
        )

Semantic Chunking

# ingestion/chunker.py
from dataclasses import dataclass
import re
from typing import List

@dataclass
class Chunk:
    content: str
    metadata: dict
    token_count: int

class SemanticChunker:
    """Chunk by semantic boundaries (headers, paragraphs)."""

    def __init__(self, max_tokens: int = 512, overlap: int = 50):
        self.max_tokens = max_tokens
        self.overlap = overlap

    def chunk(self, doc: Document) -> List[Chunk]:
        # Split by markdown headers
        sections = self._split_by_headers(doc.content)

        all_chunks = []
        for section_content, section_meta in sections:
            section_chunks = self._chunk_section(
                section_content, 
                {**doc.metadata, **section_meta}
            )
            all_chunks.extend(section_chunks)

        return all_chunks

    def _split_by_headers(self, text: str) -> List[tuple[str, dict]]:
        pattern = r'^(#{1,6})\s+(.+)$'
        lines = text.split('\n')

        sections = []
        current_header = {}
        current_content = []

        for line in lines:
            match = re.match(pattern, line)
            if match:
                if current_content:
                    header_path = " > ".join(v for v in current_header.values() if v)
                    sections.append(("\n".join(current_content), {"header_path": header_path}))

                level = len(match.group(1))
                header_text = match.group(2)
                current_header[f"h{level}"] = header_text
                for l in range(level + 1, 7):
                    current_header[f"h{l}"] = ""
                current_content = [line]
            else:
                current_content.append(line)

        if current_content:
            header_path = " > ".join(v for v in current_header.values() if v)
            sections.append(("\n".join(current_content), {"header_path": header_path}))

        return sections

    def _chunk_section(self, content: str, metadata: dict) -> List[Chunk]:
        paragraphs = [p.strip() for p in content.split('\n\n') if p.strip()]

        chunks = []
        current_chunk = []
        current_tokens = 0

        for para in paragraphs:
            para_tokens = self._estimate_tokens(para)

            if current_tokens + para_tokens > self.max_tokens and current_chunk:
                chunk_text = "\n\n".join(current_chunk)
                chunks.append(Chunk(chunk_text, metadata, current_tokens))

                overlap_text = self._get_overlap(current_chunk)
                current_chunk = [overlap_text, para] if overlap_text else [para]
                current_tokens = self._estimate_tokens("\n\n".join(current_chunk))
            else:
                current_chunk.append(para)
                current_tokens += para_tokens

        if current_chunk:
            chunk_text = "\n\n".join(current_chunk)
            chunks.append(Chunk(chunk_text, metadata, current_tokens))

        return chunks

    def _estimate_tokens(self, text: str) -> int:
        return len(text) // 4

    def _get_overlap(self, chunks: List[str]) -> str:
        if not chunks:
            return ""
        last = chunks[-1]
        words = last.split()
        overlap_words = words[-(self.overlap // 4):]
        return " ".join(overlap_words)

Part 2: Hybrid Retrieval Pipeline

# retrieval/pipeline.py
from dataclasses import dataclass
from typing import List, Optional, Callable
import asyncio

@dataclass
class RetrievedDoc:
    content: str
    metadata: dict
    score: float
    rank: int
    source: str  # "dense", "sparse", "hybrid"

class HybridRetrievalPipeline:
    def __init__(
        self,
        dense_embedder: Callable,
        sparse_embedder: Callable,
        vector_store,
        sparse_store,
        reranker: Callable,
        dense_weight: float = 0.7,
        sparse_weight: float = 0.3
    ):
        self.dense_embedder = dense_embedder
        self.sparse_embedder = sparse_embedder
        self.vector_store = vector_store
        self.sparse_store = sparse_store
        self.reranker = reranker
        self.dense_weight = dense_weight
        self.sparse_weight = sparse_weight

    async def retrieve(
        self,
        query: str,
        k: int = 5,
        fetch_k: int = 20,
        filters: dict = None,
        use_reranker: bool = True
    ) -> List[RetrievedDoc]:
        # Stage 1: Parallel dense + sparse retrieval
        dense_task = self._dense_search(query, fetch_k, filters)
        sparse_task = self._sparse_search(query, fetch_k, filters)

        dense_results, sparse_results = await asyncio.gather(dense_task, sparse_task)

        # Stage 2: Fuse results (RRF - Reciprocal Rank Fusion)
        fused = self._reciprocal_rank_fusion(dense_results, sparse_results, k * 2)

        # Stage 3: Rerank
        if use_reranker and fused:
            fused = await self._rerank(query, fused)

        # Stage 4: Final selection
        return fused[:k]

    async def _dense_search(self, query: str, k: int, filters: dict) -> List[RetrievedDoc]:
        query_vector = await self.dense_embedder(query)
        results = await self.vector_store.search(query_vector, k, filters)
        return [
            RetrievedDoc(
                content=r["content"],
                metadata=r["metadata"],
                score=r["score"] * self.dense_weight,
                rank=i,
                source="dense"
            )
            for i, r in enumerate(results)
        ]

    async def _sparse_search(self, query: str, k: int, filters: dict) -> List[RetrievedDoc]:
        sparse_vector = await self.sparse_embedder(query)
        results = await self.sparse_store.search(sparse_vector, k, filters)
        return [
            RetrievedDoc(
                content=r["content"],
                metadata=r["metadata"],
                score=r["score"] * self.sparse_weight,
                rank=i,
                source="sparse"
            )
            for i, r in enumerate(results)
        ]

    def _reciprocal_rank_fusion(
        self, 
        dense: List[RetrievedDoc], 
        sparse: List[RetrievedDoc], 
        k: int,
        k_constant: int = 60
    ) -> List[RetrievedDoc]:
        """RRF fusion for hybrid search."""
        # Combine by content hash
        doc_scores = {}

        for results, weight in [(dense, self.dense_weight), (sparse, self.sparse_weight)]:
            for rank, doc in enumerate(results):
                content_hash = hash(doc.content[:200])
                if content_hash not in doc_scores:
                    doc_scores[content_hash] = {
                        "doc": doc,
                        "rrf_score": 0
                    }
                # RRF score: 1 / (k + rank)
                doc_scores[content_hash]["rrf_score"] += weight / (k_constant + rank + 1)

        # Sort by RRF score
        fused = sorted(
            doc_scores.values(), 
            key=lambda x: x["rrf_score"], 
            reverse=True
        )

        return [
            RetrievedDoc(
                content=item["doc"].content,
                metadata=item["doc"].metadata,
                score=item["rrf_score"],
                rank=i,
                source="hybrid"
            )
            for i, item in enumerate(fused[:k])
        ]

    async def _rerank(self, query: str, docs: List[RetrievedDoc]) -> List[RetrievedDoc]:
        contents = [d.content for d in docs]
        rerank_scores = await self.reranker(query, contents)

        for doc, score in zip(docs, rerank_scores):
            doc.score = score
            doc.source = "reranked"

        docs.sort(key=lambda x: x.score, reverse=True)
        return docs

Part 3: Generation with Citations & Guardrails

# generation/rag_chain.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field
from typing import List, Optional

class CitedAnswer(BaseModel):
    answer: str = Field(description="Complete answer with inline citations [doc_id]")
    citations: List[dict] = Field(description="Citations with doc_id, snippet, metadata")
    confidence: float = Field(ge=0, le=1, description="Confidence 0-1")
    needs_clarification: bool = Field(default=False)

RAG_PROMPT = """You are a precise assistant that answers questions using ONLY the provided context.

## Context Documents
{context}

## Question
{question}

## Instructions
1. Answer ONLY from the provided context
2. If context is insufficient, say "I don't have enough information to answer this"
3. Cite sources inline using [doc_id] format
4. Be concise but complete
5. Include confidence score (0-1) based on context coverage

## Output Format (JSON)
{{
    "answer": "Your answer with [doc_id] citations",
    "citations": [
        {{"doc_id": "abc123", "snippet": "relevant excerpt", "metadata": {{...}}}}
    ],
    "confidence": 0.9,
    "needs_clarification": false
}}"""

def build_rag_chain(model_name: str = "gpt-4o"):
    llm = ChatOpenAI(model=model_name, temperature=0)
    prompt = ChatPromptTemplate.from_template(RAG_PROMPT)
    parser = JsonOutputParser(pydantic_object=CitedAnswer)
    return prompt | llm | parser

def format_context(docs: List[RetrievedDoc]) -> str:
    formatted = []
    for i, doc in enumerate(docs):
        doc_id = doc.metadata.get("chunk_hash", f"doc_{i}")
        formatted.append(
            f"[Document {doc_id}]\n"
            f"Source: {doc.metadata.get('doc_title', 'Unknown')}\n"
            f"Section: {doc.metadata.get('header_path', 'N/A')}\n"
            f"Content: {doc.content}\n"
        )
    return "\n---\n".join(formatted)

Guardrails

# generation/guardrails.py
import re
from typing import Tuple

class RAGGuardrails:
    def __init__(self):
        self.pii_patterns = [
            (r'\b\d{3}-\d{2}-\d{4}\b', 'SSN'),
            (r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', 'EMAIL'),
            (r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', 'CREDIT_CARD'),
            (r'\b\d{10}\b', 'PHONE'),
        ]

        self.refusal_triggers = [
            "ignore previous instructions",
            "system prompt",
            "reveal your prompt",
            "bypass",
            "jailbreak",
        ]

    def check_input(self, query: str) -> Tuple[bool, str]:
        query_lower = query.lower()

        for trigger in self.refusal_triggers:
            if trigger in query_lower:
                return False, "I can't help with that request."

        for pattern, pii_type in self.pii_patterns:
            if re.search(pattern, query):
                return False, f"Please don't share {pii_type} in your question."

        return True, ""

    def check_output(self, answer: str, context: str) -> Tuple[bool, str]:
        # Hallucination check: numbers in answer must exist in context
        answer_numbers = re.findall(r'\b\d+\.?\d*\b', answer)
        context_numbers = set(re.findall(r'\b\d+\.?\d*\b', context))

        for num in answer_numbers:
            if num not in context_numbers and len(num) > 2:
                return False, f"Unverified claim detected: {num}"

        # PII leakage check
        for pattern, pii_type in self.pii_patterns:
            if re.search(pattern, answer):
                return False, f"Potential {pii_type} leakage in response."

        return True, ""

    def sanitize_output(self, answer: str) -> str:
        sanitized = answer
        for pattern, _ in self.pii_patterns:
            sanitized = re.sub(pattern, "[REDACTED]", sanitized)
        return sanitized

Part 4: LangGraph Orchestration

# orchestration/rag_graph.py
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
from typing import TypedDict, Annotated, List
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class RAGState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    question: str
    retrieved_docs: List[RetrievedDoc]
    answer: CitedAnswer
    guardrail_passed: bool
    retry_count: int

async def retrieve_node(state: RAGState):
    pipeline = get_retrieval_pipeline()
    docs = await pipeline.retrieve(state["question"])
    return {"retrieved_docs": docs}

async def generate_node(state: RAGState):
    chain = get_rag_chain()
    context = format_context(state["retrieved_docs"])
    answer = await chain.ainvoke({"context": context, "question": state["question"]})
    return {"answer": answer}

async def guardrail_node(state: RAGState):
    guardrails = get_guardrails()
    context = format_context(state["retrieved_docs"])
    passed, reason = guardrails.check_output(state["answer"].answer, context)

    if not passed and state["retry_count"] < 2:
        return {
            "guardrail_passed": False,
            "retry_count": state["retry_count"] + 1,
            "messages": [HumanMessage(content=f"Regenerate: {reason}")]
        }

    return {"guardrail_passed": passed}

def route_after_guardrail(state: RAGState) -> str:
    if state["guardrail_passed"]:
        return END
    return "generate"

# Build graph
workflow = StateGraph(RAGState)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("generate", generate_node)
workflow.add_node("guardrail", guardrail_node)

workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "guardrail")
workflow.add_conditional_edges("guardrail", route_after_guardrail, {
    "generate": "generate",
    END: END
})

rag_app = workflow.compile(checkpointer=SqliteSaver.from_conn_string("sqlite:///rag.db"))

Part 5: FastAPI Production Server

# server.py
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from typing import Optional
import uvicorn

app = FastAPI(title="Production RAG API", version="1.0.0")

class QueryRequest(BaseModel):
    question: str
    filters: Optional[dict] = None
    k: int = 5
    use_reranker: bool = True
    stream: bool = False

class QueryResponse(BaseModel):
    answer: str
    citations: List[dict]
    confidence: float
    retrieved_count: int
    latency_ms: float

@app.post("/query", response_model=QueryResponse)
async def query(request: QueryRequest):
    start = time.time()

    # Input guardrails
    guardrails = get_guardrails()
    passed, reason = guardrails.check_input(request.question)
    if not passed:
        raise HTTPException(400, reason)

    # Execute RAG
    config = {"configurable": {"thread_id": f"query-{uuid.uuid4()}"}}
    result = await rag_app.ainvoke({
        "question": request.question,
        "retry_count": 0
    }, config)

    latency = (time.time() - start) * 1000

    # Log for monitoring
    background_tasks.add_task(log_metrics, {
        "question": request.question,
        "latency_ms": latency,
        "retrieved": len(result["retrieved_docs"]),
        "confidence": result["answer"].confidence,
        "guardrail_passed": result["guardrail_passed"]
    })

    return QueryResponse(
        answer=result["answer"].answer,
        citations=result["answer"].citations,
        confidence=result["answer"].confidence,
        retrieved_count=len(result["retrieved_docs"]),
        latency_ms=latency
    )

@app.post("/ingest")
async def ingest(request: IngestRequest, background_tasks: BackgroundTasks):
    background_tasks.add_task(run_ingestion, request.paths)
    return {"status": "ingestion_started"}

@app.get("/health")
async def health():
    return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()}

@app.get("/metrics")
async def metrics():
    return get_system_metrics()

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Part 6: Evaluation & Monitoring

# evaluation/rag_eval.py
from dataclasses import dataclass
from typing import List
import json

@dataclass
class RAGTestCase:
    question: str
    expected_answer: str
    relevant_doc_ids: List[str]
    metadata_filters: dict = None

class RAGEvaluator:
    def __init__(self, rag_system):
        self.rag = rag_system

    def evaluate_retrieval(self, test_cases: List[RAGTestCase]) -> dict:
        metrics = {"precision@k": [], "recall@k": [], "mrr": []}

        for tc in test_cases:
            results = await self.rag.retrieve(tc.question, k=10)
            retrieved_ids = [r.metadata.get("chunk_hash") for r in results]
            relevant = set(tc.relevant_doc_ids)

            for k in [1, 3, 5, 10]:
                retrieved_k = set(retrieved_ids[:k])
                metrics[f"precision@{k}"].append(len(retrieved_k & relevant) / k)
                metrics[f"recall@{k}"].append(len(retrieved_k & relevant) / len(relevant))

            for i, rid in enumerate(retrieved_ids):
                if rid in relevant:
                    metrics["mrr"].append(1 / (i + 1))
                    break
            else:
                metrics["mrr"].append(0)

        return {k: sum(v) / len(v) for k, v in metrics.items()}

    async def evaluate_generation(self, test_cases: List[RAGTestCase]) -> dict:
        scores = {"accuracy": [], "completeness": [], "citation_quality": []}

        for tc in test_cases:
            result = await self.rag.query(tc.question)
            eval_result = await llm_judge(
                task="Answer question using provided context",
                input_text=tc.question,
                response=result.answer,
                criteria=["accuracy", "completeness", "citation_quality", "faithfulness"]
            )

            for k in scores:
                scores[k].append(eval_result.scores.get(k, 0))

        return {k: sum(v) / len(v) for k, v in scores.items()}

Production Monitoring

# monitoring/production_monitor.py
from dataclasses import dataclass, asdict
from datetime import datetime
import json

@dataclass
class RAGMetrics:
    timestamp: str
    question_hash: str
    latency_ms: float
    retrieval_latency_ms: float
    generation_latency_ms: float
    num_retrieved: int
    num_cited: int
    confidence: float
    user_feedback: Optional[str] = None
    error: Optional[str] = None

class ProductionMonitor:
    def __init__(self, log_path: str = "logs/rag_metrics.jsonl"):
        self.log_path = log_path
        self.buffer = []

    def log(self, metrics: RAGMetrics):
        self.buffer.append(asdict(metrics))
        if len(self.buffer) >= 100:
            self.flush()

    def flush(self):
        with open(self.log_path, "a") as f:
            for m in self.buffer:
                f.write(json.dumps(m) + "\n")
        self.buffer.clear()

    def get_stats(self, hours: int = 24) -> dict:
        import pandas as pd
        df = pd.read_json(self.log_path, lines=True)
        df["timestamp"] = pd.to_datetime(df["timestamp"])
        cutoff = pd.Timestamp.utcnow() - pd.Timedelta(hours=hours)
        df = df[df["timestamp"] > cutoff]

        return {
            "total_queries": len(df),
            "avg_latency_ms": df["latency_ms"].mean(),
            "p95_latency_ms": df["latency_ms"].quantile(0.95),
            "avg_confidence": df["confidence"].mean(),
            "error_rate": df["error"].notna().mean(),
            "avg_retrieved": df["num_retrieved"].mean(),
            "feedback_positive_rate": (
                df["user_feedback"].eq("positive").mean() 
                if "user_feedback" in df.columns else 0
            )
        }

Part 7: Deployment

Docker

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# System dependencies
RUN apt-get update && apt-get install -y \
    gcc \
    libpq-dev \
    && rm -rf /var/lib/apt/lists/*

# Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Application
COPY . .

# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000"]

Kubernetes

# k8s/rag-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: rag-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: rag-api
  template:
    spec:
      containers:
      - name: rag-api
        image: your-registry/rag-api:v1.0
        ports:
        - containerPort: 8000
        env:
        - name: QDRANT_URL
          value: "http://qdrant:6333"
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: rag-secrets
              key: openai-key
        - name: REDIS_URL
          value: "redis://redis:6379"
        resources:
          requests:
            memory: "2Gi"
            cpu: "1000m"
          limits:
            memory: "4Gi"
            cpu: "2000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 10
---
apiVersion: v1
kind: Service
metadata:
  name: rag-api
spec:
  selector:
    app: rag-api
  ports:
  - port: 80
    targetPort: 8000

Complete Requirements

# requirements.txt
langchain==0.2.0
langgraph==0.1.0
langchain-openai==0.1.0
qdrant-client==1.10.0
sentence-transformers==3.0.0
fastapi==0.110.0
uvicorn==0.29.0
pydantic==2.7.0
pydantic-settings==2.3.0
python-multipart==0.0.9
redis==5.0.0
psycopg2-binary==2.9.9
pytest==8.2.0
pytest-asyncio==0.23.0

Production Checklist

Ingestion

[ ] Semantic chunking with header preservation
[ ] Metadata enrichment (source, date, department, confidentiality)
[ ] Deduplication (exact + near-duplicate via MinHash)
[ ] Incremental updates (upsert by chunk hash)
[ ] Dead letter queue for failed documents
[ ] Progress tracking & resumability

Retrieval

[ ] Hybrid search (dense + sparse/BM25)
[ ] Cross-encoder reranking (ms-marco-MiniLM-L-6-v2)
[ ] MMR for diversity
[ ] Query rewriting/expansion
[ ] Metadata filtering (ACL, date range, department)
[ ] Retrieval latency < 200ms p95

Generation

[ ] Structured output with citations (JSON schema)
[ ] Input/output guardrails (PII, injection, hallucination)
[ ] Confidence scoring
[ ] Fallback for low-confidence answers
[ ] Generation latency < 2s p95

Evaluation

[ ] Golden test set (100+ cases)
[ ] Retrieval metrics (Precision@K, Recall@K, MRR)
[ ] Generation metrics (Accuracy, Faithfulness, Citation quality)
[ ] Regression testing in CI/CD
[ ] A/B testing framework

Monitoring

[ ] Latency percentiles (p50, p95, p99)
[ ] Error rates by type
[ ] Confidence distribution
[ ] User feedback collection (explicit + implicit)
[ ] Automated alerts (quality drop, latency spike, cost surge)
[ ] Cost tracking per query

Operations

[ ] Blue-green deployments
[ ] Rollback capability (< 5 min)
[ ] Vector index rebuild procedure
[ ] Embedding model versioning
[ ] Data retention policy
[ ] Backup/restore tested monthly

Resources

Production RAG is 10% retrieval, 90% engineering. Build observability, guardrails, and evaluation in from day one — retrofitting is 10x harder.

Production RAG with LangChain & LangGraph: End-to-End Implementation

Phil IPS — Mon, 12 Oct 2026 18:02:35 +0000

Production RAG with LangChain & LangGraph: End-to-End Implementation

Build a production-ready RAG system with document ingestion, hybrid retrieval, reranking, citations, guardrails, and monitoring.

Introduction

RAG (Retrieval-Augmented Generation) is the standard for grounding LLMs in proprietary data. This guide builds a production system — not a prototype.

Production RAG Requirements

Requirement	Prototype	Production
Ingestion	Manual, one-off	Automated, incremental, monitored
Retrieval	Vector only	Hybrid (dense + sparse + rerank)
Generation	Basic prompt	Structured output + citations + guardrails
Evaluation	Manual	Automated metrics + LLM judge + A/B
Monitoring	None	Latency, quality, cost, drift alerts
Deployment	Local notebook	Containerized, scalable, resilient

System Architecture

┌─────────────┐     ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  Ingest     │────▶│  Embedding  │────▶│  Retrieval  │────▶│ Generation  │
│  Pipeline   │     │  & Index    │     │  Pipeline   │     │  + Guardrails
└─────────────┘     └─────────────┘     └─────────────┘     └─────────────┘
       │                   │                   │                   │
       ▼                   ▼                   ▼                   ▼
  Chunking            Vector DB           Hybrid Search      Structured Output
  Metadata            (Qdrant)            (Dense+Sparse)     Citations
  Deduplication       Quantization        Reranking          Guardrails
  Incremental         Replication         Filtering          Evaluation

Part 1: Document Ingestion Pipeline

# ingestion/pipeline.py
from dataclasses import dataclass
from typing import List, Optional
from pathlib import Path
import hashlib
import asyncio
from datetime import datetime

@dataclass
class Document:
    id: str
    content: str
    metadata: dict
    source: str
    created_at: str
    chunk_hash: str = ""

@dataclass
class Chunk:
    id: str
    document_id: str
    content: str
    metadata: dict
    token_count: int
    embedding: List[float] = None

class IngestionPipeline:
    def __init__(
        self,
        chunker,
        embedder,
        vector_store,
        doc_store,
        deduplicator
    ):
        self.chunker = chunker
        self.embedder = embedder
        self.vector_store = vector_store
        self.doc_store = doc_store
        self.deduplicator = deduplicator

    async def ingest_file(self, file_path: Path, metadata: dict = None) -> dict:
        """Ingest a single file."""
        # 1. Load document
        doc = await self._load_document(file_path, metadata)

        # 2. Check for duplicates
        if await self.deduplicator.is_duplicate(doc.chunk_hash):
            return {"status": "duplicate", "document_id": doc.id}

        # 3. Chunk
        chunks = self.chunker.chunk(doc)

        # 4. Embed
        texts = [c.content for c in chunks]
        embeddings = await self.embedder.embed_batch(texts)
        for chunk, emb in zip(chunks, embeddings):
            chunk.embedding = emb

        # 5. Store
        await self.doc_store.save_document(doc)
        await self.vector_store.upsert_chunks(chunks)
        await self.deduplicator.mark_seen(doc.chunk_hash)

        return {"status": "success", "document_id": doc.id, "chunks": len(chunks)}

    async def ingest_batch(self, file_paths: List[Path]) -> dict:
        """Parallel batch ingestion."""
        semaphore = asyncio.Semaphore(10)

        async def limited_ingest(path):
            async with semaphore:
                return await self.ingest_file(path)

        results = await asyncio.gather(*[limited_ingest(p) for p in file_paths])
        return {
            "total": len(results),
            "successful": sum(1 for r in results if r["status"] == "success"),
            "duplicates": sum(1 for r in results if r["status"] == "duplicate"),
            "details": results
        }

    async def _load_document(self, path: Path, metadata: dict) -> Document:
        # Implement based on file type (PDF, MD, HTML, etc.)
        content = path.read_text(encoding="utf-8")
        chunk_hash = hashlib.sha256(content.encode()).hexdigest()[:16]

        return Document(
            id=hashlib.md5(str(path).encode()).hexdigest()[:12],
            content=content,
            metadata=metadata or {},
            source=str(path),
            created_at=datetime.utcnow().isoformat(),
            chunk_hash=chunk_hash
        )

Semantic Chunking

# ingestion/chunker.py
from dataclasses import dataclass
import re
from typing import List

@dataclass
class Chunk:
    content: str
    metadata: dict
    token_count: int

class SemanticChunker:
    """Chunk by semantic boundaries (headers, paragraphs)."""

    def __init__(self, max_tokens: int = 512, overlap: int = 50):
        self.max_tokens = max_tokens
        self.overlap = overlap

    def chunk(self, doc: Document) -> List[Chunk]:
        # Split by markdown headers
        sections = self._split_by_headers(doc.content)

        all_chunks = []
        for section_content, section_meta in sections:
            section_chunks = self._chunk_section(
                section_content, 
                {**doc.metadata, **section_meta}
            )
            all_chunks.extend(section_chunks)

        return all_chunks

    def _split_by_headers(self, text: str) -> List[tuple[str, dict]]:
        pattern = r'^(#{1,6})\s+(.+)$'
        lines = text.split('\n')

        sections = []
        current_header = {}
        current_content = []

        for line in lines:
            match = re.match(pattern, line)
            if match:
                if current_content:
                    header_path = " > ".join(v for v in current_header.values() if v)
                    sections.append(("\n".join(current_content), {"header_path": header_path}))

                level = len(match.group(1))
                header_text = match.group(2)
                current_header[f"h{level}"] = header_text
                for l in range(level + 1, 7):
                    current_header[f"h{l}"] = ""
                current_content = [line]
            else:
                current_content.append(line)

        if current_content:
            header_path = " > ".join(v for v in current_header.values() if v)
            sections.append(("\n".join(current_content), {"header_path": header_path}))

        return sections

    def _chunk_section(self, content: str, metadata: dict) -> List[Chunk]:
        paragraphs = [p.strip() for p in content.split('\n\n') if p.strip()]

        chunks = []
        current_chunk = []
        current_tokens = 0

        for para in paragraphs:
            para_tokens = self._estimate_tokens(para)

            if current_tokens + para_tokens > self.max_tokens and current_chunk:
                chunk_text = "\n\n".join(current_chunk)
                chunks.append(Chunk(chunk_text, metadata, current_tokens))

                overlap_text = self._get_overlap(current_chunk)
                current_chunk = [overlap_text, para] if overlap_text else [para]
                current_tokens = self._estimate_tokens("\n\n".join(current_chunk))
            else:
                current_chunk.append(para)
                current_tokens += para_tokens

        if current_chunk:
            chunk_text = "\n\n".join(current_chunk)
            chunks.append(Chunk(chunk_text, metadata, current_tokens))

        return chunks

    def _estimate_tokens(self, text: str) -> int:
        return len(text) // 4

    def _get_overlap(self, chunks: List[str]) -> str:
        if not chunks:
            return ""
        last = chunks[-1]
        words = last.split()
        overlap_words = words[-(self.overlap // 4):]
        return " ".join(overlap_words)

Part 2: Hybrid Retrieval Pipeline

# retrieval/pipeline.py
from dataclasses import dataclass
from typing import List, Optional, Callable
import asyncio

@dataclass
class RetrievedDoc:
    content: str
    metadata: dict
    score: float
    rank: int
    source: str  # "dense", "sparse", "hybrid"

class HybridRetrievalPipeline:
    def __init__(
        self,
        dense_embedder: Callable,
        sparse_embedder: Callable,
        vector_store,
        sparse_store,
        reranker: Callable,
        dense_weight: float = 0.7,
        sparse_weight: float = 0.3
    ):
        self.dense_embedder = dense_embedder
        self.sparse_embedder = sparse_embedder
        self.vector_store = vector_store
        self.sparse_store = sparse_store
        self.reranker = reranker
        self.dense_weight = dense_weight
        self.sparse_weight = sparse_weight

    async def retrieve(
        self,
        query: str,
        k: int = 5,
        fetch_k: int = 20,
        filters: dict = None,
        use_reranker: bool = True
    ) -> List[RetrievedDoc]:
        # Stage 1: Parallel dense + sparse retrieval
        dense_task = self._dense_search(query, fetch_k, filters)
        sparse_task = self._sparse_search(query, fetch_k, filters)

        dense_results, sparse_results = await asyncio.gather(dense_task, sparse_task)

        # Stage 2: Fuse results (RRF - Reciprocal Rank Fusion)
        fused = self._reciprocal_rank_fusion(dense_results, sparse_results, k * 2)

        # Stage 3: Rerank
        if use_reranker and fused:
            fused = await self._rerank(query, fused)

        # Stage 4: Final selection
        return fused[:k]

    async def _dense_search(self, query: str, k: int, filters: dict) -> List[RetrievedDoc]:
        query_vector = await self.dense_embedder(query)
        results = await self.vector_store.search(query_vector, k, filters)
        return [
            RetrievedDoc(
                content=r["content"],
                metadata=r["metadata"],
                score=r["score"] * self.dense_weight,
                rank=i,
                source="dense"
            )
            for i, r in enumerate(results)
        ]

    async def _sparse_search(self, query: str, k: int, filters: dict) -> List[RetrievedDoc]:
        sparse_vector = await self.sparse_embedder(query)
        results = await self.sparse_store.search(sparse_vector, k, filters)
        return [
            RetrievedDoc(
                content=r["content"],
                metadata=r["metadata"],
                score=r["score"] * self.sparse_weight,
                rank=i,
                source="sparse"
            )
            for i, r in enumerate(results)
        ]

    def _reciprocal_rank_fusion(
        self, 
        dense: List[RetrievedDoc], 
        sparse: List[RetrievedDoc], 
        k: int,
        k_constant: int = 60
    ) -> List[RetrievedDoc]:
        """RRF fusion for hybrid search."""
        # Combine by content hash
        doc_scores = {}

        for results, weight in [(dense, self.dense_weight), (sparse, self.sparse_weight)]:
            for rank, doc in enumerate(results):
                content_hash = hash(doc.content[:200])
                if content_hash not in doc_scores:
                    doc_scores[content_hash] = {
                        "doc": doc,
                        "rrf_score": 0
                    }
                # RRF score: 1 / (k + rank)
                doc_scores[content_hash]["rrf_score"] += weight / (k_constant + rank + 1)

        # Sort by RRF score
        fused = sorted(
            doc_scores.values(), 
            key=lambda x: x["rrf_score"], 
            reverse=True
        )

        return [
            RetrievedDoc(
                content=item["doc"].content,
                metadata=item["doc"].metadata,
                score=item["rrf_score"],
                rank=i,
                source="hybrid"
            )
            for i, item in enumerate(fused[:k])
        ]

    async def _rerank(self, query: str, docs: List[RetrievedDoc]) -> List[RetrievedDoc]:
        contents = [d.content for d in docs]
        rerank_scores = await self.reranker(query, contents)

        for doc, score in zip(docs, rerank_scores):
            doc.score = score
            doc.source = "reranked"

        docs.sort(key=lambda x: x.score, reverse=True)
        return docs

Part 3: Generation with Citations & Guardrails

# generation/rag_chain.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field
from typing import List, Optional

class CitedAnswer(BaseModel):
    answer: str = Field(description="Complete answer with inline citations [doc_id]")
    citations: List[dict] = Field(description="Citations with doc_id, snippet, metadata")
    confidence: float = Field(ge=0, le=1, description="Confidence 0-1")
    needs_clarification: bool = Field(default=False)

RAG_PROMPT = """You are a precise assistant that answers questions using ONLY the provided context.

## Context Documents
{context}

## Question
{question}

## Instructions
1. Answer ONLY from the provided context
2. If context is insufficient, say "I don't have enough information to answer this"
3. Cite sources inline using [doc_id] format
4. Be concise but complete
5. Include confidence score (0-1) based on context coverage

## Output Format (JSON)
{{
    "answer": "Your answer with [doc_id] citations",
    "citations": [
        {{"doc_id": "abc123", "snippet": "relevant excerpt", "metadata": {{...}}}}
    ],
    "confidence": 0.9,
    "needs_clarification": false
}}"""

def build_rag_chain(model_name: str = "gpt-4o"):
    llm = ChatOpenAI(model=model_name, temperature=0)
    prompt = ChatPromptTemplate.from_template(RAG_PROMPT)
    parser = JsonOutputParser(pydantic_object=CitedAnswer)
    return prompt | llm | parser

def format_context(docs: List[RetrievedDoc]) -> str:
    formatted = []
    for i, doc in enumerate(docs):
        doc_id = doc.metadata.get("chunk_hash", f"doc_{i}")
        formatted.append(
            f"[Document {doc_id}]\n"
            f"Source: {doc.metadata.get('doc_title', 'Unknown')}\n"
            f"Section: {doc.metadata.get('header_path', 'N/A')}\n"
            f"Content: {doc.content}\n"
        )
    return "\n---\n".join(formatted)

Guardrails

# generation/guardrails.py
import re
from typing import Tuple

class RAGGuardrails:
    def __init__(self):
        self.pii_patterns = [
            (r'\b\d{3}-\d{2}-\d{4}\b', 'SSN'),
            (r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', 'EMAIL'),
            (r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', 'CREDIT_CARD'),
            (r'\b\d{10}\b', 'PHONE'),
        ]

        self.refusal_triggers = [
            "ignore previous instructions",
            "system prompt",
            "reveal your prompt",
            "bypass",
            "jailbreak",
        ]

    def check_input(self, query: str) -> Tuple[bool, str]:
        query_lower = query.lower()

        for trigger in self.refusal_triggers:
            if trigger in query_lower:
                return False, "I can't help with that request."

        for pattern, pii_type in self.pii_patterns:
            if re.search(pattern, query):
                return False, f"Please don't share {pii_type} in your question."

        return True, ""

    def check_output(self, answer: str, context: str) -> Tuple[bool, str]:
        # Hallucination check: numbers in answer must exist in context
        answer_numbers = re.findall(r'\b\d+\.?\d*\b', answer)
        context_numbers = set(re.findall(r'\b\d+\.?\d*\b', context))

        for num in answer_numbers:
            if num not in context_numbers and len(num) > 2:
                return False, f"Unverified claim detected: {num}"

        # PII leakage check
        for pattern, pii_type in self.pii_patterns:
            if re.search(pattern, answer):
                return False, f"Potential {pii_type} leakage in response."

        return True, ""

    def sanitize_output(self, answer: str) -> str:
        sanitized = answer
        for pattern, _ in self.pii_patterns:
            sanitized = re.sub(pattern, "[REDACTED]", sanitized)
        return sanitized

Part 4: LangGraph Orchestration

# orchestration/rag_graph.py
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
from typing import TypedDict, Annotated, List
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class RAGState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    question: str
    retrieved_docs: List[RetrievedDoc]
    answer: CitedAnswer
    guardrail_passed: bool
    retry_count: int

async def retrieve_node(state: RAGState):
    pipeline = get_retrieval_pipeline()
    docs = await pipeline.retrieve(state["question"])
    return {"retrieved_docs": docs}

async def generate_node(state: RAGState):
    chain = get_rag_chain()
    context = format_context(state["retrieved_docs"])
    answer = await chain.ainvoke({"context": context, "question": state["question"]})
    return {"answer": answer}

async def guardrail_node(state: RAGState):
    guardrails = get_guardrails()
    context = format_context(state["retrieved_docs"])
    passed, reason = guardrails.check_output(state["answer"].answer, context)

    if not passed and state["retry_count"] < 2:
        return {
            "guardrail_passed": False,
            "retry_count": state["retry_count"] + 1,
            "messages": [HumanMessage(content=f"Regenerate: {reason}")]
        }

    return {"guardrail_passed": passed}

def route_after_guardrail(state: RAGState) -> str:
    if state["guardrail_passed"]:
        return END
    return "generate"

# Build graph
workflow = StateGraph(RAGState)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("generate", generate_node)
workflow.add_node("guardrail", guardrail_node)

workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "guardrail")
workflow.add_conditional_edges("guardrail", route_after_guardrail, {
    "generate": "generate",
    END: END
})

rag_app = workflow.compile(checkpointer=SqliteSaver.from_conn_string("sqlite:///rag.db"))

Part 5: FastAPI Production Server

# server.py
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from typing import Optional
import uvicorn

app = FastAPI(title="Production RAG API", version="1.0.0")

class QueryRequest(BaseModel):
    question: str
    filters: Optional[dict] = None
    k: int = 5
    use_reranker: bool = True
    stream: bool = False

class QueryResponse(BaseModel):
    answer: str
    citations: List[dict]
    confidence: float
    retrieved_count: int
    latency_ms: float

@app.post("/query", response_model=QueryResponse)
async def query(request: QueryRequest):
    start = time.time()

    # Input guardrails
    guardrails = get_guardrails()
    passed, reason = guardrails.check_input(request.question)
    if not passed:
        raise HTTPException(400, reason)

    # Execute RAG
    config = {"configurable": {"thread_id": f"query-{uuid.uuid4()}"}}
    result = await rag_app.ainvoke({
        "question": request.question,
        "retry_count": 0
    }, config)

    latency = (time.time() - start) * 1000

    # Log for monitoring
    background_tasks.add_task(log_metrics, {
        "question": request.question,
        "latency_ms": latency,
        "retrieved": len(result["retrieved_docs"]),
        "confidence": result["answer"].confidence,
        "guardrail_passed": result["guardrail_passed"]
    })

    return QueryResponse(
        answer=result["answer"].answer,
        citations=result["answer"].citations,
        confidence=result["answer"].confidence,
        retrieved_count=len(result["retrieved_docs"]),
        latency_ms=latency
    )

@app.post("/ingest")
async def ingest(request: IngestRequest, background_tasks: BackgroundTasks):
    background_tasks.add_task(run_ingestion, request.paths)
    return {"status": "ingestion_started"}

@app.get("/health")
async def health():
    return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()}

@app.get("/metrics")
async def metrics():
    return get_system_metrics()

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Part 6: Evaluation & Monitoring

# evaluation/rag_eval.py
from dataclasses import dataclass
from typing import List
import json

@dataclass
class RAGTestCase:
    question: str
    expected_answer: str
    relevant_doc_ids: List[str]
    metadata_filters: dict = None

class RAGEvaluator:
    def __init__(self, rag_system):
        self.rag = rag_system

    def evaluate_retrieval(self, test_cases: List[RAGTestCase]) -> dict:
        metrics = {"precision@k": [], "recall@k": [], "mrr": []}

        for tc in test_cases:
            results = await self.rag.retrieve(tc.question, k=10)
            retrieved_ids = [r.metadata.get("chunk_hash") for r in results]
            relevant = set(tc.relevant_doc_ids)

            for k in [1, 3, 5, 10]:
                retrieved_k = set(retrieved_ids[:k])
                metrics[f"precision@{k}"].append(len(retrieved_k & relevant) / k)
                metrics[f"recall@{k}"].append(len(retrieved_k & relevant) / len(relevant))

            for i, rid in enumerate(retrieved_ids):
                if rid in relevant:
                    metrics["mrr"].append(1 / (i + 1))
                    break
            else:
                metrics["mrr"].append(0)

        return {k: sum(v) / len(v) for k, v in metrics.items()}

    async def evaluate_generation(self, test_cases: List[RAGTestCase]) -> dict:
        scores = {"accuracy": [], "completeness": [], "citation_quality": []}

        for tc in test_cases:
            result = await self.rag.query(tc.question)
            eval_result = await llm_judge(
                task="Answer question using provided context",
                input_text=tc.question,
                response=result.answer,
                criteria=["accuracy", "completeness", "citation_quality", "faithfulness"]
            )

            for k in scores:
                scores[k].append(eval_result.scores.get(k, 0))

        return {k: sum(v) / len(v) for k, v in scores.items()}

Production Monitoring

# monitoring/production_monitor.py
from dataclasses import dataclass, asdict
from datetime import datetime
import json

@dataclass
class RAGMetrics:
    timestamp: str
    question_hash: str
    latency_ms: float
    retrieval_latency_ms: float
    generation_latency_ms: float
    num_retrieved: int
    num_cited: int
    confidence: float
    user_feedback: Optional[str] = None
    error: Optional[str] = None

class ProductionMonitor:
    def __init__(self, log_path: str = "logs/rag_metrics.jsonl"):
        self.log_path = log_path
        self.buffer = []

    def log(self, metrics: RAGMetrics):
        self.buffer.append(asdict(metrics))
        if len(self.buffer) >= 100:
            self.flush()

    def flush(self):
        with open(self.log_path, "a") as f:
            for m in self.buffer:
                f.write(json.dumps(m) + "\n")
        self.buffer.clear()

    def get_stats(self, hours: int = 24) -> dict:
        import pandas as pd
        df = pd.read_json(self.log_path, lines=True)
        df["timestamp"] = pd.to_datetime(df["timestamp"])
        cutoff = pd.Timestamp.utcnow() - pd.Timedelta(hours=hours)
        df = df[df["timestamp"] > cutoff]

        return {
            "total_queries": len(df),
            "avg_latency_ms": df["latency_ms"].mean(),
            "p95_latency_ms": df["latency_ms"].quantile(0.95),
            "avg_confidence": df["confidence"].mean(),
            "error_rate": df["error"].notna().mean(),
            "avg_retrieved": df["num_retrieved"].mean(),
            "feedback_positive_rate": (
                df["user_feedback"].eq("positive").mean() 
                if "user_feedback" in df.columns else 0
            )
        }

Part 7: Deployment

Docker

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# System dependencies
RUN apt-get update && apt-get install -y \
    gcc \
    libpq-dev \
    && rm -rf /var/lib/apt/lists/*

# Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Application
COPY . .

# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000"]

Kubernetes

# k8s/rag-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: rag-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: rag-api
  template:
    spec:
      containers:
      - name: rag-api
        image: your-registry/rag-api:v1.0
        ports:
        - containerPort: 8000
        env:
        - name: QDRANT_URL
          value: "http://qdrant:6333"
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: rag-secrets
              key: openai-key
        - name: REDIS_URL
          value: "redis://redis:6379"
        resources:
          requests:
            memory: "2Gi"
            cpu: "1000m"
          limits:
            memory: "4Gi"
            cpu: "2000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 10
---
apiVersion: v1
kind: Service
metadata:
  name: rag-api
spec:
  selector:
    app: rag-api
  ports:
  - port: 80
    targetPort: 8000

Complete Requirements

# requirements.txt
langchain==0.2.0
langgraph==0.1.0
langchain-openai==0.1.0
qdrant-client==1.10.0
sentence-transformers==3.0.0
fastapi==0.110.0
uvicorn==0.29.0
pydantic==2.7.0
pydantic-settings==2.3.0
python-multipart==0.0.9
redis==5.0.0
psycopg2-binary==2.9.9
pytest==8.2.0
pytest-asyncio==0.23.0

Production Checklist

Ingestion

[ ] Semantic chunking with header preservation
[ ] Metadata enrichment (source, date, department, confidentiality)
[ ] Deduplication (exact + near-duplicate via MinHash)
[ ] Incremental updates (upsert by chunk hash)
[ ] Dead letter queue for failed documents
[ ] Progress tracking & resumability

Retrieval

[ ] Hybrid search (dense + sparse/BM25)
[ ] Cross-encoder reranking (ms-marco-MiniLM-L-6-v2)
[ ] MMR for diversity
[ ] Query rewriting/expansion
[ ] Metadata filtering (ACL, date range, department)
[ ] Retrieval latency < 200ms p95

Generation

[ ] Structured output with citations (JSON schema)
[ ] Input/output guardrails (PII, injection, hallucination)
[ ] Confidence scoring
[ ] Fallback for low-confidence answers
[ ] Generation latency < 2s p95

Evaluation

[ ] Golden test set (100+ cases)
[ ] Retrieval metrics (Precision@K, Recall@K, MRR)
[ ] Generation metrics (Accuracy, Faithfulness, Citation quality)
[ ] Regression testing in CI/CD
[ ] A/B testing framework

Monitoring

[ ] Latency percentiles (p50, p95, p99)
[ ] Error rates by type
[ ] Confidence distribution
[ ] User feedback collection (explicit + implicit)
[ ] Automated alerts (quality drop, latency spike, cost surge)
[ ] Cost tracking per query

Operations

[ ] Blue-green deployments
[ ] Rollback capability (< 5 min)
[ ] Vector index rebuild procedure
[ ] Embedding model versioning
[ ] Data retention policy
[ ] Backup/restore tested monthly

Resources

Production RAG is 10% retrieval, 90% engineering. Build observability, guardrails, and evaluation in from day one — retrofitting is 10x harder.

Learning LangChain

Sabi Mantock — Mon, 12 Oct 2026 17:44:01 +0000

A while back I built Synapse, a small app that answers questions using my own notes. I did every part of it by hand: load the files, turn them into vectors, find the closest ones to the question, and hand those to an LLM. It worked, and I understood it, because I'd written every line myself.

So when I started LangChain, a framework that promised to do all of that for me felt a little suspicious. My honest question was: what is this doing that I haven't already done?

It turns out the answer is less "new ideas" and more "a standard shape for the ones you already have." Let me walk through it the way I learned it.

Start small: messages and templates

A chat model doesn't take one big string. It takes a list of messages: a system message that sets the role, then human messages from you, then AI messages from the model. LangChain gives each of those its own class. Nothing magic, it's the same thing the raw API wants.

Prompt templates were the next small step. Instead of building a string every time, you write the prompt once with a gap in it:

from langchain_core.prompts import ChatPromptTemplate

template = ChatPromptTemplate.from_template(
    "I've recently adopted a {pet}. Suggest three {pet} names."
)

Then an output parser takes the model's text reply and turns it into something you can actually use, like a Python list. The thing that stuck with me here is that the parser is only as good as your instructions. Vague instructions about the format gave me a messy list. Strict ones gave me a clean one.

The pipe

This is the bit where it clicked for me. LangChain lets you join those pieces with a |, like a Unix pipe:

chain = chat_template | chat | list_output_parser
chain.invoke({"pet": "dog"})
# ['Max', 'Bella', 'Luna']

A prompt goes in one end, the model sits in the middle, and a clean list comes out the other end. Each step's output is the next step's input. That's the whole idea.

It even works for ordinary Python functions. Wrap them in RunnableLambda and they snap in like any other piece:

from langchain_core.runnables import RunnableLambda

add_up = RunnableLambda(lambda x: sum(x))
square = RunnableLambda(lambda x: x**2)

(add_up | square).invoke([1, 2, 5])   # 8, then 64

You can also draw a chain. chain.get_graph().print_ascii() prints the boxes and arrows, which is surprisingly handy when a chain gets long.

Two more things I liked. RunnableParallel runs two chains side by side and gives both results to the next step. And batch is quicker than looping: two requests through batch took 2.67 seconds in my notebook, while the same two as separate calls took about 7.5.

Now RAG, with the same pieces

Remember my five hand-built steps? LangChain has a piece for each.

For loading, there are loaders for different file types. I used one for PDFs and one for Word documents. Splitting was the new part, since Synapse never did any. I split a six-page document twice. First by its headings, so every chunk remembers which lecture it came from, then by size, about 500 characters with a little overlap. That left me with 20 chunks, each labelled with where it came from.

Each chunk then became 1,536 numbers using OpenAI's embedding model. I wanted to see if it really worked, so I compared three pairs of chunks. They scored 0.88, 0.80 and 0.79, so related text does end up close together. The vectors all have a length of about 1, which is why a plain dot product works as a similarity score.

Those vectors went into Chroma, which saves them to a folder on disk so I don't have to re-embed every time I open the notebook.

Retrieval comes in two flavours. A plain similarity search gives you the closest chunks. MMR gives you close chunks that are also different from each other, so you don't get five copies of the same paragraph.

And the last step is where the pipe comes back. The whole thing is one chain:

chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt_template
    | chat
    | StrOutputParser()
)

chain.invoke("What software do data scientists use?")

The retriever fills in the context, the question passes straight through, the prompt tells the model to answer only from that context and name the lecture it used, and the parser hands back plain text. The answer ended with:

Resources: Programming Languages & Software Employed in Data Science - All the Tools You Need

That's the same "only use these notes" guard I wrote by hand in Synapse, plus the sources I'd been putting off adding.

What this changed for me

LangChain didn't teach me a new idea. Everything in that pipeline is something I'd already done by hand: load, embed, find the closest, ask. What it gave me was a standard shape for each step, and the pipe to join them. Having built it the long way first is what made the short way make sense.

Next up is LangGraph.

Building AI Agents with LangGraph: From Fundamentals to Production

Phil IPS — Mon, 12 Oct 2026 17:25:45 +0000

Building AI Agents with LangGraph: From Fundamentals to Production

A comprehensive guide to building reliable, stateful AI agents using LangGraph — the orchestration framework for agentic workflows.

Introduction

AI agents are LLMs equipped with tools, memory, and reasoning loops. LangGraph provides the orchestration layer to make them production-ready.

Why LangGraph?

Feature	LangChain Agents	LangGraph
State Management	Implicit, limited	Explicit, full control
Cycles/Loops	Difficult	Native support
Human-in-the-Loop	Basic	First-class
Debugging	Hard	Time-travel debugging
Persistence	Manual	Built-in checkpoints
Multi-Agent	Ad-hoc	Structured graphs

Core Concepts

1. State

The single source of truth for your agent:

# state.py
from typing import TypedDict, List, Annotated
from langgraph.graph import add_messages
from pydantic import BaseModel

class AgentState(TypedDict):
    # Messages with automatic reducer (appends)
    messages: Annotated[List[BaseModel], add_messages]

    # Custom fields
    user_id: str
    session_id: str
    tool_call_count: int
    current_plan: str
    retrieved_docs: List[dict]
    error_count: int

2. Nodes

Functions that process state and return updates:

# nodes.py
from langgraph.graph import END
from langchain_core.messages import AIMessage, ToolMessage
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)

async def agent_node(state: AgentState):
    """Main reasoning node."""
    response = await llm_with_tools.ainvoke(state["messages"])

    return {
        "messages": [response],
        "tool_call_count": state.get("tool_call_count", 0) + 
                          len(response.tool_calls) if response.tool_calls else 0
    }

async def tool_node(state: AgentState):
    """Execute tool calls."""
    last_message = state["messages"][-1]
    tool_outputs = []

    for tool_call in last_message.tool_calls:
        tool = get_tool_by_name(tool_call["name"])
        try:
            result = await tool.ainvoke(tool_call["args"])
            tool_outputs.append(ToolMessage(
                content=str(result),
                tool_call_id=tool_call["id"]
            ))
        except Exception as e:
            tool_outputs.append(ToolMessage(
                content=f"Error: {str(e)}",
                tool_call_id=tool_call["id"]
            ))

    return {"messages": tool_outputs}

def should_continue(state: AgentState) -> str:
    """Route: continue to tools or end."""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

3. Edges & Conditional Routing

# graph.py
from langgraph.graph import StateGraph, START

def build_agent_graph():
    workflow = StateGraph(AgentState)

    # Add nodes
    workflow.add_node("agent", agent_node)
    workflow.add_node("tools", tool_node)
    workflow.add_node("validate", validation_node)

    # Edges
    workflow.add_edge(START, "agent")
    workflow.add_conditional_edges(
        "agent",
        should_continue,
        {"tools": "tools", END: END}
    )
    workflow.add_edge("tools", "agent")

    # Compile with checkpointing
    return workflow.compile(
        checkpointer=MemorySaver(),  # Or PostgresSaver for production
        interrupt_before=["tools"]  # Human-in-the-loop
    )

Building a Production Agent

Complete Example: Research Assistant

# research_agent.py
from typing import TypedDict, List, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
import json
import sqlite3
from datetime import datetime

# ============ STATE ============
class ResearchState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    research_topic: str
    sub_questions: List[str]
    gathered_info: List[dict]
    current_step: str
    iteration: int
    max_iterations: int
    final_report: str

# ============ TOOLS ============
@tool
def web_search(query: str, num_results: int = 5) -> str:
    """Search the web for information."""
    # Implementation using your preferred search API
    # Return formatted results
    pass

@tool
def save_research_note(topic: str, content: str, source: str) -> str:
    """Save a research finding."""
    conn = sqlite3.connect("research.db")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS notes (
            id INTEGER PRIMARY KEY,
            topic TEXT, content TEXT, source TEXT, created_at TEXT
        )
    """)
    conn.execute(
        "INSERT INTO notes (topic, content, source, created_at) VALUES (?, ?, ?, ?)",
        (topic, content, source, datetime.utcnow().isoformat())
    )
    conn.commit()
    conn.close()
    return f"Saved note for '{topic}'"

tools = [web_search, save_research_note]

# ============ NODES ============
llm = ChatOpenAI(model="gpt-4o", temperature=0)
planner_llm = llm.bind_tools(tools)
researcher_llm = llm.bind_tools([web_search])
writer_llm = llm.bind_tools([save_research_note])

async def planner(state: ResearchState):
    """Plan the research approach."""
    topic = state["research_topic"]

    prompt = f"""Break down this research topic into 3-5 specific sub-questions:
    Topic: {topic}

    Return as JSON: {{"sub_questions": ["q1", "q2", ...]}}"""

    response = await planner_llm.ainvoke([HumanMessage(content=prompt)])
    plan = json.loads(response.content)

    return {
        "sub_questions": plan["sub_questions"],
        "current_step": "researching",
        "iteration": 0,
        "gathered_info": []
    }

async def researcher(state: ResearchState):
    """Research each sub-question."""
    if state["iteration"] >= len(state["sub_questions"]):
        return {"current_step": "writing"}

    question = state["sub_questions"][state["iteration"]]

    prompt = f"""Research this question thoroughly using web search:
    Question: {question}

    Search for multiple sources and synthesize findings."""

    response = await researcher_llm.ainvoke([
        HumanMessage(content=prompt)
    ])

    # Execute tool calls
    tool_results = []
    for tool_call in response.tool_calls:
        if tool_call["name"] == "web_search":
            result = await web_search.ainvoke(tool_call["args"])
            tool_results.append({
                "question": question,
                "result": result,
                "source": "web_search"
            })

    return {
        "gathered_info": state["gathered_info"] + tool_results,
        "iteration": state["iteration"] + 1
    }

async def writer(state: ResearchState):
    """Compile final report."""
    prompt = f"""Write a comprehensive research report on: {state['research_topic']}

    Sub-questions and findings:
    {json.dumps(state['gathered_info'], indent=2)}

    Structure:
    1. Executive Summary
    2. Detailed Findings (per sub-question)
    3. Key Insights
    4. Sources
    5. Limitations"""

    response = await writer_llm.ainvoke([HumanMessage(content=prompt)])

    # Save final report
    await save_research_note.ainvoke({
        "topic": state["research_topic"],
        "content": response.content,
        "source": "final_report"
    })

    return {
        "final_report": response.content,
        "current_step": "complete"
    }

def route_research(state: ResearchState) -> Literal["researcher", "writer"]:
    return state["current_step"]

# ============ GRAPH ============
def build_research_agent():
    workflow = StateGraph(ResearchState)

    workflow.add_node("planner", planner)
    workflow.add_node("researcher", researcher)
    workflow.add_node("writer", writer)

    workflow.add_edge(START, "planner")
    workflow.add_edge("planner", "researcher")
    workflow.add_conditional_edges(
        "researcher",
        route_research,
        {"researcher": "researcher", "writer": "writer"}
    )
    workflow.add_edge("writer", END)

    # Persistence
    checkpointer = SqliteSaver.from_conn_string("sqlite:///research_agent.db")

    return workflow.compile(checkpointer=checkpointer)

# ============ USAGE ============
async def run_research(topic: str, thread_id: str):
    agent = build_research_agent()

    config = {"configurable": {"thread_id": thread_id}}

    initial_state = {
        "messages": [HumanMessage(content=f"Research: {topic}")],
        "research_topic": topic,
        "max_iterations": 5
    }

    # Stream for progress
    async for chunk in agent.astream(initial_state, config):
        for node, update in chunk.items():
            print(f"[{node}] {update.get('current_step', '')}")
            if update.get("final_report"):
                print("\n=== FINAL REPORT ===")
                print(update["final_report"])

# Run
# asyncio.run(run_research("Impact of AI on software development jobs", "thread-123"))

Advanced Patterns

1. Multi-Agent Collaboration

# multi_agent.py
from langgraph.graph import StateGraph
from typing import Literal

class MultiAgentState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    next_agent: str
    task: str
    results: dict

def supervisor(state: MultiAgentState) -> Literal["researcher", "coder", "reviewer", END]:
    """Route to appropriate specialist."""
    last_msg = state["messages"][-1].content.lower()

    if "research" in last_msg or "find" in last_msg:
        return "researcher"
    elif "code" in last_msg or "implement" in last_msg:
        return "coder"
    elif "review" in last_msg or "check" in last_msg:
        return "reviewer"
    return END

# Each specialist is a subgraph
researcher_graph = build_research_subgraph()
coder_graph = build_coder_subgraph()
reviewer_graph = build_reviewer_subgraph()

workflow = StateGraph(MultiAgentState)
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("researcher", researcher_graph)
workflow.add_node("coder", coder_graph)
workflow.add_node("reviewer", reviewer_graph)

workflow.add_edge(START, "supervisor")
workflow.add_conditional_edges("supervisor", supervisor)
workflow.add_edge("researcher", "supervisor")
workflow.add_edge("coder", "supervisor")
workflow.add_edge("reviewer", "supervisor")

2. Human-in-the-Loop

# human_in_loop.py
from langgraph.types import interrupt, Command

async def sensitive_action(state: AgentState):
    """Action requiring human approval."""
    action = state.get("pending_action")

    # Interrupt and wait for human
    human_response = interrupt({
        "action": action,
        "question": f"Approve {action['name']} with params {action['params']}?"
    })

    if human_response.get("approved"):
        result = await execute_action(action)
        return {"messages": [ToolMessage(content=result, tool_call_id=action["id"])]}
    else:
        return {"messages": [ToolMessage(
            content=f"Rejected: {human_response.get('reason')}",
            tool_call_id=action["id"]
        )]}

# Resume after human input
config = {"configurable": {"thread_id": "thread-1"}}
agent.invoke(Command(resume={"approved": True, "reason": "Looks good"}), config)

3. Time-Travel Debugging

# debugging.py
# Get state history
history = list(agent.get_state_history(config))

for state_snapshot in history:
    print(f"Step: {state_snapshot.metadata['step']}")
    print(f"Messages: {len(state_snapshot.values['messages'])}")
    print(f"Next: {state_snapshot.next}")
    print("---")

# Replay from specific point
past_config = agent.get_state({"configurable": {"thread_id": "thread-1", "checkpoint_id": "checkpoint-5"}})
agent.invoke(new_input, past_config)  # Branches from that point

4. Structured Output with Validation

# structured_output.py
from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field, field_validator

class CodeReview(BaseModel):
    file: str
    issues: List[dict] = Field(default_factory=list)
    suggestions: List[str] = Field(default_factory=list)
    score: int = Field(ge=0, le=100)

    @field_validator("score")
    @classmethod
    def validate_score(cls, v):
        if not 0 <= v <= 100:
            raise ValueError("Score must be 0-100")
        return v

parser = PydanticOutputParser(pydantic_object=CodeReview)

review_prompt = f"""Review this code and output JSON:
{parser.get_format_instructions()}

Code:
{{code}}"""

async def code_reviewer(state):
    code = state["code_to_review"]
    response = await llm.ainvoke(review_prompt.format(code=code))

    try:
        review = parser.parse(response.content)
        return {"review": review.model_dump()}
    except Exception as e:
        # Retry with error feedback
        return {"error": str(e), "retry": True}

Production Checklist

Reliability

[ ] Checkpointing: Use PostgresSaver or RedisSaver for production
[ ] Error handling: Wrap nodes in try/catch, add retry logic
[ ] Timeouts: Set timeout on LLM calls and tool executions
[ ] Rate limiting: Implement token bucket for external APIs
[ ] Circuit breakers: Fail fast when dependencies are down

Observability

[ ] LangSmith tracing: os.environ["LANGCHAIN_TRACING_V2"] = "true"
[ ] Custom metrics: Log latency, token usage, error rates
[ ] State logging: Periodic state snapshots for debugging

# monitoring.py
from langsmith import traceable

@traceable(name="agent_step")
async def monitored_node(state: AgentState):
    start = time.time()
    result = await agent_node(state)
    duration = time.time() - start

    # Custom metrics
    log_metric("node_latency", duration, tags={"node": "agent"})
    log_metric("token_usage", result.get("token_count", 0))

    return result

Scaling

[ ] Horizontal scaling: Stateless nodes + external checkpointer
[ ] Queue-based: Use Celery/RQ for async node execution
[ ] Caching: Cache LLM responses for repeated queries

# caching.py
import hashlib
import redis

redis_client = redis.Redis(decode_responses=True)

async def cached_llm_call(prompt: str, model: str) -> str:
    cache_key = f"llm:{model}:{hashlib.md5(prompt.encode()).hexdigest()}"

    cached = redis_client.get(cache_key)
    if cached:
        return cached

    response = await llm.ainvoke(prompt)
    redis_client.setex(cache_key, 3600, response.content)  # 1hr TTL
    return response.content

Deployment

Docker

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Install LangGraph
RUN pip install langgraph langchain-openai langchain-core

# Copy app
COPY . .

# Run
CMD ["python", "server.py"]

FastAPI Server

# server.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langgraph.checkpoint.sqlite import SqliteSaver
import uvicorn

app = FastAPI()
agent = build_research_agent()
checkpointer = SqliteSaver.from_conn_string("sqlite:///checkpoints.db")

class ChatRequest(BaseModel):
    message: str
    thread_id: str

class ChatResponse(BaseModel):
    response: str
    thread_id: str

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    config = {"configurable": {"thread_id": request.thread_id}}

    try:
        result = await agent.ainvoke(
            {"messages": [HumanMessage(content=request.message)]},
            config
        )

        last_msg = result["messages"][-1]
        return ChatResponse(
            response=last_msg.content,
            thread_id=request.thread_id
        )
    except Exception as e:
        raise HTTPException(500, str(e))

@app.get("/threads/{thread_id}/history")
async def get_history(thread_id: str):
    config = {"configurable": {"thread_id": thread_id}}
    state = agent.get_state(config)
    return {"messages": [m.content for m in state.values["messages"]]}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Kubernetes

# k8s/langgraph-agent.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: langgraph-agent
spec:
  replicas: 3
  selector:
    matchLabels:
      app: langgraph-agent
  template:
    spec:
      containers:
      - name: agent
        image: your-registry/langgraph-agent:v1
        ports:
        - containerPort: 8000
        env:
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: agent-secrets
              key: openai-key
        - name: DATABASE_URL
          value: "postgresql://user:pass@postgres:5432/agent"
        resources:
          requests:
            memory: "1Gi"
            cpu: "500m"
          limits:
            memory: "2Gi"
            cpu: "1000m"
---
apiVersion: v1
kind: Service
metadata:
  name: langgraph-agent
spec:
  selector:
    app: langgraph-agent
  ports:
  - port: 80
    targetPort: 8000

Testing Strategies

# test_agent.py
import pytest
from langgraph.graph import StateGraph

# Unit test nodes
@pytest.mark.asyncio
async def test_planner_node():
    state = {"research_topic": "AI agents", "messages": []}
    result = await planner(state)

    assert "sub_questions" in result
    assert len(result["sub_questions"]) >= 3
    assert result["current_step"] == "researching"

# Integration test with mocked LLM
@pytest.mark.asyncio
async def test_full_graph(mock_llm):
    agent = build_research_agent()

    config = {"configurable": {"thread_id": "test-1"}}
    result = await agent.ainvoke(
        {"messages": [HumanMessage(content="Test topic")], "research_topic": "Test"},
        config
    )

    assert "final_report" in result
    assert len(result["final_report"]) > 100

# Property-based testing
from hypothesis import given, strategies as st

@given(st.text(min_size=10, max_size=200))
@pytest.mark.asyncio
async def test_agent_handles_any_input(topic):
    agent = build_research_agent()
    config = {"configurable": {"thread_id": f"test-{hash(topic)}"}}

    # Should not crash
    result = await agent.ainvoke(
        {"messages": [HumanMessage(content=topic)], "research_topic": topic},
        config
    )
    assert "messages" in result

Migration from LangChain Agents

# Old LangChain Agent
from langchain.agents import AgentExecutor, create_openai_tools_agent

agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = executor.invoke({"input": "Research AI trends"})

# New LangGraph (drop-in replacement)
from langgraph.prebuilt import create_react_agent

agent = create_react_agent(llm, tools, checkpointer=MemorySaver())
result = agent.invoke(
    {"messages": [HumanMessage(content="Research AI trends")]},
    {"configurable": {"thread_id": "1"}}
)

Resources

LangGraph transforms agents from fragile prototypes into reliable, debuggable, production systems. Start simple, add persistence, then scale.

Building AI Agents with LangGraph: A Practical Guide

Phil IPS — Mon, 12 Oct 2026 17:02:03 +0000

Building AI Agents with LangGraph: A Practical Guide

Create stateful, multi-step AI agents that can reason, use tools, and maintain context across complex workflows.

Introduction

LangGraph extends LangChain with graph-based orchestration — enabling cycles, branching, and persistent state. This is essential for building agents that:

Handle multi-turn conversations with memory
Execute complex workflows with decision points
Recover from errors and retry
Human-in-the-loop approval flows

Why LangGraph Over Chains?

Feature	LCEL Chains	LangGraph
Cycles/Loops	❌	✅
Branching	Limited	✅ Full control
State Persistence	Manual	✅ Built-in
Human-in-the-loop	Complex	✅ Native
Debugging	Hard	✅ Visual graph
Parallel Execution	❌	✅

Installation

pip install langgraph langchain-openai langchain-anthropic tavily-python

Core Concepts

1. State

The shared data structure that flows through the graph:

from typing import Annotated, List
from langgraph.graph import StateGraph, MessagesState
from langchain_core.messages import BaseMessage

class AgentState(MessagesState):
    # Extend MessagesState with custom fields
    user_id: str
    session_data: dict
    tool_call_count: int = 0

2. Nodes

Functions that process state and return updates:

from langchain_core.runnables import RunnableConfig

def agent_node(state: AgentState, config: RunnableConfig) -> AgentState:
    # Process messages, call LLM, return updated state
    return {"messages": [response]}

3. Edges

Control flow between nodes — conditional or unconditional:

def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return "end"

Building a ReAct Agent with Tools

Complete Example

# agent.py
from typing import Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
import json

# 1. Define Tools
@tool
def search_web(query: str) -> str:
    """Search the web for current information."""
    # In production, use Tavily, SerpAPI, etc.
    return f"Search results for: {query}"

@tool
def calculator(expression: str) -> str:
    """Evaluate a mathematical expression safely."""
    try:
        # Safe eval for basic math
        allowed = set("0123456789+-*/(). ")
        if all(c in allowed for c in expression):
            return str(eval(expression))
        return "Invalid expression"
    except Exception as e:
        return f"Error: {e}"

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    # Mock - replace with real API
    return f"Weather in {city}: Sunny, 22°C"

tools = [search_web, calculator, get_weather]

# 2. Define State
class AgentState(MessagesState):
    pass

# 3. Create LLM with Tools
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)

# 4. Define Nodes
def agent(state: AgentState) -> AgentState:
    """Main agent reasoning node."""
    messages = state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

tool_node = ToolNode(tools)

# 5. Build Graph
workflow = StateGraph(AgentState)

workflow.add_node("agent", agent)
workflow.add_node("tools", tool_node)

workflow.add_edge(START, "agent")

workflow.add_conditional_edges(
    "agent",
    lambda state: "tools" if state["messages"][-1].tool_calls else "end",
    {"tools": "tools", "end": END}
)

workflow.add_edge("tools", "agent")

app = workflow.compile()

# 6. Run
if __name__ == "__main__":
    # Initial state
    inputs = {
        "messages": [
            HumanMessage(content="What's 25 * 4 + 10? Also search for 'LangGraph tutorial'")
        ]
    }

    # Execute
    for chunk in app.stream(inputs):
        for node, output in chunk.items():
            print(f"\n--- {node} ---")
            if "messages" in output:
                msg = output["messages"][-1]
                print(f"{msg.type}: {msg.content}")
                if msg.tool_calls:
                    print(f"Tool calls: {msg.tool_calls}")

Adding Memory & Persistence

Short-term Memory (Thread-level)

from langgraph.checkpoint.memory import MemorySaver

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

# Run with thread_id for conversation memory
config = {"configurable": {"thread_id": "user-123"}}

# First message
result = app.invoke(
    {"messages": [HumanMessage(content="My name is Alice")]},
    config=config
)

# Second message - remembers name!
result = app.invoke(
    {"messages": [HumanMessage(content="What's my name?")]},
    config=config
)

Long-term Memory (Cross-session)

from langgraph.checkpoint.sqlite import SqliteSaver

# Persistent across restarts
conn = sqlite3.connect("agent_memory.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)

app = workflow.compile(checkpointer=checkpointer)

Human-in-the-Loop Approval

from langgraph.types import interrupt, Command

def sensitive_action(state: AgentState) -> AgentState:
    """Node that requires human approval."""
    # Pause execution and wait for human input
    decision = interrupt({
        "question": "Approve sending this email?",
        "draft": state["email_draft"]
    })

    if decision["approved"]:
        # Send email logic here
        return {"messages": [AIMessage(content="Email sent!")]}
    else:
        return {"messages": [AIMessage(content="Email cancelled.")]}

# Add to graph
workflow.add_node("human_approval", sensitive_action)
workflow.add_edge("agent", "human_approval")
workflow.add_edge("human_approval", "agent")

# Resume after human input
app.invoke(Command(resume={"approved": True}), config=config)

Multi-Agent Workflow

from langgraph.graph import StateGraph
from typing import Literal

class MultiAgentState(MessagesState):
    next_agent: str

# Specialist agents
def researcher(state: MultiAgentState) -> MultiAgentState:
    # Research logic
    return {"messages": [AIMessage(content="Research complete")], "next_agent": "writer"}

def writer(state: MultiAgentState) -> MultiAgentState:
    # Writing logic
    return {"messages": [AIMessage(content="Article written")], "next_agent": "editor"}

def editor(state: MultiAgentState) -> MultiAgentState:
    # Editing logic
    return {"messages": [AIMessage(content="Article edited")], "next_agent": "end"}

# Router
def router(state: MultiAgentState) -> Literal["researcher", "writer", "editor", "end"]:
    return state.get("next_agent", "end")

# Build
workflow = StateGraph(MultiAgentState)
workflow.add_node("researcher", researcher)
workflow.add_node("writer", writer)
workflow.add_node("editor", editor)

workflow.add_edge(START, "researcher")
workflow.add_conditional_edges("researcher", router)
workflow.add_conditional_edges("writer", router)
workflow.add_conditional_edges("editor", router)

app = workflow.compile()

Streaming & Real-time Updates

# Stream tokens as they're generated
for chunk in app.stream(inputs, stream_mode="messages"):
    msg, metadata = chunk
    if msg.content:
        print(msg.content, end="", flush=True)

# Stream state updates
for chunk in app.stream(inputs, stream_mode="values"):
    print(f"State keys: {chunk.keys()}")

Debugging & Visualization

# Visualize graph
from langgraph.graph import draw_graph

# Save as PNG
draw_graph(app.get_graph()).write_png("agent_graph.png")

# Print Mermaid diagram
print(app.get_graph().draw_mermaid())

Output:

graph TD
    START --> agent
    agent -->|tools| tools
    agent -->|end| END
    tools --> agent

Production Checklist

[ ] Error handling — Wrap nodes in try/catch, add retry logic
[ ] Rate limiting — Use RunnableConfig for API limits
[ ] Observability — Add LangSmith tracing
[ ] Validation — Validate tool inputs/outputs with Pydantic
[ ] Timeouts — Set execution timeouts per node
[ ] State schema — Use Pydantic for strict state validation
[ ] Testing — Unit test each node independently

# Example: Node with retry and timeout
from langgraph.graph import RetryPolicy

workflow.add_node(
    "unreliable_api",
    unreliable_node,
    retry=RetryPolicy(max_attempts=3, backoff_factor=2)
)

Advanced Patterns

1. Parallel Tool Execution

from langgraph.graph import StateGraph
from concurrent.futures import ThreadPoolExecutor

def parallel_tools(state: AgentState) -> AgentState:
    tool_calls = state["messages"][-1].tool_calls
    with ThreadPoolExecutor() as executor:
        results = list(executor.map(execute_tool, tool_calls))
    return {"messages": [ToolMessage(content=r, tool_call_id=tc["id"]) 
                            for r, tc in zip(results, tool_calls)]}

2. Dynamic Tool Selection

def select_tools(state: AgentState) -> list:
    """Choose tools based on query type."""
    query = state["messages"][-1].content.lower()
    if "calculate" in query:
        return [calculator]
    elif "weather" in query:
        return [get_weather]
    return [search_web]

3. Reflection & Self-Correction

def reflect(state: AgentState) -> AgentState:
    """Critique the last response and improve."""
    critique_prompt = f"Critique this answer: {state['messages'][-1].content}"
    critique = llm.invoke([HumanMessage(content=critique_prompt)])
    return {"messages": [critique]}

workflow.add_node("reflect", reflect)
workflow.add_edge("agent", "reflect")
workflow.add_conditional_edges("reflect", 
    lambda s: "agent" if "needs improvement" in s["messages"][-1].content.lower() else "end")

Deployment

FastAPI Server

# server.py
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class ChatRequest(BaseModel):
    message: str
    thread_id: str = "default"

@app.post("/chat")
async def chat(request: ChatRequest):
    config = {"configurable": {"thread_id": request.thread_id}}
    result = agent_app.invoke(
        {"messages": [HumanMessage(content=request.message)]},
        config=config
    )
    return {"response": result["messages"][-1].content}

Docker

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

Resources

Part of the AI Agents series. Next: Multi-Agent Systems with LangGraph.

LangGraph Practical Guide: When to Use It Over LangChain

Amrendra N Mishra — Sat, 10 Oct 2026 06:38:57 +0000

Why Automate Content Creation

Creating content manually for 9 channels would take 8+ hours daily. I have a full-time job as VP at JPMorgan. The math doesnt work.

The Pipeline

Trending Topic CSV
       ↓
  AI Script (Ollama/Gemini)
       ↓
  Google Cloud TTS (Voice)
       ↓
  Pexels Images + MoviePy Video
       ↓
  YouTube API Upload

Scheduling with Mac LaunchAgents

<plist version="1.0">
<dict>
  <key>Label</key>
  <string>com.gyaaninfive.youtube</string>
  <key>ProgramArguments</key>
  <array>
    <string>/usr/bin/python3</string>
    <string>/path/to/upload.py</string>
  </array>
  <key>StartCalendarInterval</key>
  <dict>
    <key>Hour</key><integer>18</integer>
    <key>Minute</key><integer>15</integer>
  </dict>
</dict>
</plist>

One plist per channel. Runs at the exact scheduled time.

Total Monthly Cost

Google Cloud TTS: Free (4M chars/month)
Pexels API: Free
YouTube API: Free
Ollama: Free (local)

Total: ₹0/month

Full code on GitHub

9 channels. Daily uploads. Zero manual work.

What Are LLMs, and How Do You Build Stuff With Them?

Aditya Negandhi — Fri, 09 Oct 2026 22:52:19 +0000

Ever wondered how ChatGPT, Claude, Gemini, Deepseek, Llama and friends understand you and write back like a person? Here's the idea this whole post hangs on: an LLM only predicts the next token. So every useful app is the scaffolding you build around it: what context it sees, which tools it can call, and what it remembers.

So what is an LLM?

A Large Language Model is an AI system built to process and generate human-like text. Under the hood it's deep learning, specifically the Transformer architecture from the paper Attention Is All You Need. These models are trained on enormous piles of text like fineweb, which is how they pick up language patterns, context, and meaning. That's what makes them good at generating text, translating, summarising, and a lot more.

Old rule-based systems followed instructions someone wrote. LLMs are probabilistic. They predict the most likely next word based on what they saw in training. That's why they can handle such a wide range of questions and still give answers that fit the context.

The five words you'll keep hearing

Tokenization. Breaking text into smaller units called tokens. A token can be a word, part of a word, or a single character, depending on the tokenizer. Models work on tokens, not raw text.
Embeddings. Lists of numbers (vectors) that represent words or phrases and capture how they relate to each other. They're how a model knows two different words mean similar things.
Context window. The most text a model can handle at once. Bigger window, more it can keep in mind, more coherent the answers. Go past it and the early parts of the conversation fall off.
Temperature. The randomness dial. Turn it up for creative, varied output. Turn it down for predictable, repeatable output.
Fine-tuning. Training a model further on a specific dataset so it gets better at a specialised job, like medical, legal, or financial work.

What LLMs are good and bad at

LLMs, and AI in general, are NOT a magic bullet. Here's where they shine and where they fall on their face.

Where they're awesome

Writing human-like, coherent text that fits the context.
Pulling out and summarising big chunks of information fast.
Working across lots of languages.
Producing structured output.

Where they struggle

They hallucinate, and they're confident about it.
Short memory and a limited context window.
They can be biased, depending on the training data.
Logical reasoning isn't their strong suit.
They're expensive and energy-hungry to run.

OK, that's the model. Now, how do you actually build something with it?

How to build with LLMs

On its own, an LLM can't do much. If you want to solve a real problem and not just ship one more chatbot, you have to give it resources: internet access, tools for the tasks you need (calling an API, hitting a database), and so on. For that, you want LangChain.

LangChain

If you're serious about building apps with LLMs, LangChain is worth having in your toolbox. It gives you a structured way to talk to models and makes it much easier to hook them up to outside data sources, databases, and APIs. It comes with pre-built pieces for memory, for chaining several LLM calls together, and for using external tools like search APIs. Chatbot, research assistant, AI-powered knowledge base, whatever you're making, it takes a lot of the plumbing off your plate.

Why LangChain rocks

Integrations. Plug into different models, databases, and external APIs without much fuss.
Tools. The model can use search APIs, calculators, and other tools to give better answers.
Chains. String several LLM calls together like a conversation flow.
Memory. Keep conversation context across multiple exchanges.

Keeping your AI chat on track

LLMs have a limited attention span (that context window again), so long conversations get messy. Three ways to keep them manageable:

Trimming. Cut the parts of the conversation you don't need.
Filtering. Keep only the most relevant messages.
Summarising. Turn a long-winded chat into a short recap.

What can you build with it?

Plenty: chatbots, smart search, writing assistants, automated research tools. It's great for anything that needs to talk to databases, external APIs, and knowledge retrieval systems.

It has limits though. LangChain structures your AI interactions, but it's still stuck with the model's own constraints, like the context window and no long-term memory. And on its own it doesn't give you advanced decision-making or complex multi-step orchestration. That's where LangGraph comes in.

LangGraph

LangGraph is a flexible framework for building dynamic, multi-step AI apps. Where a simple app is prompt in, answer out, LangGraph lets you build workflows with multiple decision points, several agents talking to each other, and structured automation. It's especially handy when your AI needs to go back and forth with users, juggle different workflows, or remember things across sessions.

With it you can route queries, break a big task into smaller steps, and run things in parallel. Automating customer service flows, walking users through a process step by step, generating content with AI in the loop, it gives you the pieces to orchestrate all of that.

Storing conversation data

Want your AI to "remember" things? LangGraph gives you options:

Memory Saver. Stores data locally for quick access.
Postgres Saver. Uses PostgreSQL for external storage.
MongoDB. Stores conversations in a NoSQL database.
Redis Saver. Fast, efficient storage for AI memory.

Getting it production-ready

Use a different thread ID for each conversation.
Save conversation history with persistent checkpoints.
Avoid huge user prompts that overwhelm the model.
Stream responses so it feels smooth.
Handle errors (rate limits, content moderation, and so on).

Prompt engineering

To get accurate, relevant answers out of an LLM, the prompt matters a lot. Prompt engineering is part art, part science, and how you structure the question changes the quality of the answer.

Who's who in a conversation

SystemMessage. Sets the ground rules, behaviour, and personality of the AI. It's where you define a persona, a tone, or overall guidelines.
HumanMessage. What the user sends, usually text typed by a person.
AIMessage. What the model sends back: text, or a request to call a tool.
ToolMessage. Carries the result of a tool call back to the model, usually when outside data got fetched or processed.

Where I'd stop trusting this post

You might not need a framework at all. The best counter-argument comes from Anthropic's Building effective agents, which says to start with direct API calls, because frameworks add layers that hide the prompts and make debugging harder. If your app is one or two model calls, the plain SDK is less to learn and less to break.

This is a map from February 2025. Framework APIs in this space change every few months, and my list of things LLMs "struggle" with has shrunk since. Check the current docs before you copy a pattern from here.

It's an overview, not a build log. There's no code in this post and no measurements. For a worked example, Finchat is the app these notes came out of.

The takeaway

Building with LLMs is easier than ever with frameworks like LangChain and LangGraph. Chatbots, smart assistants, knowledge-search tools, they all come down to prompt engineering, context management, and conversation workflows.

If you remember one thing: decide what the model needs to see, what it's allowed to do, and what it should remember. In that order. The right framework is whichever one makes those three decisions easiest to read.

✌️ Stay curious, Keep coding, Peace nerds!

Updated 10 October 2026: rewrote this in plainer language and dropped the GIF. The central idea, primary sources, and limits section were added on 9 October 2026.

Blender rendering from an OpenAI Agents SDK or LangChain agent, and the 5-second MCP timeout that breaks it

JANCTION Render — Fri, 09 Oct 2026 09:53:53 +0000

Official JANCTION Render: https://render.janction.jp

Update (2026-10-12): the welcome credit changed on 2026-10-12. A new key now gets 100 JPY of GPU time with no card, and 400 JPY more after saving a card on a page that charges nothing (once per card), up to 500 JPY in all. The prices below were updated to match https://render.janction.jp/pricing.

JANCTION Render is a cloud GPU render farm for Blender that AI agents call as an MCP server or HTTP API. It is free to start (100 JPY of GPU time with no card, 400 JPY more after saving a card that is not charged), then 0.1 JPY per GPU-second. This post is for people building their own agent: how to connect it from the OpenAI Agents SDK, LangChain or plain Python, and the one setting that makes the first call fail.

The error

McpError: Timed out while waiting for response to ClientRequest. Waited 5.0 seconds.

MCPServerStreamableHttp in the OpenAI Agents SDK (0.23.1) waits 5 seconds for a tool call by default (client_session_timeout_seconds=5). A render tool answers when the image exists, not when the job is queued. We called render_preview on the sample scene with four lighting presets, once with the default and once with 120 seconds:

default: failed after 5.1 s with the error above
120 s: the image came back after 7.3 s (4.03 GPU-seconds of rendering)

A lighter preview of the same scene (one preset) came back in 4.7 to 5.0 seconds, right at the edge, so the default can look fine in a quick test and then fail on the first real scene. Set the timeout explicitly.

OpenAI Agents SDK

The service is a remote MCP server at https://render.janction.jp/mcp. The key goes in as a bearer token; curl -s -X POST https://render.janction.jp/v1/keys gives one, with no sign-up.

import asyncio
from agents import Agent, Runner
from agents.mcp import MCPServerStreamableHttp

async def main():
    async with MCPServerStreamableHttp(name="janction-render", client_session_timeout_seconds=120,
                                       params={"url": "https://render.janction.jp/mcp",
                                               "headers": {"Authorization": "Bearer jr_..."}}) as render:
        agent = Agent(name="3D artist", instructions="Render Blender scenes with janction-render.", mcp_servers=[render])
        result = await Runner.run(agent, "Render a preview of https://render.janction.jp/samples/cube_scene.py")
        print(result.final_output)

asyncio.run(main())

The agent sees 12 tools. The ones it uses in order are scene_info, render_preview, render_estimate, render_final, render_status and render_download. Long renders do not need a longer timeout: a short final returns its links in one call, a longer one returns a job id, and render_download(job_id, wait_seconds=45) waits up to 45 seconds per call. 120 seconds covers every tool.

LangChain

With langchain-mcp-adapters (0.3.2), the same server is one entry in MultiServerMCPClient. The transport name is streamable_http.

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient

async def main():
    client = MultiServerMCPClient({"janction-render": {"url": "https://render.janction.jp/mcp", "transport": "streamable_http",
                                                       "headers": {"Authorization": "Bearer jr_..."}}})
    tools = await client.get_tools()           # render_preview, render_final, render_status, render_download, ...
    print([t.name for t in tools])

asyncio.run(main())

Without an agent framework

If your code decides what to render and only needs the frames, the Python client is shorter. pip install janction-render; the first call creates a free key and keeps it in ~/.janction-render.json.

from janction_render.client import Client

c = Client()                                   # or Client(api_key="jr_...")
scene = c.upload("scene.py")                   # a bpy script, a .blend, or a .glb / .fbx / .usd / .obj file
job = c.submit(scene["scene_id"], kind="final", frame_start=1, frame_end=48, output="mp4")
job = c.wait(job["job_id"], timeout=1800)
print(job["status"], c.download(job["job_id"], "out", only="mp4"))

For a service rather than a script: pass notify_url (https) to POST /v1/jobs and get one JSON POST when the job finishes instead of polling, and send an Idempotency-Key header so a retry does not start a second job.

What it costs an agent

Each new key gets 100 JPY of GPU time with no card, and 400 JPY more after saving a card on a page that charges nothing (once per card): up to 500 JPY, or 5,000 GPU-seconds, for 14 days. After that, previews stay free up to 2 GPU-minutes a day and other GPU time is 0.1 JPY per GPU-second. A 24-frame 720p MP4 at 64 samples took about 58 GPU-seconds, about 6 JPY. When a key is out of credit, the job is refused before it runs: the HTTP API answers 402 with a top-up link, and over MCP the tool result carries the same link for a person to open. Each key also has a spending cap per job and per day. More on that: https://dev.to/_903a3934b362f0afe2a8/letting-an-ai-agent-pay-for-renders-without-surprises-free-credit-first-a-top-up-link-spending-3i1m

The same examples, kept current: https://github.com/JasmyLab-JANCTION/janction-render#from-your-own-agent-or-app. A plain-text summary for models: https://render.janction.jp/llms.txt

Official JANCTION Render: https://render.janction.jp (pricing: https://render.janction.jp/pricing, facts: https://render.janction.jp/facts). Code and client: https://github.com/JasmyLab-JANCTION/janction-render