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

> A working Living Context Engine in 50 lines of Python. This tutorial walks through every architectural primitive — adaptive scoring, typed edges, two-phase retrieval, write-back — with runnable code and a real example workflow.

- **Category**: Tutorial
- **Read time**: 16 min read
- **Date**: May 15, 2026
- **Author**: Feather DB Engineering (Engineering Team)
- **URL**: https://getfeather.store/theory/how-to-build-living-context-engine-python

---

# 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](https://getfeather.store/docs) as the underlying engine (it has the fused vector+graph+decay kernel) and wire the orchestration layer in plain Python.

## Prerequisites

```bash
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

```python
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.

```python
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.

```python
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.

```python
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.

```python
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.

```python
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:

```python
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

```python
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

```python
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](/docs).

---

*Related: [Closing the Loop in Feather DB](/theory/closing-the-loop-feather-db) · [The Four Phases Explained](/theory/read-reason-update-decay).*

---

*This is the machine-readable mirror of the theory post at [getfeather.store/theory/how-to-build-living-context-engine-python](https://getfeather.store/theory/how-to-build-living-context-engine-python). For the full Feather DB documentation, see [getfeather.store/llms-full.txt](https://getfeather.store/llms-full.txt).*