Feather DB + LangChain: Add Persistent Memory to Your LangChain App

LangChain's default memory options (ConversationBufferMemory, ConversationSummaryMemory) do not persist between Python sessions and do not support adaptive decay or graph traversal. Feather DB integrates with LangChain as a custom memory class, providing sub-millisecond retrieval, temporal decay, and persistent storage that survives process restarts. This guide shows how to implement a FeatherDBMemory class, connect it to LangChain agents and chains, and use it in production.

Prerequisites

pip install feather-db langchain langchain-openai openai

export OPENAI_API_KEY="your-key-here"

Why Replace LangChain's Default Memory

LangChain ships several memory classes:

• ConversationBufferMemory: Stores the full conversation. Runs out of context window quickly. Not persistent between sessions.
• ConversationSummaryMemory: Summarizes old turns to save tokens. Loses specific facts in the summary. Not persistent between sessions.
• ConversationBufferWindowMemory: Sliding window. Drops old turns. Not persistent between sessions.
• VectorStoreRetrieverMemory: Uses a vector store for retrieval. Persistent if you use a persistent store. No adaptive decay, no graph traversal.

Feather DB replaces VectorStoreRetrieverMemory with a class that adds adaptive temporal decay, graph context traversal, and an MIT-licensed embedded backend with 0.19ms p50 retrieval.

Step 1: Implement FeatherDBMemory

import feather_db as fdb
from langchain.memory.chat_memory import BaseChatMemory
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
from langchain_core.outputs import LLMResult
from typing import Any, Dict, List, Optional
from datetime import datetime

class FeatherDBMemory(BaseChatMemory):
    """
    LangChain memory backed by Feather DB.
    Provides persistent, cross-session memory with adaptive decay
    and graph context traversal.

    Features:
    - Persists between Python sessions (stored in .feather file)
    - Adaptive temporal decay (memories fade unless recalled)
    - Graph traversal for related context (context_chain)
    - 0.19ms p50 retrieval latency
    - MIT license, no cloud dependency
    """

    db_path: str = "langchain_memory.feather"
    user_id: str = "default_user"
    top_k: int = 8
    use_graph: bool = True
    graph_hops: int = 2
    memory_key: str = "chat_history"
    return_messages: bool = True

    class Config:
        arbitrary_types_allowed = True

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._db = fdb.FeatherDB(self.db_path)

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

    def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        """Load relevant memories based on the current input."""
        # Extract the current query from inputs
        query = ""
        for key in ["input", "question", "human_input", "query"]:
            if key in inputs:
                query = str(inputs[key])
                break

        if not query:
            return {self.memory_key: []}

        # Retrieve relevant memories
        if self.use_graph:
            memories = self._db.context_chain(
                query=query,
                hops=self.graph_hops,
                top_k=self.top_k,
                filters={"user_id": self.user_id}
            )
        else:
            memories = self._db.search(
                query=query,
                top_k=self.top_k,
                filters={"user_id": self.user_id}
            )

        # Format as LangChain messages
        messages = []
        for memory in memories:
            # Reconstruct as human/AI message pairs where possible
            text = memory['text']
            role = memory.get('metadata', {}).get('role', 'human')
            if role == 'human':
                messages.append(HumanMessage(content=text))
            else:
                messages.append(AIMessage(content=text))

        return {self.memory_key: messages}

    def save_context(
        self, inputs: Dict[str, Any], outputs: Dict[str, str]
    ) -> None:
        """Save a conversation turn to Feather DB."""
        # Save human input
        human_text = ""
        for key in ["input", "question", "human_input"]:
            if key in inputs:
                human_text = str(inputs[key])
                break

        if human_text:
            self._db.add(
                text=human_text,
                metadata={
                    "user_id": self.user_id,
                    "role": "human",
                    "stored_at": datetime.now().isoformat()
                },
                importance=0.5
            )

        # Save AI output
        ai_text = ""
        for key in ["output", "answer", "response", "text"]:
            if key in outputs:
                ai_text = str(outputs[key])
                break

        if ai_text:
            self._db.add(
                text=ai_text,
                metadata={
                    "user_id": self.user_id,
                    "role": "ai",
                    "stored_at": datetime.now().isoformat()
                },
                importance=0.5
            )

    def clear(self) -> None:
        """Clear all memories for this user."""
        self._db.delete_by_filter({"user_id": self.user_id})

    @property
    def total_memories(self) -> int:
        """Return total memory count for this user."""
        return self._db.count(filters={"user_id": self.user_id})

Step 2: Use FeatherDBMemory with a LangChain Conversation Chain

from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

# Initialize Feather DB memory
memory = FeatherDBMemory(
    db_path="my_agent_memory.feather",
    user_id="user_priya",
    top_k=8,
    use_graph=True
)

# Build the chat prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant with memory of past conversations. "
               "Use the provided context to give personalized responses."),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}")
])

# Create the chain
llm = ChatOpenAI(model="gpt-4o", temperature=0)
chain = ConversationChain(
    llm=llm,
    memory=memory,
    prompt=prompt,
    verbose=True
)

# Use the chain — memory persists between Python sessions
response1 = chain.predict(input="My name is Priya and I build fintech apps")
print(response1)

response2 = chain.predict(input="What should I consider for PCI compliance?")
print(response2)

print(f"Total memories stored: {memory.total_memories}")

Step 3: Use FeatherDBMemory with a LangChain Agent

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.tools import tool
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

