Persisting LangChain memory to Postgres

LangChain memory postgres persistence solves a fundamental problem: conversation history vanishes when your process restarts. The built-in memory classes keep state in memory, which works for notebooks but fails in production. This guide walks through building a Postgres-backed memory implementation you can drop into any chain or agent.

Step 1: Install dependencies and configure the database

Start with a clean environment. You need langchain-core, langchain-community, and a Postgres driver. asyncpg is the standard choice for async workloads; psycopg2 works for synchronous code.

pip install langchain-core langchain-community asyncpg psycopg2-binary

Create a database and a dedicated user. Avoid running as postgres in production.

CREATE DATABASE langchain_memory;
CREATE USER langchain_app WITH PASSWORD 'your_secure_password';
GRANT ALL PRIVILEGES ON DATABASE langchain_memory TO langchain_app;

Set the connection string as an environment variable. Never hardcode credentials.

export DATABASE_URL="postgresql://langchain_app:your_secure_password@localhost:5432/langchain_memory"

Step 2: Design the schema

LangChain’s BaseChatMessageHistory expects a sequence of messages with roles (human, ai, system, tool). A minimal schema needs conversation identity, message ordering, and the message payload itself.

-- Run this once via psql or your migration tool
CREATE TABLE IF NOT EXISTS conversation_messages (
    id BIGSERIAL PRIMARY KEY,
    conversation_id UUID NOT NULL,
    message_index INTEGER NOT NULL,
    role VARCHAR(32) NOT NULL,
    content TEXT NOT NULL,
    additional_kwargs JSONB DEFAULT '{}'::jsonb,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX IF NOT EXISTS idx_conversation_messages_lookup
    ON conversation_messages (conversation_id, message_index);

-- Optional: a conversations table for metadata, TTL, or ownership
CREATE TABLE IF NOT EXISTS conversations (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id UUID,
    metadata JSONB DEFAULT '{}'::jsonb,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Trigger to keep updated_at fresh
CREATE OR REPLACE FUNCTION update_updated_at_column()
RETURNS TRIGGER LANGUAGE plpgsql AS $$
BEGIN
    NEW.updated_at = now();
    RETURN NEW;
END;
$$;

CREATE TRIGGER update_conversations_updated_at
    BEFORE UPDATE ON conversations
    FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();

The message_index column guarantees ordering without relying on timestamps, which can collide under high throughput. The additional_kwargs JSONB column stores tool call metadata, function names, or any provider-specific fields LangChain attaches to messages.

Step 3: Implement a Postgres-backed chat message history

Subclass BaseChatMessageHistory from langchain_core.chat_history. Implement the four required methods: messages (getter), add_message, clear, and the async variants if you use asyncpg.

# postgres_memory.py
import os
import uuid
from typing import List, Optional
from contextlib import asynccontextmanager

import asyncpg
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import (
    BaseMessage,
    HumanMessage,
    AIMessage,
    SystemMessage,
    ToolMessage,
    message_to_dict,
    messages_from_dict,
)

ROLE_MAP = {
    "human": HumanMessage,
    "ai": AIMessage,
    "system": SystemMessage,
    "tool": ToolMessage,
}

class PostgresChatMessageHistory(BaseChatMessageHistory):
    def __init__(
        self,
        conversation_id: uuid.UUID,
        pool: asyncpg.Pool,
        *,
        max_messages: Optional[int] = None,
    ):
        self.conversation_id = conversation_id
        self.pool = pool
        self.max_messages = max_messages

    @classmethod
    async def create_pool(cls, dsn: str, **pool_kwargs) -> asyncpg.Pool:
        return await asyncpg.create_pool(dsn, **pool_kwargs)

    def _row_to_message(self, row: asyncpg.Record) -> BaseMessage:
        msg_class = ROLE_MAP.get(row["role"])
        if msg_class is None:
            raise ValueError(f"Unknown role: {row['role']}")
        data = {"type": row["role"], "content": row["content"]}
        if row["additional_kwargs"]:
            data["additional_kwargs"] = row["additional_kwargs"]
        return messages_from_dict([data])[0]

    async def _fetch_messages(self) -> List[BaseMessage]:
        async with self.pool.acquire() as conn:
            rows = await conn.fetch(
                """
                SELECT role, content, additional_kwargs
                FROM conversation_messages
                WHERE conversation_id = $1
                ORDER BY message_index
                """,
                self.conversation_id,
            )
        return [self._row_to_message(r) for r in rows]

    @property
    def messages(self) -> List[BaseMessage]:
        # Synchronous property required by BaseChatMessageHistory
        # In async contexts, use `await self.aget_messages()` instead
        import asyncio
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = None
        if loop and loop.is_running():
            raise RuntimeError(
                "Accessing .messages from a running event loop. "
                "Use `await aget_messages()` in async code."
            )
        return asyncio.run(self._fetch_messages())

    async def aget_messages(self) -> List[BaseMessage]:
        return await self._fetch_messages()

    async def _get_next_index(self, conn: asyncpg.Connection) -> int:
        row = await conn.fetchrow(
            "SELECT COALESCE(MAX(message_index), -1) + 1 AS next_idx "
            "FROM conversation_messages WHERE conversation_id = $1",
            self.conversation_id,
        )
        return row["next_idx"]

    async def _enforce_max_messages(self, conn: asyncpg.Connection, next_index: int):
        if self.max_messages is None:
            return
        # Delete oldest messages beyond the window
        await conn.execute(
            """
            DELETE FROM conversation_messages
            WHERE conversation_id = $1
              AND message_index < $2
            """,
            self.conversation_id,
            next_index - self.max_messages,
        )

    def add_message(self, message: BaseMessage) -> None:
        import asyncio
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = None
        if loop and loop.is_running():
            raise RuntimeError(
                "Calling add_message from a running event loop. "
                "Use `await aadd_message()` in async code."
            )
        asyncio.run(self.aadd_message(message))

    async def aadd_message(self, message: BaseMessage) -> None:
        msg_dict = message_to_dict(message)
        role = msg_dict["type"]
        content = msg_dict["content"]
        additional_kwargs = msg_dict.get("additional_kwargs", {})

        async with self.pool.acquire() as conn:
            async with conn.transaction():
                next_index = await self._get_next_index(conn)
                await self._enforce_max_messages(conn, next_index)
                await conn.execute(
                    """
                    INSERT INTO conversation_messages
                        (conversation_id, message_index, role, content, additional_kwargs)
                    VALUES ($1, $2, $3, $4, $5)
                    """,
                    self.conversation_id,
                    next_index,
                    role,
                    content,
                    additional_kwargs,
                )

    def clear(self) -> None:
        import asyncio
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = None
        if loop and loop.is_running():
            raise RuntimeError(
                "Calling clear from a running event loop. "
                "Use `await aclear()` in async code."
            )
        asyncio.run(self.aclear())

    async def aclear(self) -> None:
        async with self.pool.acquire() as conn:
            await conn.execute(
                "DELETE FROM conversation_messages WHERE conversation_id = $1",
                self.conversation_id,
            )

Key design decisions in this implementation:

• Connection pooling: The class accepts an asyncpg.Pool, not a raw connection. Create the pool once at startup and share it.
• Dual sync/async API: LangChain’s base class defines synchronous messages, add_message, and clear. The implementation bridges to async methods but raises a clear error if you call the sync version inside a running event loop. In async code, always use aget_messages, aadd_message, aclear.
• Windowed history: The optional max_messages parameter implements a sliding window, deleting the oldest messages on each insert. This prevents unbounded growth without a separate cleanup job.
• Transaction safety: Each aadd_message runs in its own transaction, so a failure mid-write leaves history consistent.

Step 4: Wire it into a chain or agent

LangChain’s RunnableWithMessageHistory wraps any runnable and injects history automatically. You provide a factory function that returns a BaseChatMessageHistory instance for a given session ID.

# main.py
import os
import uuid
import asyncio
from contextlib import asynccontextmanager

import asyncpg
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI

from postgres_memory import PostgresChatMessageHistory

DATABASE_URL = os.environ["DATABASE_URL"]
OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]

@asynccontextmanager
async def lifespan():
    pool = await PostgresChatMessageHistory.create_pool(
        DATABASE_URL,
        min_size=2,
        max_size=10,
    )
    try:
        yield pool
    finally:
        await pool.close()

def get_session_history(session_id: str) -> PostgresChatMessageHistory:
    # In a real app, validate session_id format, check ownership, etc.
    return PostgresChatMessageHistory(
        conversation_id=uuid.UUID(session_id),
        pool=pool,  # captured from lifespan
        max_messages=20,  # keep last 20 messages
    )

async def main():
    async with lifespan() as pool:
        global pool  # hack for demo; use dependency injection in real code

        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a concise assistant."),
            MessagesPlaceholder(variable_name="history"),
            ("human", "{input}"),
        ])

        llm = ChatOpenAI(model="gpt-4o-mini", api_key=OPENAI_API_KEY)
        chain = prompt | llm

        with_history = RunnableWithMessageHistory(
            chain,
            get_session_history,
            input_messages_key="input",
            history_messages_key="history",
        )

        session_id = str(uuid.uuid4())
        config = {"configurable": {"session_id": session_id}}

        # Turn 1
        resp1 = await with_history.ainvoke({"input": "My name is Alex."}, config=config)
        print(f"AI: {resp1.content}")

        # Turn 2 — history should be injected
        resp2 = await with_history.ainvoke({"input": "What's my name?"}, config=config)
        print(f"AI: {resp2.content}")

        # Verify persistence by creating a new chain instance
        with_history2 = RunnableWithMessageHistory(
            chain,
            get_session_history,
            input_messages_key="input",
            history_messages_key="history",
        )
        resp3 = await with_history2.ainvoke({"input": "Remind me, what did I say my name was?"}, config=config)
        print(f"AI (new chain instance): {resp3.content}")

