LangChain persistent memory in production: EU-hosted options compared

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory()
# This is a dict in RAM. Nothing is persisted anywhere.

from langchain.memory import VectorStoreRetrieverMemory
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# FAISS is in-memory — loses data on restart
vectorstore = FAISS.from_texts([""], embedding=OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs=dict(k=5))
memory = VectorStoreRetrieverMemory(retriever=retriever)
# Still in-process. Still not production-ready.

# DIY pgvector memory — simplified
import psycopg2
from openai import OpenAI

client = OpenAI()

def remember(user_id: str, content: str, conn):
    embedding = client.embeddings.create(
        input=content,
        model="text-embedding-3-small"
    ).data[0].embedding

    with conn.cursor() as cur:
        cur.execute(
            "INSERT INTO memories (user_id, content, embedding) VALUES (%s, %s, %s)",
            (user_id, content, embedding)
        )
    conn.commit()

def recall(user_id: str, query: str, conn, top_k: int = 5):
    query_embedding = client.embeddings.create(
        input=query,
        model="text-embedding-3-small"
    ).data[0].embedding

    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT content, 1 - (embedding <=> %s::vector) AS similarity
            FROM memories
            WHERE user_id = %s
            ORDER BY similarity DESC
            LIMIT %s
            """,
            (query_embedding, user_id, top_k)
        )
        return cur.fetchall()

# You also need: user isolation, GDPR endpoints, confidence scoring, API layer...

from langchain_community.memory import ZepChatMessageHistory
from langchain.memory import ConversationBufferMemory

# Requires your own Zep instance running on EU infra
history = ZepChatMessageHistory(
    session_id="user-123-session-456",
    url="https://your-zep.your-eu-domain.com",
    api_key="your-zep-key",
)

memory = ConversationBufferMemory(
    chat_memory=history,
    return_messages=True,
)

from kronvex import KronvexClient

kv = KronvexClient(api_key="kv-your-key", base_url="https://api.kronvex.io")

from langchain.memory import BaseChatMemory
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from kronvex import KronvexClient
from typing import Any

kv = KronvexClient(api_key="kv-your-key", base_url="https://api.kronvex.io")
AGENT_ID = "production-agent"


class KronvexMemory(BaseChatMemory):
    """
    Production LangChain memory backed by Kronvex.
    - Persists across process restarts and deploys
    - Isolated per user_id
    - GDPR-compliant (EU-hosted, Art. 17/20 endpoints available)
    """

    kv_client: Any
    agent_id: str
    user_id: str
    top_k: int = 5
    memory_key: str = "history"
    min_confidence: float = 0.4  # filter out low-confidence recalls

    class Config:
        arbitrary_types_allowed = True

    @property
    def memory_variables(self) -> list[str]:
        return [self.memory_key]

    def load_memory_variables(self, inputs: dict) -> dict:
        query = inputs.get("input", "")
        if not query:
            return {self.memory_key: "No prior context."}

        memories = self.kv_client.recall(
            agent_id=self.agent_id,
            query=query,
            user_id=self.user_id,
            top_k=self.top_k,
        )

        # Filter by confidence threshold
        relevant = [m for m in memories if m.get("confidence", 0) >= self.min_confidence]

        if not relevant:
            return {self.memory_key: "No relevant prior context found."}

        lines = [f"- {m['content']} [confidence: {m['confidence']:.2f}]" for m in relevant]
        return {self.memory_key: "\n".join(lines)}

    def save_context(self, inputs: dict, outputs: dict) -> None:
        human_input = inputs.get("input", "")
        ai_output = outputs.get("response", "")

        # Store what matters — not raw turns, but meaningful statements
        if human_input and len(human_input) > 20:  # skip trivial inputs
            self.kv_client.remember(
                agent_id=self.agent_id,
                content=human_input,
                user_id=self.user_id,
            )

    def clear(self) -> None:
        pass  # Use the GDPR erasure endpoint for intentional deletion


def create_agent_for_user(user_id: str) -> ConversationChain:
    memory = KronvexMemory(
        kv_client=kv,
        agent_id=AGENT_ID,
        user_id=user_id,
    )

    prompt = PromptTemplate(
        input_variables=["history", "input"],
        template=(
            "You are a helpful assistant. "
            "Use the context from past sessions to give consistent, personalized responses.\n\n"
            "Past session context:\n{history}\n\n"
            "Human: {input}\n"
            "Assistant:"
        ),
    )

    return ConversationChain(
        llm=ChatOpenAI(model="gpt-4o-mini"),
        memory=memory,
        prompt=prompt,
        verbose=False,
    )


# Usage — each user gets their own isolated memory
agent_alice = create_agent_for_user("user-alice")
agent_bob = create_agent_for_user("user-bob")

# Session 1 — alice
agent_alice.predict(input="I'm building a fintech app in Python using FastAPI.")

# Later — different process, same user_id
agent_alice_new_session = create_agent_for_user("user-alice")
response = agent_alice_new_session.predict(
    input="What framework should I use for the API layer?"
)
# Agent recalls alice's FastAPI + fintech context
print(response)

pip install kronvex

# Get a demo API key
curl -X POST https://api.kronvex.io/auth/demo