How to Integrate LangChain for retail banking with PostgreSQL for multi-agent systems

Combining LangChain for retail banking with PostgreSQL gives you a clean pattern for building multi-agent systems that need both reasoning and durable state. In practice, that means one agent can handle customer intent, another can fetch account context, and PostgreSQL keeps the shared memory, audit trail, and workflow state consistent across turns.

For retail banking, this is the difference between a demo chatbot and an operational assistant that can coordinate balance checks, card disputes, loan pre-qualification, and handoffs without losing context.

Prerequisites

• •Python 3.10+
• •A PostgreSQL 14+ instance reachable from your app
• •A LangChain-compatible banking agent setup
• •Installed packages:
  • •langchain
  • •langchain-openai or your preferred LLM provider
  • •langchain-community
  • •psycopg2-binary
  • •sqlalchemy
• •A PostgreSQL database and credentials
• •Environment variables set:
  • •OPENAI_API_KEY
  • •POSTGRES_URL or equivalent connection string
• •Basic familiarity with:
  • •LangChain tools/agents
  • •SQLAlchemy engine creation
  • •PostgreSQL schemas and tables

Integration Steps

1. •

   Install dependencies and create the database connection

   Start by wiring PostgreSQL through SQLAlchemy. LangChain’s Postgres-backed components expect a normal DB connection underneath, so keep this part boring and explicit.

   import os
   from sqlalchemy import create_engine
   
   POSTGRES_URL = os.environ["POSTGRES_URL"]
   engine = create_engine(POSTGRES_URL, pool_pre_ping=True)
   
   with engine.connect() as conn:
       result = conn.execute("SELECT 1")
       print(result.scalar())
   

2. •

   Create a shared state table for multi-agent memory

   Multi-agent systems fail when each agent keeps its own isolated context. Store conversation state in PostgreSQL so the orchestrator, banking agent, and escalation agent all read the same record.

   from sqlalchemy import text
   
   ddl = """
   CREATE TABLE IF NOT EXISTS agent_state (
       session_id TEXT PRIMARY KEY,
       customer_id TEXT NOT NULL,
       active_agent TEXT NOT NULL,
       state JSONB NOT NULL DEFAULT '{}'::jsonb,
       updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
   );
   """
   
   with engine.begin() as conn:
       conn.execute(text(ddl))
       conn.execute(text("""
           INSERT INTO agent_state (session_id, customer_id, active_agent, state)
           VALUES (:session_id, :customer_id, :active_agent, :state::jsonb)
           ON CONFLICT (session_id) DO UPDATE SET
               active_agent = EXCLUDED.active_agent,
               state = EXCLUDED.state,
               updated_at = NOW()
       """), {
           "session_id": "sess_001",
           "customer_id": "cust_123",
           "active_agent": "banking_router",
           "state": '{"intent":"balance_check","step":"start"}'
       })
   

3. •

   Build a LangChain banking agent and expose PostgreSQL-backed tools

   The practical pattern is: LangChain handles reasoning; PostgreSQL handles persistence; tools bridge the two. Use SQL queries as tools for account lookup, case status, or session recovery.

   from langchain_openai import ChatOpenAI
   from langchain_core.tools import tool
   from langchain.agents import initialize_agent, AgentType
   
   llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
   
   @tool
   def get_session_state(session_id: str) -> str:
       """Fetch multi-agent session state from PostgreSQL."""
       with engine.connect() as conn:
           row = conn.execute(
               text("SELECT state::text FROM agent_state WHERE session_id = :sid"),
               {"sid": session_id},
           ).fetchone()
       return row[0] if row else "{}"
   
   @tool
   def update_session_agent(session_id: str, active_agent: str) -> str:
       """Update which agent is currently handling the session."""
       with engine.begin() as conn:
           conn.execute(
               text("""
                   UPDATE agent_state
                   SET active_agent = :active_agent,
                       updated_at = NOW()
                   WHERE session_id = :sid
               """),
               {"sid": session_id, "active_agent": active_agent},
           )
       return f"updated:{session_id}:{active_agent}"
   
   tools = [get_session_state, update_session_agent]
   
   agent = initialize_agent(
       tools=tools,
       llm=llm,
       agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
       verbose=True,
   )
   

4. •

   Add a routing layer for multi-agent banking workflows

   In retail banking, you usually want a router agent to decide whether to answer directly or hand off to a specialist. Keep the router’s decision in PostgreSQL so every downstream agent sees the same workflow state.

   def route_customer_request(session_id: str, message: str) -> str:
       prompt = f"""
       You are a retail banking router.
       Decide whether this should go to balance_lookup, fraud_review, or loan_prequal.
       Session: {session_id}
       Message: {message}
       """
       decision = agent.run(prompt)
   
       with engine.begin() as conn:
           conn.execute(
               text("""
                   UPDATE agent_state
                   SET active_agent = :agent_name,
                       state = jsonb_set(state, '{route_decision}', to_jsonb(:decision::text), true),
                       updated_at = NOW()
                   WHERE session_id = :sid
               """),
               {"sid": session_id, "agent_name": decision.strip(), "decision": decision.strip()},
           )
   
       return decision
   

5. •

   Persist outcomes after each tool call

   Don’t only store routing decisions. Store outcomes too: fetched balances, escalation IDs, verification status. That gives you replayability and auditability when compliance asks what happened in the conversation.

    def persist_outcome(session_id: str, key: str, value: str) -> None:
        with engine.begin() as conn:
            conn.execute(
                text("""
                    UPDATE agent_state
                    SET state = jsonb_set(state, ARRAY[:key], to_jsonb(:value::text), true),
                        updated_at = NOW()
                    WHERE session_id = :sid
                """),
                {"sid": session_id, "key": key, "value": value},
            )
   
    persist_outcome("sess_001", "balance_lookup_result", "$4,218.22")
   

Testing the Integration

Run a simple end-to-end check: write state to PostgreSQL, read it through a LangChain tool call path, then verify the record updates correctly.

response1 = route_customer_request(
    "sess_001",
    "What's my current checking account balance?"
)
print("ROUTE:", response1)

with engine.connect() as conn:
    row = conn.execute(
        text("SELECT active_agent, state::text FROM agent_state WHERE session_id = :sid"),
        {"sid": "sess_001"},
    ).fetchone()

print("ACTIVE_AGENT:", row[0])
print("STATE:", row[1])

Expected output:

ROUTE: balance_lookup
ACTIVE_AGENT: balance_lookup
STATE: {"intent":"balance_check","step":"start","route_decision":"balance_lookup"}

Real-World Use Cases

• •

  Balance and transaction support

  • •One agent identifies intent.
  • •Another queries account data.
  • •PostgreSQL stores the case context and audit trail.
• •

  Fraud triage workflows

  • •A router detects suspicious activity.
  • •A fraud specialist agent collects evidence.
  • •Shared Postgres state keeps escalation details consistent across agents.
• •

  Loan prequalification assistants

  • •The intake agent gathers income and employment details.
  • •A scoring agent evaluates eligibility.
  • •PostgreSQL stores application progress so customers can resume later without starting over.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.