if __name__ == "__main__":
    asyncio.run(main())

Run it:

python main.py

You should see the model recall “Alex” in the second and third turns, even though the third turn uses a brand-new RunnableWithMessageHistory instance. That proves the history survived in Postgres.

Step 5: Handle concurrent access correctly

If two requests for the same conversation_id arrive simultaneously, both may read the same message_index max, then both try to insert at max + 1. One will succeed; the other gets a unique constraint violation on (conversation_id, message_index).

Add a unique index and retry logic:

ALTER TABLE conversation_messages
ADD CONSTRAINT uq_conversation_message_index
UNIQUE (conversation_id, message_index);

Update aadd_message to retry on serialization failures:

import asyncpg

MAX_RETRIES = 3

async def aadd_message(self, message: BaseMessage) -> None:
    msg_dict = message_to_dict(message)
    role = msg_dict["type"]
    content = msg_dict["content"]
    additional_kwargs = msg_dict.get("additional_kwargs", {})

    for attempt in range(MAX_RETRIES):
        async with self.pool.acquire() as conn:
            async with conn.transaction():
                try:
                    next_index = await self._get_next_index(conn)
                    await self._enforce_max_messages(conn, next_index)
                    await conn.execute(
                        """
                        INSERT INTO conversation_messages
                            (conversation_id, message_index, role, content, additional_kwargs)
                        VALUES ($1, $2, $3, $4, $5)
                        """,
                        self.conversation_id,
                        next_index,
                        role,
                        content,
                        additional_kwargs,
                    )
                    return  # success
                except asyncpg.UniqueViolationError:
                    if attempt == MAX_RETRIES - 1:
                        raise
                    # brief backoff before retry
                    await asyncio.sleep(0.01 * (attempt + 1))

