Building an Enterprise-Grade Bank Policy Assistant with LangGraph: Multi-Agent RAG with Memory and State

Introduction: The Chain-of-Thought & Few-Shot Reality Check

Before diving into the architecture, let’s address the core question regarding Chain-of-Thought (CoT) and Few-Shot Prompting:

Factualness: Both techniques significantly improve factual accuracy in complex domains like banking.
- CoT forces the model to "show its work," reducing hallucinations by breaking down logical steps (e.g., "First, check the customer's tenure; second, verify the loan type...").
- Few-Shot provides concrete examples of correct reasoning patterns, anchoring the model to specific compliance standards.
Latency: Both increase latency.
- CoT generates more tokens because it outputs intermediate reasoning steps before the final answer.
- Few-Shot increases the input context size, which can slow down initial token generation depending on the LLM provider.
- Enterprise Trade-off: In banking, accuracy and auditability trump speed. A 2-second delay is acceptable if it prevents a compliance violation.

Real-Time Use Case: "PolicyGuard" – The Intelligent Compliance Officer

Scenario: A large bank has thousands of policy documents (PDFs) covering loans, KYC, fraud detection, and interest rates. Customer support agents often struggle to find the exact clause for edge cases.

The Problem:

Standard RAG retrieves relevant chunks but fails at multi-hop reasoning (e.g., "Is this customer eligible for a gold loan if they have a minor credit default from 3 years ago?").
No memory of previous interactions within a session.
No structured audit trail for compliance.

The Solution: A Multi-Agent LangGraph System that:

Router Agent: Determines if the query is about Loans, KYC, or General Info.
Retriever Agent: Fetches relevant policy documents using Vector Search.
Reasoning Agent: Uses Chain-of-Thought to interpret the policy against the user's specific context.
Compliance Checker: Validates the answer against strict regulatory rules.
Memory Layer: Stores the conversation state for follow-up questions.

Technology Stack

LangGraph: For orchestrating multi-agent workflows and state management.
LangChain: For RAG pipelines and tool integration.
FastAPI: To expose the system as a RESTful API.
Pydantic: For strict data validation and schema definition.
ChromaDB / Pinecone: Vector database for storing policy embeddings.
OpenAI GPT-4o / Azure OpenAI: LLM for reasoning and generation.
Redis: For short-term conversational memory (session state).

Step-by-Step Implementation

1. Project Structure & Dependencies

pip install langgraph langchain-openai langchain-chroma fastapi uvicorn pydantic redis

2. Define the State Schema (Pydantic + Annotated)

We use Annotated to define the state that flows between agents. This ensures type safety and clarity.

from typing import Annotated, List, Dict, Optional
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, START, END
from langgraph.messages import add_messages
import operator

class PolicyState(BaseModel):
    """State object passed between nodes in the LangGraph."""
    
    # User input
    question: str = Field(..., description="The user's query regarding bank policy")
    
    # Retrieved context from Vector DB
    context_chunks: List[str] = Field(default_factory=list, description="Relevant policy documents retrieved")
    
    # Intermediate reasoning steps (Chain-of-Thought)
    reasoning_steps: List[str] = Field(default_factory=list, description="Step-by-step logical deduction")
    
    # Final answer
    answer: Optional[str] = Field(None, description="The final compliant response to the user")
    
    # Metadata for audit
    source_documents: List[str] = Field(default_factory=list, description="IDs of documents used")
    
    # Conversation history for memory
    messages: Annotated[List[dict], operator.add] = Field(default_factory=list, description="Chat history")
    
    # Next agent to route to
    next_agent: Optional[str] = Field(None, description="Routing decision")

3. The Retrieval Agent (RAG Component)

This agent fetches relevant policy documents based on the user's question.

from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

# Mock embedding function for demonstration
embeddings = OpenAIEmbeddings(model="text-embedding-ada-002")

# Assume 'policy_db' is already populated with bank policies
vector_store = Chroma(persist_directory="./bank_policies", embedding_function=embeddings)

def retrieve_policy_docs(state: PolicyState) -> PolicyState:
    """Retrieves relevant policy documents based on the question."""
    print(f"🔍 Retrieving docs for: {state.question}")
    
    # Perform similarity search
    docs = vector_store.similarity_search(state.question, k=3)
    
    # Extract page content
    context = [doc.page_content for doc in docs]
    source_ids = [doc.metadata.get('source', 'Unknown') for doc in docs]
    
    return PolicyState(
        **state.dict(),
        context_chunks=context,
        source_documents=source_ids
    )

