Feather DB + OpenAI Agents SDK: Giving Your Agent Persistent Context

Setup

pip install feather-db openai openai-agents

This guide uses the OpenAI Agents SDK (the openai-agents package, not the plain openai package). The Agents SDK provides first-class tool definition and agent composition — exactly what we need to wire in Feather DB as a persistent memory backend.

Step 1: Initialize Feather DB

import os
import feather_db as fdb
from openai import OpenAI

openai_client = OpenAI()

# Shared memory file — all agents in this system share one .feather file
db = fdb.DB.open("shared_agent_memory.feather", dim=1536)  # text-embedding-3-small dim

def embed(text: str) -> list:
    """Embed text using OpenAI text-embedding-3-small."""
    response = openai_client.embeddings.create(
        input=[text],
        model="text-embedding-3-small"
    )
    return response.data[0].embedding

Step 2: Define memory tool functions

from datetime import datetime
from typing import Optional

def search_memory(query: str, namespace: str = "default",
                   k: int = 5, half_life: int = 30) -> str:
    """
    Search agent memory for relevant context.

    Args:
        query: The search query — what you're looking for
        namespace: Agent or user namespace to search within
        k: Number of results to return (default 5)
        half_life: How quickly memories decay in days (default 30)

    Returns:
        Formatted string of relevant memories with scores
    """
    vec = embed(query)
    results = db.context_chain(
        vec,
        k=k,
        namespace=namespace,
        max_depth=2,
        half_life=half_life
    )

    if not results:
        return "No relevant memories found."

    lines = [f"Found {len(results)} relevant memories:"]
    for i, mem in enumerate(results, 1):
        mem_type = mem.meta.get_attribute("type") or "fact"
        importance = mem.meta.get_attribute("importance") or 1.0
        lines.append(f"{i}. [{mem_type}] (score={mem.score:.3f}, importance={importance}) {mem.text}")

    return "\n".join(lines)

def add_memory(text: str, namespace: str = "default",
               memory_type: str = "fact", importance: float = 1.0,
               entity: str = "general") -> str:
    """
    Store a new memory for future retrieval.

    Args:
        text: The content to remember
        namespace: Namespace to store in (use agent_id or user_id)
        memory_type: Type tag (fact, decision, preference, observation)
        importance: Importance weight 0.1-3.0 (default 1.0)
        entity: Logical grouping within namespace (default 'general')

    Returns:
        Confirmation with the memory ID
    """
    vec = embed(text)
    mem = db.add(vec, text=text,
                  namespace=namespace,
                  entity=entity)
    mem.meta.set_attribute("type", memory_type)
    mem.meta.set_attribute("importance", importance)
    mem.meta.set_attribute("created_at", datetime.utcnow().isoformat())

    return f"Memory saved (id={mem.id}): {text[:80]}..."

def delete_memory(memory_id: int, namespace: str = "default") -> str:
    """
    Delete a specific memory by ID.

    Args:
        memory_id: The numeric ID of the memory to delete
        namespace: Namespace the memory belongs to

    Returns:
        Confirmation of deletion
    """
    db.delete(memory_id)
    return f"Memory {memory_id} deleted."

Step 3: Wire into the OpenAI Agents SDK

from agents import Agent, Runner, function_tool

# Wrap functions as Agents SDK tools
@function_tool
def search_agent_memory(query: str, namespace: str = "default",
                         k: int = 5) -> str:
    """Search persistent memory for context relevant to the query."""
    return search_memory(query, namespace=namespace, k=k)

@function_tool
def save_to_memory(text: str, memory_type: str = "fact",
                    importance: float = 1.0, entity: str = "general") -> str:
    """Save important information to persistent memory for future retrieval."""
    # Use a fixed namespace per agent — or pass it as a parameter
    return add_memory(text, namespace="research-agent",
                       memory_type=memory_type, importance=importance,
                       entity=entity)

# Create the agent with memory tools
research_agent = Agent(
    name="Research Agent",
    instructions="""You are a research agent with persistent memory.

At the START of every task:
1. Call search_agent_memory with the task description to retrieve relevant prior research
2. Use retrieved context to avoid re-doing work you've already done
3. Identify gaps: what do you still need to find out?

At the END of every task:
1. Call save_to_memory for each new fact, decision, or insight you discovered
2. Set importance=2.0 for key findings, 1.0 for supporting details
3. Set entity appropriately (e.g., 'ml-benchmarks', 'company-research', 'product-analysis')""",
    tools=[search_agent_memory, save_to_memory],
    model="gpt-4o"
)

Step 4: Run the agent with write-back

import asyncio

async def run_with_memory(task: str):
    """Run the research agent and ensure write-back completes."""
    print(f"Task: {task}")
    print("-" * 60)

    result = await Runner.run(
        research_agent,
        input=task,
        max_turns=10  # allow multi-turn tool use
    )

    print(f"\nResult: {result.final_output}")

    # Check what was saved
    check_vec = embed(task)
    saved = db.search(check_vec, k=5, namespace="research-agent")
    print(f"\nMemories now in store: {db.count(namespace='research-agent')}")
    for mem in saved:
        print(f"  - {mem.text[:80]}")

    return result.final_output

async def main():
    # First run — agent researches and saves findings
    await run_with_memory(
        "Research the current state of open-source embedding models in 2026. "
        "Focus on recall benchmarks and deployment cost."
    )

    print("\n" + "="*60)
    print("Second run — agent retrieves prior research")
    print("="*60 + "\n")

    # Second run — agent recalls prior research and builds on it
    await run_with_memory(
        "Based on what you know about embedding models, "
        "which model would you recommend for a budget-conscious deployment "
        "processing 10M tokens per day?"
    )

asyncio.run(main())

Step 5: Multi-agent memory sharing

# Agent A: research specialist — writes findings to shared namespace
research_specialist = Agent(
    name="Research Specialist",
    instructions="Research topics deeply and save all findings to memory.",
    tools=[
        function_tool(lambda q: search_memory(q, namespace="shared-project")),
        function_tool(lambda t, tp="fact", imp=1.0: add_memory(
            t, namespace="shared-project", memory_type=tp, importance=imp
        ))
    ],
    model="gpt-4o"
)

# Agent B: synthesis specialist — reads from the same shared namespace
synthesis_specialist = Agent(
    name="Synthesis Specialist",
    instructions="Synthesize research findings from memory into clear recommendations.",
    tools=[
        function_tool(lambda q: search_memory(q, namespace="shared-project", k=10))
    ],
    model="gpt-4o-mini"  # cheaper model for synthesis
)

async def multi_agent_pipeline(research_task: str, synthesis_task: str):
    """Two-agent pipeline: research then synthesize via shared memory."""
    # Step 1: Research agent fills memory
    print("[Phase 1: Research]")
    await Runner.run(research_specialist, input=research_task)

    # Step 2: Synthesis agent reads from the same .feather file
    print("[Phase 2: Synthesis]")
    result = await Runner.run(synthesis_specialist, input=synthesis_task)
    return result.final_output

asyncio.run(multi_agent_pipeline(
    research_task="Research Feather DB's HNSW benchmarks and compare to Pinecone and Weaviate.",
    synthesis_task="Based on the research in memory, write a 3-bullet comparison for a CTO audience."
))

The shared .feather file is the coordination primitive. Agent A writes, Agent B reads — no message passing, no shared state server, no Redis. Just a file that both agents open with fdb.DB.open(). For concurrent write safety in multi-process deployments, use feather-serve as a single server and hit its REST API from both agents instead of opening the file directly.

Install: pip install feather-db openai openai-agents · GitHub: github.com/feather-store/feather