Wiring Feather DB into an Agent Loop: Read, Reason, Update, Decay

The four-phase agent memory loop

Every memory-backed agent runs the same loop, regardless of its task. The specific tools, LLM, and objectives vary — but the memory interaction has four invariant phases:

READ: At the start of each turn, retrieve relevant memories to ground the LLM's reasoning
REASON: Pass retrieved context alongside the current query to the LLM; let it reason with the grounding
UPDATE: After the LLM responds, write new knowledge back to memory
DECAY: Let the scoring formula handle passive decay; selectively remove stale facts explicitly

Getting this loop right is more important than the choice of embedding model or the specific HNSW parameters. A wrong READ (irrelevant context) leads to hallucination. A missing UPDATE (no write-back) means the agent never learns. A missing DECAY pattern means stale facts accumulate and eventually crowd out current knowledge.

Phase 1: READ — context retrieval at turn start

import feather_db as fdb
from anthropic import Anthropic
import requests
import json
from datetime import datetime

db = fdb.DB.open("research_agent.feather", dim=768)
client = Anthropic()

def read_context(query: str, agent_id: str, k: int = 8) -> list:
    """
    Phase 1: READ
    Retrieve relevant memories using context_chain for full graph traversal.
    Returns a list of memory objects ranked by adaptive score.
    """
    vec = embed(query)

    # context_chain: ANN seeds + BFS traversal of edges
    # This surfaces not just direct matches but connected knowledge
    chain = db.context_chain(
        vec,
        k=k,
        namespace=agent_id,
        max_depth=2,
        half_life=30   # web research: 30-day half_life, news fades fast
    )
    return chain

def format_context(memories: list) -> str:
    """Format retrieved memories for the LLM system prompt."""
    if not memories:
        return "No prior research on this topic."

    lines = []
    for mem in memories:
        age_hint = ""
        created = mem.meta.get_attribute("created_at")
        if created:
            from datetime import datetime
            try:
                age_days = (datetime.utcnow() -
                            datetime.fromisoformat(created)).days
                age_hint = f" [{age_days}d ago]"
            except Exception:
                pass
        mem_type = mem.meta.get_attribute("type") or "fact"
        lines.append(f"- [{mem_type}{age_hint}] {mem.text}")

    return "\n".join(lines)

Phase 2: REASON — grounded LLM call

def reason(query: str, context: str, tools: list = None) -> tuple:
    """
    Phase 2: REASON
    Call the LLM with retrieved context and available tools.
    Returns (response_text, tool_calls).
    """
    system = f"""You are a web research agent with persistent memory.

What you already know about this topic:
{context}

Instructions:
- Use your existing knowledge to avoid re-researching what you already know
- Identify gaps: what do you NOT know that would answer the query?
- Use search tools to fill those gaps
- After reasoning, explicitly state new facts you've learned"""

    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=2000,
        system=system,
        messages=[{"role": "user", "content": query}],
        tools=tools or []
    )

    text = ""
    tool_calls = []
    for block in response.content:
        if block.type == "text":
            text += block.text
        elif block.type == "tool_use":
            tool_calls.append(block)

    return text, tool_calls

Phase 3: UPDATE — write-back new knowledge

EXTRACT_FACTS_PROMPT = """Extract new factual claims from this research response.
Output one JSON line per fact: {{"text": "...", "type": "fact|source|hypothesis", "confidence": 0.0-1.0}}
Only include facts that are concrete, verifiable, and not already common knowledge.
Response: {response}"""

def update_memory(response_text: str, query: str, agent_id: str) -> list:
    """
    Phase 3: UPDATE
    Extract new facts from the LLM response and write them to memory.
    """
    extract_response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=500,
        messages=[{"role": "user",
                    "content": EXTRACT_FACTS_PROMPT.format(response=response_text[:1000])}]
    )

    saved = []
    for line in extract_response.content[0].text.strip().split("\n"):
        line = line.strip()
        if not line:
            continue
        try:
            fact_data = json.loads(line)
            text = fact_data.get("text", "")
            fact_type = fact_data.get("type", "fact")
            confidence = float(fact_data.get("confidence", 0.7))

            if len(text) < 20 or confidence < 0.5:
                continue

            vec = embed(text)
            mem = db.add(vec, text=text,
                          namespace=agent_id,
                          entity="research-facts")
            mem.meta.set_attribute("type", fact_type)
            mem.meta.set_attribute("confidence", confidence)
            mem.meta.set_attribute("importance", confidence * 1.5)
            mem.meta.set_attribute("query_source", query[:100])
            mem.meta.set_attribute("created_at", datetime.utcnow().isoformat())
            saved.append(mem)
        except (json.JSONDecodeError, ValueError):
            continue

    return saved

