How to Integrate LangChain for fintech with PostgreSQL for production AI

Combining LangChain for fintech with PostgreSQL gives you a practical pattern for production AI agents: the model handles reasoning and tool use, while PostgreSQL stores customer context, transaction history, audit trails, and retrieval data. In fintech, that matters because you need deterministic state, queryable memory, and an audit-friendly backend instead of ephemeral chat history.

Prerequisites

• •Python 3.10+
• •A running PostgreSQL instance
• •A database user with read/write permissions
• •pip installed
• •Access to your LangChain-compatible LLM provider
• •These Python packages:
  • •langchain
  • •langchain-openai or your chosen provider package
  • •langchain-community
  • •psycopg2-binary
  • •sqlalchemy
• •A .env file or secret manager for:
  • •DATABASE_URL
  • •OPENAI_API_KEY or equivalent model key

Integration Steps

1. •Install the dependencies.

pip install langchain langchain-openai langchain-community psycopg2-binary sqlalchemy python-dotenv

2. •Create a PostgreSQL connection and schema for agent memory.

For production AI in fintech, keep the schema explicit. Don’t dump everything into one JSON column unless you enjoy debugging opaque data later.

import os
from dotenv import load_dotenv
from sqlalchemy import create_engine, text

load_dotenv()

DATABASE_URL = os.environ["DATABASE_URL"]
engine = create_engine(DATABASE_URL, pool_pre_ping=True)

with engine.begin() as conn:
    conn.execute(text("""
        CREATE TABLE IF NOT EXISTS agent_sessions (
            session_id TEXT PRIMARY KEY,
            customer_id TEXT NOT NULL,
            conversation_summary TEXT,
            updated_at TIMESTAMP DEFAULT NOW()
        )
    """))

    conn.execute(text("""
        CREATE TABLE IF NOT EXISTS agent_events (
            id SERIAL PRIMARY KEY,
            session_id TEXT NOT NULL,
            event_type TEXT NOT NULL,
            payload JSONB NOT NULL,
            created_at TIMESTAMP DEFAULT NOW()
        )
    """))

3. •Wire LangChain to PostgreSQL-backed retrieval using PGVector.

If you need semantic search over policy docs, KYC notes, or product FAQs, store embeddings in Postgres with PGVector. This keeps your retrieval layer inside the same operational boundary as your transactional data.

import os
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores.pgvector import PGVector

connection_string = os.environ["DATABASE_URL"]
collection_name = "fintech_knowledge_base"

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

vectorstore = PGVector(
    connection_string=connection_string,
    collection_name=collection_name,
    embedding_function=embeddings,
)

docs = [
    "KYC requires government-issued ID and proof of address.",
    "Suspicious activity thresholds should trigger manual review.",
    "Card chargeback disputes must be logged within the SLA window."
]

vectorstore.add_texts(docs)
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

4. •Build a LangChain agent that reads from PostgreSQL and writes audit events.

For fintech systems, every meaningful action should be persisted. The pattern below loads session context from Postgres, uses it in the prompt, then writes the outcome back as an auditable event.

import json
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a fintech support agent. Use retrieved context and keep responses concise."),
    ("human", "Customer question: {question}\n\nContext: {context}")
])

def load_session_context(session_id: str):
    with engine.begin() as conn:
        row = conn.execute(
            text("SELECT conversation_summary FROM agent_sessions WHERE session_id = :session_id"),
            {"session_id": session_id},
        ).fetchone()
    return row[0] if row else ""

def save_event(session_id: str, event_type: str, payload: dict):
    with engine.begin() as conn:
        conn.execute(
            text("""
                INSERT INTO agent_events (session_id, event_type, payload)
                VALUES (:session_id, :event_type, CAST(:payload AS jsonb))
            """),
            {
                "session_id": session_id,
                "event_type": event_type,
                "payload": json.dumps(payload),
            },
        )

def answer_customer(session_id: str, question: str):
    context_docs = retriever.invoke(question)
    context_text = "\n".join(doc.page_content for doc in context_docs)
    session_context = load_session_context(session_id)

    full_context = f"{session_context}\n{context_text}".strip()
    messages = prompt.format_messages(question=question, context=full_context)
    response = llm.invoke(messages)

    save_event(session_id, "assistant_response", {
        "question": question,
        "answer": response.content,
    })

    return response.content

print(answer_customer("sess_1001", "What documents do I need for account verification?"))

5. •Add a summary writer so long-running agents don’t bloat prompts.

Production agents need compact memory. Store summaries in PostgreSQL after each interaction so the next turn starts with useful state instead of raw chat logs.

def upsert_session_summary(session_id: str, customer_id: str, summary: str):
    with engine.begin() as conn:
        conn.execute(text("""
            INSERT INTO agent_sessions (session_id, customer_id, conversation_summary)
            VALUES (:session_id, :customer_id, :summary)
            ON CONFLICT (session_id)
            DO UPDATE SET conversation_summary = EXCLUDED.conversation_summary,
                          updated_at = NOW()
        """), {
            "session_id": session_id,
            "customer_id": customer_id,
            "summary": summary,
        })

summary_prompt = ChatPromptTemplate.from_messages([
    ("system", "Summarize this fintech support conversation in 5 lines max."),
    ("human", "{conversation}")
])

def summarize_and_store(session_id: str, customer_id: str, conversation: str):
    summary_messages = summary_prompt.format_messages(conversation=conversation)
    summary_response = llm.invoke(summary_messages)
    upsert_session_summary(session_id, customer_id, summary_response.content)

Testing the Integration

Run a simple retrieval-and-response test against both Postgres and LangChain:

test_question = "How do I handle suspicious transaction alerts?"
answer = answer_customer("sess_test_01", test_question)
print(answer)

with engine.begin() as conn:
    events = conn.execute(
        text("SELECT event_type, payload FROM agent_events WHERE session_id = :session_id ORDER BY created_at DESC LIMIT 1"),
        {"session_id": "sess_test_01"},
    ).fetchone()

print(events[0])
print(events[1])

Expected output:

Use manual review for suspicious alerts and log the case immediately.
assistant_response
{'question': 'How do I handle suspicious transaction alerts?', 'answer': '...'}

Real-World Use Cases

• •Customer support agents that answer banking questions using policy docs stored in PostgreSQL and persist every interaction for audit.
• •Fraud triage assistants that retrieve past incident patterns from vector search and write structured review events into Postgres.
• •Relationship manager copilots that summarize client conversations into durable session records and surface relevant account history on demand.

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.