4. The Reasoning Agent (Chain-of-Thought)

This agent uses Few-Shot prompting and CoT to analyze the retrieved documents.

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

REASONING_PROMPT = """
You are a Senior Bank Compliance Officer. 
Analyze the following policy documents and answer the user's question.

**Instructions:**
1. Think step-by-step (Chain-of-Thought).
2. Cite specific clauses from the provided context.
3. If the information is missing, state that clearly.
4. Ensure the tone is professional and precise.

**Few-Shot Example:**
User: "Can I get a home loan if I'm self-employed?"
Thought: 
- Step 1: Check self-employment eligibility criteria.
- Step 2: Verify income proof requirements.
- Step 3: Cross-reference with current interest rates.
Answer: "Yes, self-employed individuals are eligible provided they submit 2 years of ITR..."

**Context:**
{context}

**Question:**
{question}

**Your Reasoning & Answer:**
"""

def reason_and_answer(state: PolicyState) -> PolicyState:
    """Generates a reasoned answer using CoT and Few-Shot principles."""
    print("🧠 Reasoning Agent processing...")
    
    context_text = "\n\n".join(state.context_chunks)
    
    prompt = REASONING_PROMPT.format(context=context_text, question=state.question)
    
    response = llm.invoke(prompt)
    
    # Simple parsing to separate reasoning from final answer (in production, use structured output)
    full_response = response.content
    
    return PolicyState(
        **state.dict(),
        answer=full_response,
        reasoning_steps=[full_response.split("Answer:")[0]] if "Answer:" in full_response else [full_response]
    )

5. The Compliance Checker Agent (Guardrail)

This agent ensures the answer doesn't violate any hard-coded rules.

COMPLIANCE_PROMPT = """
You are a Compliance Auditor. Review the following answer for any regulatory violations.
Check for:
1. Misleading financial advice.
2. Missing disclaimers.
3. Incorrect interest rate citations.

If valid, return "APPROVED". If not, return "REJECTED" with reasons.

Answer to review:
{answer}
"""

def check_compliance(state: PolicyState) -> PolicyState:
    """Validates the generated answer against compliance rules."""
    print("🛡️ Compliance Checker validating...")
    
    prompt = COMPLIANCE_PROMPT.format(answer=state.answer)
    response = llm.invoke(prompt)
    
    if "REJECTED" in response.content:
        # In a real system, you might loop back to the Reasoning Agent
        state.answer = f"Compliance Issue Detected: {response.content}. Please consult a human officer."
    
    return state

6. Building the LangGraph Workflow

# Define the graph
workflow = StateGraph(PolicyState)

# Add nodes
workflow.add_node("retriever", retrieve_policy_docs)
workflow.add_node("reasoner", reason_and_answer)
workflow.add_node("compliance_checker", check_compliance)

# Define edges
workflow.add_edge(START, "retriever")
workflow.add_edge("retriever", "reasoner")
workflow.add_edge("reasoner", "compliance_checker")
workflow.add_edge("compliance_checker", END)

# Compile the graph
app = workflow.compile()

7. FastAPI Endpoint with Memory Integration

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="Bank Policy AI Assistant")

class QueryRequest(BaseModel):
    question: str
    session_id: str = "default_session"

@app.post("/ask-policy")
async def ask_policy(request: QueryRequest):
    """
    End-to-end endpoint that handles state, memory, and multi-agent reasoning.
    """
    
    # Initialize state with current question and history (mocked memory retrieval)
    initial_state = PolicyState(
        question=request.question,
        messages=[{"role": "user", "content": request.question}]
    )
    
    # Invoke the LangGraph
    final_state = await app.ainvoke(initial_state)
    
    # Save to memory (Redis/DB) - Mocked here
    # save_to_memory(request.session_id, final_state.messages)
    
    return {
        "answer": final_state.answer,
        "sources": final_state.source_documents,
        "reasoning_trace": final_state.reasoning_steps,
        "status": "success"
    }

Key Takeaways for Enterprise Deployment

State Management: LangGraph’s StateGraph allows us to pass complex objects (PolicyState) between agents, ensuring no data loss during multi-hop reasoning.
Auditability: By storing reasoning_steps and source_documents, we create a transparent trail for compliance audits.
Modularity: Each agent (Retriever, Reasoner, Compliance) can be tested and improved independently.
Latency vs. Accuracy: The CoT approach adds ~2-3 seconds but drastically reduces legal risk.