# Define tools for the agent
@tool
def search_web(query: str) -> str:
    """Search the web for current information."""
    # In production, connect to a real search API
    return f"Search results for: {query} (placeholder)"

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression."""
    try:
        return str(eval(expression))  # Use numexpr in production
    except Exception as e:
        return f"Error: {e}"

# Initialize Feather DB memory
memory = FeatherDBMemory(
    db_path="agent_memory.feather",
    user_id="user_001",
    top_k=10,
    use_graph=True,
    graph_hops=2
)

# Agent prompt with memory slot
agent_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful AI assistant with tools and persistent memory."
               "Use past context to give personalized responses."),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad")
])

llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [search_web, calculate]

agent = create_openai_tools_agent(llm, tools, agent_prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True
)

# Run the agent — memories persist across restarts
result = agent_executor.invoke({"input": "What is 15% of 8,500?"})
print(result["output"])

Step 4: Advanced Memory — Store Semantic Facts Separately

For better retrieval quality, store extracted semantic facts alongside raw conversation turns:

import feather_db as fdb
from openai import OpenAI
import json

db = fdb.FeatherDB("agent_memory.feather")
openai_client = OpenAI()

class EnhancedFeatherDBMemory(FeatherDBMemory):
    """FeatherDBMemory with automatic fact extraction."""

    extract_facts: bool = True
    extraction_model: str = "gpt-4o-mini"  # Cheap model for extraction

    def save_context(
        self, inputs: Dict[str, Any], outputs: Dict[str, str]
    ) -> None:
        """Save context and optionally extract semantic facts."""
        # Save raw conversation turn (inherited)
        super().save_context(inputs, outputs)

        if not self.extract_facts:
            return

        # Extract memorable facts from this turn
        human_text = inputs.get("input", "")
        ai_text = outputs.get("output", "")
        turn_text = f"User: {human_text}\nAssistant: {ai_text}"

        extraction_response = openai_client.chat.completions.create(
            model=self.extraction_model,
            messages=[{
                "role": "user",
                "content": f"""Extract memorable facts from this conversation.
Return JSON: {{"facts": [{{"text": "...", "importance": 0.0-1.0}}]}}
Only extract genuinely useful facts to remember.
Conversation: {turn_text}"""
            }],
            response_format={"type": "json_object"}
        )

        result = json.loads(extraction_response.choices[0].message.content)
        for fact in result.get("facts", []):
            self._db.add(
                text=fact["text"],
                metadata={
                    "user_id": self.user_id,
                    "type": "extracted_fact",
                    "stored_at": datetime.now().isoformat()
                },
                importance=fact.get("importance", 0.6),
                half_life_days=180  # Facts last longer than raw turns
            )

# Usage:
enhanced_memory = EnhancedFeatherDBMemory(
    db_path="enhanced_memory.feather",
    user_id="user_001",
    extract_facts=True,
    extraction_model="gpt-4o-mini"
)

Step 5: Test Persistence Across Sessions

# Session 1 (first Python process):
memory = FeatherDBMemory(db_path="test.feather", user_id="test_user")
chain = ConversationChain(llm=ChatOpenAI(), memory=memory)
chain.predict(input="I work at a startup called Kapital that does B2B SaaS")
chain.predict(input="We use FastAPI and PostgreSQL")
print(f"Session 1 ended. Memories: {memory.total_memories}")
# Session 1 ended. Memories: 4

# Session 2 (new Python process, same .feather file):
memory = FeatherDBMemory(db_path="test.feather", user_id="test_user")
chain = ConversationChain(llm=ChatOpenAI(), memory=memory)
print(f"Session 2 started. Memories loaded: {memory.total_memories}")
# Session 2 started. Memories loaded: 4
chain.predict(input="What's my tech stack?")
# Agent responds: Based on what you told me, you use FastAPI and PostgreSQL at Kapital.

Performance Comparison: Default LangChain Memory vs FeatherDBMemory

FAQ

Does FeatherDBMemory work with LangChain's LCEL (LangChain Expression Language)?

Yes. FeatherDBMemory implements BaseChatMemory, which is compatible with LCEL chains using RunnableWithMessageHistory. Use it as the history provider in any LCEL chain that accepts a message history.

Can I use FeatherDBMemory with LangGraph?

Yes. In LangGraph, pass FeatherDBMemory as the checkpointer's memory store, or use it as a tool within the graph's memory node. LangGraph's state management is orthogonal to Feather DB's persistence — they solve different layers of the problem.

How does FeatherDBMemory compare to using ChromaDB as a VectorStoreRetrieverMemory?

Both provide persistent retrieval-based memory. Feather DB adds adaptive temporal decay, graph traversal, hybrid BM25+dense search, and 5–10x lower retrieval latency (0.19ms vs 2–20ms for ChromaDB). The API surface is similar — FeatherDBMemory is a drop-in upgrade.

What happens to memory if the .feather file is corrupted?

Feather DB uses atomic writes — each write operation completes fully before being visible. A crash mid-write rolls back to the last clean state. The file cannot be partially written in a way that corrupts the index. For critical applications, periodically copy the .feather file as a backup.

Is FeatherDBMemory thread-safe for concurrent LangChain calls?

Yes. Feather DB's reader-writer locking model allows concurrent reads (multiple LangChain chains querying simultaneously) and serializes writes. For high-concurrency applications (100+ concurrent chains), reads are non-blocking and writes queue with sub-millisecond wait times.