The unique constraint turns a race condition into a retryable error. With max_messages set, the window stays bounded even under contention.

Step 6: Add TTL and cleanup for production

Long-running applications accumulate abandoned conversations. Add a background job that deletes conversations older than your retention window.

# cleanup.py
import asyncio
import asyncpg
import os

DATABASE_URL = os.environ["DATABASE_URL"]
RETENTION_DAYS = 30

async def cleanup_old_conversations(pool: asyncpg.Pool):
    async with pool.acquire() as conn:
        # Delete messages for conversations older than retention window
        await conn.execute(
            """
            DELETE FROM conversation_messages
            WHERE conversation_id IN (
                SELECT id FROM conversations
                WHERE updated_at < now() - $1::interval
            )
            """,
            f"{RETENTION_DAYS} days",
        )
        # Delete the conversation rows themselves
        await conn.execute(
            "DELETE FROM conversations WHERE updated_at < now() - $1::interval",
            f"{RETENTION_DAYS} days",
        )

async def run_cleanup_loop():
    pool = await asyncpg.create_pool(DATABASE_URL, min_size=1, max_size=2)
    while True:
        await cleanup_old_conversations(pool)
        await asyncio.sleep(3600)  # run hourly

if __name__ == "__main__":
    asyncio.run(run_cleanup_loop())