Phase 4: DECAY — stale fact management

def decay_stale_facts(agent_id: str, max_age_days: int = 60,
                       confidence_threshold: float = 0.6):
    """
    Phase 4: DECAY
    Passive decay is handled by the scoring formula automatically.
    This explicit sweep removes facts that are both old AND low-confidence.
    Run periodically (e.g., once per day) not on every turn.
    """
    # Get all research facts
    # Use a broad semantic search to retrieve most facts
    all_facts_vec = embed("research fact information finding")
    all_facts = db.search(all_facts_vec, k=1000,
                           namespace=agent_id,
                           entity="research-facts",
                           half_life=1  # tiny half_life to surface oldest
                           )

    removed = 0
    for mem in all_facts:
        created = mem.meta.get_attribute("created_at")
        confidence = float(mem.meta.get_attribute("confidence") or 0.7)
        recall_count = mem.recall_count

        if not created:
            continue

        try:
            age_days = (datetime.utcnow() -
                        datetime.fromisoformat(created)).days
        except Exception:
            continue

        # Remove if: old, low confidence, and barely recalled
        if (age_days > max_age_days and
            confidence < confidence_threshold and
            recall_count < 2):
            db.delete(mem.id)
            removed += 1

    return removed

The complete agent loop

SEARCH_TOOL = {
    "name": "web_search",
    "description": "Search the web for current information.",
    "input_schema": {
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "Search query"}
        },
        "required": ["query"]
    }
}

def fake_web_search(query: str) -> str:
    """Placeholder — replace with real search API."""
    return f"[Search result for '{query}']: Latest findings indicate..."

def run_research_agent(user_query: str, agent_id: str = "research-agent-1") -> str:
    """Complete agent loop: READ -> REASON -> UPDATE -> (periodic) DECAY."""
    print(f"[READ] Retrieving context for: {user_query[:50]}...")
    memories = read_context(user_query, agent_id)
    context = format_context(memories)
    print(f"[READ] Retrieved {len(memories)} memories")

    print("[REASON] Calling LLM with context...")
    response_text, tool_calls = reason(user_query, context, tools=[SEARCH_TOOL])

    # Handle tool calls
    search_results = ""
    for tc in tool_calls:
        if tc.name == "web_search":
            result = fake_web_search(tc.input["query"])
            search_results += result + "\n"
            print(f"[REASON] Tool: web_search('{tc.input['query'][:40]}')")

    # If tools were called, do a second reasoning pass with results
    if search_results:
        followup = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=1500,
            messages=[
                {"role": "user", "content": user_query},
                {"role": "assistant", "content": response_text},
                {"role": "user", "content": f"Search results:\n{search_results}\nSummarize your findings."}
            ]
        )
        response_text = followup.content[0].text

    print("[UPDATE] Writing new knowledge to memory...")
    saved = update_memory(response_text, user_query, agent_id)
    print(f"[UPDATE] Saved {len(saved)} new facts")

    # Update recall counts for retrieved memories (stickiness)
    for mem in memories:
        db.update_recall(mem.id)

    return response_text

# Run it
result = run_research_agent(
    "What are the latest benchmarks comparing open-source embedding models?"
)
print(result)

# Periodic decay — run daily, not per-turn
removed = decay_stale_facts("research-agent-1")
print(f"Removed {removed} stale facts")

The four phases are simple individually. The discipline is applying them consistently: always READ before reasoning, always UPDATE after, always call update_recall on retrieved memories. The DECAY phase is mostly passive — the scoring formula handles it — but the explicit sweep removes low-confidence facts that the passive decay would merely demote. Together, these four phases give you an agent that learns, remembers, and forgets in a way that mimics how useful human memory works.

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