In the enterprise AI space, building a Multi-Agent LangGraph RAG system is only 20% of the battle. The remaining 80% is ensuring that the system behaves identically across Development, Staging, and Production environments. The AI/LLM ecosystem is notoriously volatile. langchain-core, langgraph, and vector databases like faiss-cpu or pgvector update frequently, often introducing breaking changes or clashing with legacy internal dependencies (the infamous "Pydantic v1 vs v2" wars). Furthermore, LangGraph’s State and Memory backends must seamlessly transition from lightweight local stores in Dev to distributed, highly available databases in Prod. In this end-to-end guide, we will take the Cross-Border Wire Transfer Fraud Detection Multi-Agent System and engineer a bulletproof CI/CD and environment management pipeline. We will solve dependency conflicts, ensure 100% environment reproducibility, and dynamically route LangGraph’s memory state based on the deployment tier.
1. The Real-World Use Case: Scaling the Fraud Agent
Our Fraud Detection Agent uses a sliding window for math, a RAG pipeline for context, and LangGraph for orchestration.
The Deployment Challenge:
Dev Environment: Data scientists run the agent locally using MemorySaver (in-memory state) and an in-memory FAISS vector store.
Staging Environment: QA tests the agent against a staging PostgreSQL database using PostgresSaver for persistent LangGraph state, and a managed Pinecone/Milvus vector store.
Production Environment: The agent runs in a Kubernetes cluster with strict security, connecting to a highly available Postgres cluster for state, and a production vector DB.
The Goal: A single codebase that automatically resolves dependencies, builds reproducible Docker images, and dynamically configures the LangGraph State/Memory backend based on the environment, without code changes.
2. The Enterprise Strategy
To achieve this, we implement a four-pillar strategy:
Strict Dependency Locking: Using Poetry (or modern alternatives like uv) to generate a cryptographic lockfile (poetry.lock), ensuring the exact same sub-dependencies are installed everywhere.
Multi-Stage Dockerization: Separating the heavy build environment (compiling C-extensions for FAISS) from the slim runtime environment.
Environment-Agentic Configuration: Using Pydantic BaseSettings to inject environment-specific configurations (like DB URLs) without hardcoding them.
Dynamic State Routing: Programmatically selecting the LangGraph Checkpointer (Memory vs. Postgres) at runtime based on the environment.
3. End-to-End Code Implementation
Step 1: Strict Dependency Management (pyproject.toml)
We use Poetry to manage dependencies. Notice how we pin the core LangGraph ecosystem and handle the C-extension dependencies for vector stores.
[tool.poetry]
name = "enterprise-fraud-agent"
version = "1.2.0"
description = "Multi-Agent LangGraph Fraud Detection System"
authors = ["Enterprise AI Team <[email protected]>"]
[tool.poetry.dependencies]
python = "^3.11" # Pinning Python version is critical for reproducibility
# Core LangGraph & LangChain Ecosystem
langgraph = "0.2.34"
langchain-core = "0.3.0"
langchain-openai = "0.2.0"
# Vector Stores & Embeddings
faiss-cpu = "1.8.0" # Dev/Local
langchain-postgres = "0.0.12" # Prod pgvector
# Environment & Config Management
pydantic-settings = "2.5.2"
[tool.poetry.group.dev.dependencies]
pytest = "^8.0.0"
ruff = "^0.6.0"
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"
Enterprise Rule: The poetry.lock file must be committed to version control. Never run poetry update in a CI/CD pipeline; always run poetry install --frozen-lockfile.
Step 2: Multi-Stage Dockerfile for Reproducibility
AI dependencies often require system-level C-libraries. We use a multi-stage Docker build to keep the final production image small and secure.
# --- STAGE 1: Builder ---
FROM python:3.11-slim as builder
# Install system dependencies required for compiling C-extensions (like FAISS)
RUN apt-get update && apt-get install -y --no-install-recommends \
build-essential \
gcc \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /app
# Install Poetry
RUN pip install --no-cache-dir poetry==1.8.3
# Configure Poetry to not create a virtual environment inside the Docker container
RUN poetry config virtualenvs.create false
# Copy dependency files first to leverage Docker cache
COPY pyproject.toml poetry.lock ./
# Install dependencies (frozen to ensure exact reproducibility)
RUN poetry install --only main --no-interaction --no-ansi
# --- STAGE 2: Runtime ---
FROM python:3.11-slim as runtime
# Install only runtime system dependencies (no compilers)
RUN apt-get update && apt-get install -y --no-install-recommends \
libgomp1 \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /app
# Copy installed packages from the builder stage
COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY --from=builder /usr/local/bin /usr/local/bin
# Copy application code
COPY ./src ./src
# Set environment variables
ENV PYTHONUNBUFFERED=1 \
PYTHONDONTWRITEBYTECODE=1 \
APP_ENV=production
# Run the application
CMD ["python", "-m", "src.main"]
Step 3: Environment-Agnostic Configuration (src/config.py)
We use pydantic-settings to load configurations from environment variables. This allows the exact same Docker image to be deployed to Dev, Staging, and Prod.
from pydantic_settings import BaseSettings
from typing import Literal
class Settings(BaseSettings):
# Environment tier
app_env: Literal["development", "staging", "production"] = "development"
# LLM Configuration
openai_api_key: str
llm_model: str = "gpt-4o"
# LangGraph State & Memory Configuration
# In Dev, we use in-memory. In Staging/Prod, we use Postgres.
postgres_dsn: str | None = None
# RAG Vector Store Configuration
vector_store_type: Literal["faiss", "pgvector"] = "faiss"
vector_store_url: str | None = None
class Config:
env_file = ".env"
case_sensitive = False
settings = Settings()
Step 4: Dynamic State & Memory Routing (src/main.py)
This is where the magic happens. We define the LangGraph Multi-Agent workflow, but we dynamically inject the Checkpointer (Memory/State backend) based on the environment.
import asyncio
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_postgres import PostgresSaver # For Prod State
from typing import TypedDict, Dict, Any
from src.config import settings
from src.agents import sliding_window_node, rag_node, decision_node
# 1. Define the State
class FraudState(TypedDict):
transaction_id: str
user_id: str
amount: float
window_history: list[float]
z_score: float
is_anomaly: bool
rag_context: str
final_decision: str
# 2. Build the Graph
def build_fraud_graph() -> StateGraph:
workflow = StateGraph(FraudState)
workflow.add_node("math", sliding_window_node)
workflow.add_node("rag", rag_node)
workflow.add_node("decision", decision_node)
workflow.add_edge(START, "math")
# Routing logic omitted for brevity, assumes conditional edges exist
workflow.add_edge("math", "rag")
workflow.add_edge("rag", "decision")
workflow.add_edge("decision", END)
return workflow
# 3. Dynamically Resolve the Memory/State Backend
def get_checkpointer():
"""
Enterprise Pattern: Route LangGraph State to the appropriate backend
based on the deployment environment.
"""
if settings.app_env == "development":
print(" [DEV] Using In-Memory Checkpointer for LangGraph State")
return MemorySaver()
elif settings.app_env in ["staging", "production"]:
if not settings.postgres_dsn:
raise ValueError("POSTGRES_DSN is required for staging/production environments!")
print(f" [{settings.app_env.upper()}] Connecting to Postgres for Persistent LangGraph State")
# PostgresSaver handles the persistent state and memory of the graph
return PostgresSaver.from_conn_string(settings.postgres_dsn)
else:
raise ValueError(f"Unknown environment: {settings.app_env}")
async def main():
# Initialize Graph
graph_builder = build_fraud_graph()
# Initialize Checkpointer (Memory/State)
checkpointer = get_checkpointer()
# If using Postgres, we must setup the tables for state persistence
if settings.app_env != "development":
await checkpointer.setup()
# Compile the graph with the environment-specific checkpointer
app = graph_builder.compile(checkpointer=checkpointer)
# Simulate a transaction stream
config = {"configurable": {"thread_id": "user_12345"}}
# The graph will now persist state across runs in Prod,
# but remain ephemeral in Dev!
result = await app.ainvoke({
"transaction_id": "tx_999",
"user_id": "user_12345",
"amount": 250000.00,
"window_history": []
}, config=config)
print(f"Final Decision: {result['final_decision']}")
if __name__ == "__main__":
asyncio.run(main())
Step 5: CI/CD Gatekeeping (.github/workflows/deploy.yml)
To prevent "it works on my machine" from reaching production, we enforce strict checks in our CI pipeline.
name: Enterprise Fraud Agent CI/CD
on:
push:
branches: [main, staging, develop]
jobs:
dependency-and-build-check:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install Poetry
run: pip install poetry==1.8.3
# CRITICAL: Ensure the lockfile is up to date with pyproject.toml
- name: Check Lockfile Integrity
run: poetry check --lock
# CRITICAL: Install exactly what is in the lockfile
- name: Install Dependencies
run: poetry install --no-interaction --no-ansi
- name: Run Linting and Tests
run: |
poetry run ruff check src/
poetry run pytest tests/
- name: Build Docker Image
run: |
docker build -t fraud-agent:${{ github.sha }} .
# Push to registry and deploy to K8s based on branch
- name: Deploy to Environment
run: |
if [[ ${{ github.ref }} == 'refs/heads/main' ]]; then
echo "Deploying to PRODUCTION with Postgres State..."
# kubectl set image deployment/fraud-agent ...
elif [[ ${{ github.ref }} == 'refs/heads/staging' ]]; then
echo "Deploying to STAGING with Postgres State..."
4. Why This Architecture Wins in the Enterprise
Eliminates Dependency Drift: By enforcing poetry install --frozen-lockfile in CI and Docker, you guarantee that the exact same versions of langgraph, pydantic, and faiss-cpu are running in Prod as were tested in Dev. No more surprise breaking changes from minor version bumps.
Decouples Code from Infrastructure: The Python code doesn't know or care if it's talking to an in-memory dictionary or a distributed Postgres cluster. The Settings class handles the routing. This means you can test the agent locally in seconds without spinning up Docker containers for Postgres.
Optimized Image Sizes: The multi-stage Docker build ensures that heavy compilers (gcc) and build tools are stripped from the final production image, reducing the attack surface and speeding up Kubernetes pod scaling.
Stateful Multi-Agent Reliability: By dynamically swapping the LangGraph Checkpointer, you ensure that in Production, if a pod crashes mid-investigation, the LangGraph State (the sliding window history, the RAG context retrieved, the agent's thought process) is safely persisted in Postgres and can be resumed. In Dev, you avoid the overhead of managing a database.
5. Conclusion
Building a Multi-Agent LangGraph RAG system is an exercise in managing complexity. By combining strict dependency locking (Poetry), immutable infrastructure (Multi-stage Docker), and environment-aware state routing (Pydantic Settings + Dynamic Checkpointers), you transform a fragile AI prototype into a resilient, enterprise-grade production system. This methodology ensures that your AI agents don't just work in a notebook—they scale reliably across the enterprise, maintaining perfect state and memory integrity from the developer's laptop to the production cluster.
Join the conversation! Your thoughts help the community grow.