How to Build a Living Context Engine in Python (Step-by-Step, 2026)

Tutorial · Python 3.10+ · Updated May 2026

What You'll Build

By the end of this tutorial you'll have a working Living Context Engine in Python — composite scoring, typed edges, two-phase retrieval, closed feedback loop, all of it. The code is runnable, the architecture is real, and the example workflow at the end exercises every phase of the engine.

We'll use Feather DB as the underlying engine (it has the fused vector+graph+decay kernel) and wire the orchestration layer in plain Python.

Prerequisites

pip install feather-db numpy

You'll also want an embedding function. For this tutorial we'll use a stub embed() that returns a random 768-dim vector — substitute in your real model (OpenAI, Gemini, sentence-transformers).

Step 1: Open the Store

from feather_db import DB
import numpy as np
import time
import math

db = DB.open("agent.feather", dim=768)

The file agent.feather is your engine's persistent state. It contains the HNSW index, the graph, and the decay metadata. It's portable — copy it anywhere and the agent's memory comes with it.

Step 2: Define the Node Insertion Helper

Every context node carries decay state. Wrap insertions in a helper that stamps the timestamp and initializes counters.

def add_node(db, text, modality="text", importance=1.0, half_life=90):
    vec = embed(text)
    node_id = db.next_id()
    db.add(
        id=node_id,
        vec=vec,
        modality=modality,
        metadata={
            "text": text,
            "inserted_at": int(time.time()),
            "recall_count": 0,
            "importance": importance,
            "half_life_days": half_life,
        },
    )
    return node_id

Step 3: Define the Composite Scoring Function

This is the kernel of "living". It turns a similarity score into a composite score that respects time, recall, and importance.

def composite_score(similarity, meta, time_weight=0.3, now=None):
    now = now or int(time.time())
    age_days = (now - meta["inserted_at"]) / 86400
    stickiness = 1 + math.log(1 + meta["recall_count"])
    effective_age = age_days / stickiness
    half_life = meta.get("half_life_days", 90)
    recency = 0.5 ** (effective_age / half_life)
    return (
        (1 - time_weight) * similarity
        + time_weight * recency
    ) * meta["importance"]

Step 4: Two-Phase Retrieval (read)

Phase 1: ANN search for seeds. Phase 2: traverse typed edges from each seed, scoring each hop.

def read_context(db, query_text, k=5, hops=2, edge_types=None):
    query_vec = embed(query_text)

    # Phase 1 — ANN seeds
    raw_seeds = db.search(query_vec, k=k * 2)
    seeds = []
    for sid, sim in raw_seeds:
        meta = db.get_metadata(sid)
        score = composite_score(sim, meta)
        seeds.append((sid, score, meta, 0))

    seeds.sort(key=lambda x: -x[1])
    seeds = seeds[:k]

    # Phase 2 — bounded BFS on typed edges
    visited = {sid for sid, *_ in seeds}
    frontier = list(seeds)
    results = list(seeds)

    for hop in range(1, hops + 1):
        next_frontier = []
        for sid, _, _, _ in frontier:
            for nid, edge_type in db.neighbors(sid, types=edge_types):
                if nid in visited:
                    continue
                visited.add(nid)
                nmeta = db.get_metadata(nid)
                nvec = db.get_vector(nid)
                sim = float(np.dot(query_vec, nvec))
                score = composite_score(sim, nmeta) * (0.8 ** hop)
                results.append((nid, score, nmeta, hop))
                next_frontier.append((nid, score, nmeta, hop))
        frontier = next_frontier

    results.sort(key=lambda x: -x[1])
    return results

The 0.8 ** hop factor is a hop penalty — neighbors two edges away count less than direct neighbors. Tune to taste.

Step 5: Reason (call the LLM)

Format the retrieved subgraph as a context block; preserve the graph structure in the prompt.

def format_context(results):
    by_hop = {}
    for nid, score, meta, hop in results:
        by_hop.setdefault(hop, []).append((nid, score, meta))

    lines = []
    for hop in sorted(by_hop):
        label = "Directly relevant" if hop == 0 else f"Connected (hop {hop})"
        lines.append(f"\n## {label}")
        for nid, score, meta in by_hop[hop]:
            lines.append(f"- [{score:.3f}] {meta['text']}")
    return "\n".join(lines)

def reason(llm, query, results):
    ctx = format_context(results)
    prompt = f"Context:\n{ctx}\n\nQuestion: {query}\nAnswer:"
    return llm.generate(prompt)

Step 6: Write Back (close the loop)

Persist the agent's output as a new node with edges to the inputs.

def write_back(db, output_text, input_ids, edge_type="derived_from"):
    out_id = add_node(db, output_text, modality="text")
    for src_id in input_ids:
        db.link(src_id, out_id, edge_type=edge_type)
        # bump recall_count on inputs that contributed
        meta = db.get_metadata(src_id)
        meta["recall_count"] += 1
        db.update_metadata(src_id, meta)
    db.save()
    return out_id

Step 7: Apply Decay (silent + signal-driven)

Time-based decay happens automatically inside composite_score. Signal-driven adjustments are explicit:

def reinforce(db, node_ids, signal_strength=1.0):
    for nid in node_ids:
        meta = db.get_metadata(nid)
        meta["importance"] = min(3.0, meta["importance"] + 0.1 * signal_strength)
        meta["recall_count"] += int(signal_strength)
        db.update_metadata(nid, meta)
    db.save()

The Full Loop in One Function

def run_loop(db, llm, query, edge_types=None, signal_fn=None):
    # 1. Read
    results = read_context(db, query, k=5, hops=2, edge_types=edge_types)
    input_ids = [r[0] for r in results]

    # 2. Reason
    output = reason(llm, query, results)

    # 3. Update
    out_id = write_back(db, output, input_ids)

    # 4. Decay — signal capture if a feedback function is given
    if signal_fn is not None:
        strength = signal_fn(output)
        reinforce(db, input_ids, signal_strength=strength)
        reinforce(db, [out_id], signal_strength=strength)

    return output, out_id

End-to-End Example

db = DB.open("marketing.feather", dim=768)

# Seed the store
brief_id  = add_node(db, "Q3 brand-x campaign brief: emphasize sustainability", importance=2.0)
brand_id  = add_node(db, "brand-x guidelines: warm tone, professional voice")
db.link(brief_id, brand_id, edge_type="references")

# Run the loop
output, out_id = run_loop(
    db, llm,
    query="draft a launch headline for the brand-x campaign",
    edge_types=["references", "responds_to", "derived_from"],
    signal_fn=lambda o: 1.5 if "sustain" in o.lower() else 1.0,
)

print(output)

Each subsequent call benefits from the previous one. The output node persists with edges to brief and brand. Future drafts retrieve the context graph, see the previous draft, and write decisions that build on it.

What This Buys You

You now have a Living Context Engine running locally in Python:

Adaptive scoring is applied to every retrieval automatically.
The graph topology densifies as the system runs.
Every agent output becomes context for future calls.
Decay suppresses stale entries automatically; importance multipliers protect cross-cutting material.

The full source for a more polished version of this tutorial is in the Feather DB cookbook. For production deployment patterns — sharding, multi-agent isolation, observability — see the documentation.