Deploy this as a separate process or a cron job. Keep the retention policy aligned with your data governance requirements.

Step 7: Verify success with direct database inspection

After running the example, inspect the tables directly to confirm the shape of stored data.

psql "$DATABASE_URL" -c "
SELECT
    cm.conversation_id,
    cm.message_index,
    cm.role,
    left(cm.content, 60) AS content_preview,
    cm.additional_kwargs
FROM conversation_messages cm
ORDER BY cm.conversation_id, cm.message_index;
"

Expected output for the three-turn conversation:

 conversation_id | message_index | role | content_preview                    | additional_kwargs
-----------------+---------------+------+----------------------------------+-------------------
 550e8400-e29b...| 0             | human| My name is Alex.                 | {}
 550e8400-e29b...| 1             | ai   | Nice to meet you, Alex!          | {}
 550e8400-e29b...| 2             | human| What's my name?                  | {}
 550e8400-e29b...| 3             | ai   | Your name is Alex.               | {}
 550e8400-e29b...| 4             | human| Remind me, what did I say...     | {}
 550e8400-e29b...| 5             | ai   | You said your name was Alex.     | {}

Check the conversations table for metadata:

psql "$DATABASE_URL" -c "SELECT id, user_id, metadata, created_at, updated_at FROM conversations;"

If you see rows with incrementing message_index, correct roles, and intact additional_kwargs, persistence works.

Step 8: Extend for tool calls and structured output

LangChain tool messages carry tool_call_id and name in additional_kwargs. The schema already supports this via JSONB. When using OpenAI function calling or Anthropic tool use, the message dict includes:

{
    "type": "ai",
    "content": "",
    "additional_kwargs": {
        "tool_calls": [
            {"id": "call_abc123", "function": {"name": "get_weather", "arguments": "{\"city\": \"SF\"}"}}
        ]
    }
}

The ToolMessage response includes the matching tool_call_id:

{
    "type": "tool",
    "content": "{\"temp\": 58, \"conditions\": \"foggy\"}",
    "additional_kwargs": {"tool_call_id": "call_abc123", "name": "get_weather"}
}

No schema changes required. The JSONB column stores whatever the provider returns. When you retrieve messages via messages_from_dict, LangChain reconstructs the proper AIMessage with tool_calls and ToolMessage objects.

Production checklist

Before deploying this to serve real traffic:

When to use this vs. managed memory services

This implementation gives you full control: schema ownership, query tuning, data residency, and zero vendor lock-in. It adds operational burden — you run the database, handle migrations, and own availability.

Managed alternatives (LangChain’s PostgresChatMessageHistory in langchain-postgres, Upstash Redis, Momento) trade control for convenience. Use this custom version when:

• You need custom columns (tenant ID, classification labels, encryption keys per conversation)
• You already operate Postgres and want to avoid another dependency
• You require specific transaction semantics or row-level security policies

Otherwise, langchain-postgres provides a maintained, tested implementation with the same interface.

The pattern here — BaseChatMessageHistory subclass + RunnableWithMessageHistory — works for any storage backend. Swap asyncpg for redis, asyncmy, or an HTTP call to a remote service; the chain integration stays identical.