Incremental enhancement of picobot/agent/knowledge.py from keyword-only search to three-retriever hybrid search with semantic embeddings.

Each section is independently shippable and improves search quality. Embeddings are always on — nomic[local] is a required dependency.

See report.md for research findings behind each decision.

---

Dependency setup¶

Add nomic[local] to pyproject.toml as a required dependency. Add ~/.picobot/data/embeddings.npz to .gitignore.

[project]
dependencies = [
    # ... existing deps ...
    "nomic[local]",
]

Nomic installs the nomic SDK (~5 MB) plus GPT4All for local inference (~50 MB). The GGUF model file (~140 MB) is downloaded on first use.

Nomic is the only viable embedding model that works both locally (no API key, offline) and via API with compatible vectors across modes. No need for a provider abstraction — Nomic's inference_mode parameter handles local/remote switching:

from nomic import embed
# local (default) — no API key needed
output = embed.text(texts, model='nomic-embed-text-v1.5', inference_mode='local', dimensionality=768)
# remote — faster for batch re-embedding, needs NOMIC_API_KEY
output = embed.text(texts, model='nomic-embed-text-v1.5', inference_mode='remote', dimensionality=768)

Config in config.json:

{
  "knowledge": {
    "embeddings": {
      "inference_mode": "local"
    }
  }
}

---

Full-text search with SQLite FTS5 (~80 LOC)¶

What: Replace _search_fulltext() (substring scan over every file on disk) with an in-memory SQLite FTS5 index. Zero new dependencies — Python stdlib sqlite3 includes FTS5.

Why: BM25 ranking and Porter stemming are a significant quality improvement over naive substring matching. Building the index on init also establishes the index-on-init pattern used by the embedding index.

Where: picobot/agent/knowledge.py

Changes:

1. _build_fts_index() — called lazily on first search
2. Creates :memory: SQLite database
3. CREATE VIRTUAL TABLE kb_fts USING fts5(doc_id, title, tags, body, tokenize='porter')
4. Iterates _iter_entries(), parses frontmatter, inserts rows

5. ~50-100ms for 1000 docs

6. _search_fts() replaces _search_fulltext() — single SQL query:

   SELECT doc_id, rank FROM kb_fts WHERE kb_fts MATCH ? ORDER BY rank LIMIT 20

7. Index invalidated and rebuilt lazily when store() or sync() modifies entries.

Tests: Existing KB search tests should pass (same API, better ranking). Add FTS-specific tests for stemming, BM25 ordering.

---

Embedding index (~120 LOC)¶

What: New module picobot/agent/knowledge_embeddings.py — embeds KB documents, caches vectors, provides cosine similarity search.

EmbeddingIndex class:

Embedding config: - Model: nomic-embed-text-v1.5 at 768 dimensions (full Matryoshka — at picobot's scale the memory savings of 256d are negligible, and 768d gets +1.24 MTEB) - Inference mode: local by default (configurable to remote for faster batch re-embedding) - Input: f"{title}. Tags: {tags_csv}. {body}" (full document, not chunked — KB entries are already coherent semantic units at 500-2000 words)

Sidecar cache: ~/.picobot/data/embeddings.npz — numpy compressed archive with vectors (N, 768), doc_ids, content_hashes. Content-hash invalidation: hashlib.md5(body.encode()).hexdigest() per entry, only re-embed changed docs on subsequent inits.

Vector search (entire implementation — why numpy, not FAISS):

query_vec = embed.text([query], model='nomic-embed-text-v1.5',
                       inference_mode='local', dimensionality=768)[0]
query_vec = query_vec / np.linalg.norm(query_vec)
scores = self.vectors @ query_vec  # pre-normalized at index time
top_k_idx = np.argsort(scores)[-top_k:][::-1]
return [(self.doc_ids[i], float(scores[i])) for i in top_k_idx]

numpy is already a required dependency. faiss-cpu is an additional 15-27 MB compiled C++ package that, at 1000 docs, is actually slower than raw numpy due to dispatch overhead. FAISS only wins with approximate indexes (IVF, HNSW) at 5K+ docs. Graduate to faiss-cpu at that point — it's a ~10 LOC swap from scores = vectors @ query_vec to index.search(query_vec, k).

Tests: Unit tests with mocked embed.text() (don't require Nomic model in CI). Integration tests under pytest -m embeddings marker.

---

Three-retriever RRF fusion (~60 LOC)¶

What: Wire all three retrievers into KnowledgeBase.search() and fuse results with Reciprocal Rank Fusion.

Where: picobot/agent/knowledge.py — modify search() method

Architectural change: Search currently returns early when frontmatter yields 3+ results. Hybrid search always runs all available retrievers and fuses — every signal improves ranking.

def search(self, query, topic=None):
    fm_results = self._search_frontmatter(query, topic)
    fts_results = self._search_fts(query, topic)
    vec_results = self._embedding_index.search(query, top_k=20)

    fused = self._rrf_fuse([
        [r['path'] for r in fm_results],
        [r['path'] for r in fts_results],
        [doc_id for doc_id, _ in vec_results],
    ])
    return self._format_results(fused[:10])

RRF implementation (~15 LOC):

def _rrf_fuse(self, ranked_lists, k=60):
    scores = {}
    for ranked in ranked_lists:
        for rank, doc_id in enumerate(ranked):
            scores[doc_id] = scores.get(doc_id, 0) + 1.0 / (k + rank + 1)
    return sorted(scores, key=scores.get, reverse=True)

RRF requires no score normalization, handles different score scales naturally. The k=60 constant prevents any single retriever from dominating.

Trigger embedding index update on store() and sync().

---

HyPE: Hypothetical Question Embedding (~80 LOC)¶

What: At kb store time, use the LLM to generate 5-10 hypothetical questions each document could answer, then embed those questions alongside the document vector.

Why: Agent queries are intent-based ("how do I handle errors?") which match poorly against document text. Question-to-question matching is much higher fidelity. HyPE benchmarks show up to +42pp context precision.

Where: picobot/agent/knowledge_embeddings.py — extend EmbeddingIndex

Changes:

1. generate_hype_questions(title, body) -> list[str] — calls the LLM:

   Given this knowledge base entry, generate 5-10 questions an AI agent
   might ask when it needs this information. Focus on:
   - Task-oriented: "How do I...", "What's the best way to..."
   - Lookup: "What is...", "Where is..."
   - Decision: "Should I...", "When should I..."

2. Embed both the document AND its HyPE questions as separate vectors. Each HyPE vector maps back to its source doc_id — a single document may have 6-11 vectors (1 doc + 5-10 questions).

3. Store generated questions in the .npz sidecar (regenerate only when content hash changes).

LLM models for question generation: This is a structured extraction task — doesn't need frontier reasoning. Use the cheapest model that produces well-formed questions.

Cost estimates assume ~500 input tokens per doc and ~200 output tokens. Default: use whatever LLM is already configured in picobot's config.json. Any model above ~3B parameters produces usable output.

Configurable: knowledge.embeddings.hype: true (default). Can disable for cost/speed in CI.

---

Document type weighting and recency decay (~40 LOC)¶

What: Apply document-type boost factors and journal recency decay in RRF fusion.

Where: picobot/agent/knowledge.py — extend _rrf_fuse()

Type boosts (applied after RRF scoring):

Recency decay (journal only): Exponential with 30-day half-life. A 1-day-old entry gets 0.98x, 30-day-old gets 0.5x, 90-day-old gets 0.125x.

def recency_weight(updated_timestamp, half_life_days=30):
    age_days = (time.time() - updated_timestamp) / 86400
    return math.exp(-0.693 * age_days / half_life_days)

Not in search index: Soul/identity docs are deterministically injected into every system prompt (relevant to every query — retrieval is lossy and might miss them). Root README.md and per-topic README.md are excluded (already are).

---

Scale milestones¶

---

Dependency budget¶

---

Open questions¶

1. Nomic API pricing: Sources suggest Nomic may have moved to seat-based pricing. Verify before relying on remote inference for batch re-embedding.

2. Model load lifecycle: Local Nomic loads in ~1-3 seconds. For the daemon: load once at start, keep in memory. For CLI one-shot: load on demand.

---

Summary¶

---¶

Project context (for unsupervised implementation)¶

Everything an agent needs to implement this plan in a fresh session.

Codebase orientation¶

picobot/
├── agent/
│   ├── knowledge.py          # PRIMARY FILE — KnowledgeBase class (631 lines)
│   ├── knowledge_embeddings.py  # NEW FILE — EmbeddingIndex class
│   ├── context.py             # ContextBuilder — instantiates KnowledgeBase(workspace=workspace) at line 25
│   ├── tools/
│   │   └── knowledge.py       # KBTool — delegates to KnowledgeBase, no changes needed
│   └── subagent.py
├── config/
│   └── schema.py              # Pydantic config models — KnowledgeConfig needs extension
├── providers/
│   └── litellm_provider.py    # LiteLLMProvider — use for HyPE LLM calls
├── utils/
│   └── helpers.py             # ensure_dir(), safe_filename()
tests/
└── test_knowledge.py          # 462 lines, 20+ tests — all use tmp_path fixture
pyproject.toml                 # Dependencies list — add nomic[local] here
.gitignore                     # Add embeddings.npz sidecar path

KnowledgeBase class (knowledge.py)¶

Constructor: KnowledgeBase(base_path=None, workspace=None). If workspace given, self.base = workspace / "knowledge".

Instantiated in 4 places: - picobot/agent/context.py:25 — KnowledgeBase(workspace=workspace) - picobot/agent/tools/knowledge.py:14 — KnowledgeBase(base_path=base_path, workspace=workspace) - picobot/cli/commands.py:524 — KnowledgeBase(workspace=workspace) (init command) - picobot/cli/commands.py:1170 — KnowledgeBase(workspace=workspace) (kb sync command)

Key methods to modify: - search(query, topic=None) (line 99) — currently two-phase: frontmatter then fulltext fallback if <3 results. Replace with three-retriever RRF fusion. - _search_fulltext(query, topic=None) (line 477) — substring scan over disk. Replace with FTS5. - store(topic, title, content, tags) (line 44) — add index invalidation triggers.

Key methods to keep unchanged: - _search_frontmatter(query, topic) (line 445) — first retriever, keep as-is. - _parse_frontmatter(content) (line 400) — returns (dict, str) tuple. - _iter_entries(topic=None) (line 556) — yields list[Path] of .md files. - _query_terms(query) (line 567) — splits to lowercase terms. - sync(remote) (line 180) — add index invalidation after sync.

Config schema (config/schema.py)¶

KnowledgeConfig currently only has remote: str | None = None. Extend it:

class EmbeddingsConfig(BaseModel):
    """Embedding index configuration."""
    inference_mode: str = "local"  # "local" or "remote"
    hype: bool = True  # generate HyPE questions on store

class KnowledgeConfig(BaseModel):
    """Knowledge base configuration."""
    remote: str | None = None
    embeddings: EmbeddingsConfig = Field(default_factory=EmbeddingsConfig)

The Config root model at line 83 already has knowledge: KnowledgeConfig = Field(default_factory=KnowledgeConfig), so this nests automatically into config.knowledge.embeddings.inference_mode.

LLM provider for HyPE¶

HyPE question generation should use litellm directly (same pattern as LiteLLMProvider.chat()):

from litellm import acompletion

response = await acompletion(
    model=model,  # default from config: "openai/gpt-5.2-mini"
    messages=[{"role": "user", "content": prompt}],
    max_tokens=500,
    temperature=0.3,
)
questions = response.choices[0].message.content

The default model is config.agents.defaults.model ("openai/gpt-5.2-mini"). For HyPE, any model works — gpt-4o-mini, haiku, local ollama. The agent doesn't need to accept a config option for HyPE model — just use the configured default.

Test patterns¶

All KB tests use this pattern:

def test_something(tmp_path):
    from picobot.agent.knowledge import KnowledgeBase
    kb = KnowledgeBase(workspace=tmp_path)
    kb.ensure_structure()
    # ... test ...

For embedding tests, mock nomic.embed.text() — don't require the actual model in CI:

@pytest.fixture
def mock_embed(monkeypatch):
    def fake_embed(texts, model, inference_mode, dimensionality):
        return [np.random.randn(dimensionality) for _ in texts]
    monkeypatch.setattr("nomic.embed.text", fake_embed)

For FTS5 tests, no mocking needed — sqlite3 is stdlib.

Mark integration tests that need the real Nomic model: @pytest.mark.embeddings.

Lint rules¶

• ruff: select = ["E", "F", "I", "N", "W"], ignore = ["E501", "N803", "N815"]
• N803 ignored: tool execute() methods use camelCase args to match JSON schema
• E501 ignored: no line length enforcement
• Run ruff check picobot/ tests/ to verify
• Run pytest tests/ -v — all 340 existing tests must still pass

Data paths¶

• KB root: ~/.picobot/workspace/knowledge/
• Embedding sidecar: ~/.picobot/data/embeddings.npz
• Config: ~/.picobot/config.json

---

Task breakdown¶

Six tasks with dependency graph. Tasks A and B can run in parallel after Task 0.

Dependency graph¶

Task 0: Dependency setup + config schema
    │
    ├──→ Task A: FTS5 full-text search  ──┐
    │                                       ├──→ Task C: RRF fusion wiring
    └──→ Task B: Embedding index module ──┘         │
                    │                                │
                    └──→ Task D: HyPE questions      │
                                                     │
                                              Task E: Type weighting + recency decay

Task 0: Dependency setup + config schema¶

Files: pyproject.toml, picobot/config/schema.py, .gitignore Depends on: nothing Blocks: Tasks A, B, C, D, E

Changes: 1. Add "nomic[local]" to dependencies list in pyproject.toml (after "sse-starlette>=1.0.0") 2. Add "numpy>=1.24.0" to dependencies if not already present (check — it may be a transitive dep but should be explicit) 3. Add EmbeddingsConfig model to picobot/config/schema.py with inference_mode: str = "local" and hype: bool = True 4. Add embeddings: EmbeddingsConfig = Field(default_factory=EmbeddingsConfig) to KnowledgeConfig 5. Add ~/.picobot/data/embeddings.npz pattern to .gitignore 6. Run uv pip install -e ".[dev]" to verify nomic installs 7. Run ruff check picobot/config/schema.py 8. Run pytest tests/ -v — all existing tests must pass

Task A: FTS5 full-text search (parallel with B)¶

Files: picobot/agent/knowledge.py, tests/test_knowledge.py Depends on: Task 0 Blocks: Task C

Changes to knowledge.py: 1. Add import sqlite3 at top 2. Add self._fts_db = None in __init__ 3. Add _build_fts_index() method: - Creates :memory: SQLite db - CREATE VIRTUAL TABLE kb_fts USING fts5(doc_id, title, tags, body, tokenize='porter') - Iterates self._iter_entries(), calls _parse_frontmatter() on each, inserts (rel_path, title, tags_csv, body) - Sets self._fts_db = db 4. Add _invalidate_fts() method — sets self._fts_db = None 5. Replace _search_fulltext() body with _search_fts(): - Call _build_fts_index() if self._fts_db is None - Query: SELECT doc_id, rank FROM kb_fts WHERE kb_fts MATCH ? ORDER BY rank LIMIT 20 - FTS5 MATCH syntax: join query terms with spaces (FTS5 treats as implicit AND) - If topic is set, filter results by path prefix - Return same list[dict] format as current _search_fulltext() (keys: path, title, tags, score, snippet) 6. Add self._invalidate_fts() call at end of store() (after file write, line ~95) 7. Add self._invalidate_fts() call in sync() (after merge/push, before return)

Keep _search_fulltext as the method name (or rename to _search_fts and update the one call site in search()). The search() method itself gets rewritten in Task C, so don't change search() logic here — just make _search_fulltext use FTS5 internally.

Tests to add: - test_fts_stemming — store "running quickly", search "run quick", verify found - test_fts_bm25_ordering — store 2 entries, one with term appearing 3x, one with 1x, verify ordering - test_fts_rebuild_on_store — store entry, search, store another, search again, verify new entry found - Existing test_search_fulltext_fallback should still pass (same API)

Task B: Embedding index module (parallel with A)¶

Files: picobot/agent/knowledge_embeddings.py (new), tests/test_knowledge_embeddings.py (new) Depends on: Task 0 Blocks: Tasks C, D

Create picobot/agent/knowledge_embeddings.py:

"""Embedding index for semantic KB search."""

import hashlib
from pathlib import Path

import numpy as np


class EmbeddingIndex:
    """Embeds KB documents, caches vectors, provides cosine similarity search."""

    def __init__(self, data_dir: Path, inference_mode: str = "local"):
        self.data_dir = data_dir
        self.inference_mode = inference_mode
        self.cache_path = data_dir / "embeddings.npz"
        self.vectors = None       # (N, 768) float32
        self.doc_ids = None       # list[str]
        self.content_hashes = None  # list[str]
        self._loaded = False

    def build(self, entries: list[dict]) -> None:
        """Embed all entries and save to cache.

        entries: list of {"doc_id": str, "title": str, "tags": list[str], "body": str}
        """
        ...

    def load(self) -> bool:
        """Load cached embeddings. Returns True if cache exists and loaded."""
        ...

    def update(self, entries: list[dict]) -> int:
        """Re-embed only entries with changed content hash. Returns count updated."""
        ...

    def search(self, query: str, top_k: int = 20) -> list[tuple[str, float]]:
        """Cosine similarity search. Returns [(doc_id, score), ...]."""
        ...

    @staticmethod
    def content_hash(text: str) -> str:
        return hashlib.md5(text.encode()).hexdigest()

    def _embed_texts(self, texts: list[str]) -> np.ndarray:
        """Call Nomic to embed texts. Returns (N, 768) array, L2-normalized."""
        from nomic import embed
        output = embed.text(
            texts,
            model="nomic-embed-text-v1.5",
            inference_mode=self.inference_mode,
            dimensionality=768,
        )
        vecs = np.array(output["embeddings"], dtype=np.float32)
        norms = np.linalg.norm(vecs, axis=1, keepdims=True)
        norms[norms == 0] = 1
        return vecs / norms

    def _embedding_text(self, entry: dict) -> str:
        """Construct embedding input from entry metadata + body."""
        title = entry.get("title", "")
        tags_csv = ", ".join(entry.get("tags", []))
        body = entry.get("body", "")
        return f"{title}. Tags: {tags_csv}. {body}"

Key implementation details: - build(): embed all, save np.savez_compressed(cache_path, vectors=..., doc_ids=..., content_hashes=...) - load(): np.load(cache_path, allow_pickle=True), populate self.vectors/doc_ids/content_hashes - update(): compare content hashes, only re-embed changed entries, update cache - search(): embed query, scores = self.vectors @ query_vec, argsort, return top_k - Ensure data_dir exists (data_dir.mkdir(parents=True, exist_ok=True)) - L2-normalize vectors at index time so search is a simple dot product

Tests (tests/test_knowledge_embeddings.py): - Mock nomic.embed.text in all unit tests — return random normalized vectors - test_build_and_search — build index from 5 entries, search, verify returns doc_ids with scores - test_cache_roundtrip — build, save to npz, create new EmbeddingIndex, load, verify vectors match - test_update_only_changed — build 5, change 1, update, verify only 1 re-embedded (check mock call count) - test_content_hash — verify deterministic hash - test_search_empty_index — search on empty index returns empty list - test_data_dir_created — verify EmbeddingIndex creates data_dir if missing

Task C: RRF fusion wiring¶

Files: picobot/agent/knowledge.py, tests/test_knowledge.py Depends on: Tasks A and B Blocks: Task E

Changes to knowledge.py: 1. Add from picobot.agent.knowledge_embeddings import EmbeddingIndex at top 2. Modify __init__ to accept optional data_dir and inference_mode params:

def __init__(self, base_path=None, workspace=None, data_dir=None, inference_mode="local"):
    # ... existing base setup ...
    self._embedding_index = EmbeddingIndex(
        data_dir=data_dir or Path("~/.picobot/data").expanduser(),
        inference_mode=inference_mode,
    )

3. Add _ensure_indexes() method — lazily builds both FTS and embedding indexes: - Calls _build_fts_index() if needed - Calls _embedding_index.load(), if no cache: collects entries via _iter_entries(), builds with _embedding_index.build(entries) 4. Replace search() method body:

def search(self, query, topic=None):
    self._ensure_indexes()
    fm_results = self._search_frontmatter(query, topic)
    fts_results = self._search_fts(query, topic)
    vec_results = self._embedding_index.search(query, top_k=20)

    fused = self._rrf_fuse([
        [r['path'] for r in fm_results],
        [r['path'] for r in fts_results],
        [doc_id for doc_id, _ in vec_results],
    ])
    return self._format_results(fused[:10])

5. Add _rrf_fuse(ranked_lists, k=60) method (15 LOC, see plan above) 6. Add _format_results(doc_ids) — reads frontmatter for each doc_id, formats same output string as current search() 7. Modify store() — after file write, invalidate FTS and update embedding index:

self._invalidate_fts()
# Update embedding index for new entry
entry = {"doc_id": str(rel_path), "title": title, "tags": tags or [], "body": content}
self._embedding_index.update([entry])

8. Modify sync() — after sync completes, invalidate both indexes:

self._invalidate_fts()
self._embedding_index = EmbeddingIndex(...)  # force full rebuild on next search

Update instantiation sites — the 4 call sites just pass workspace= which still works, but context.py should pass config values: - picobot/agent/context.py:25 — pass data_dir and inference_mode from config if available (this can be deferred — defaults work)

Tests: - All existing tests must pass (mock nomic.embed.text in a conftest fixture or monkeypatch at module level) - test_rrf_fuse — unit test with hand-crafted ranked lists, verify fusion ordering - test_search_hybrid — store 3 entries, search with term only in body of one, verify RRF finds it - test_search_formats_results — verify output string format matches current format (existing tests cover this)

Important: Existing tests import KnowledgeBase and call search(). The new search() uses the embedding index, which calls nomic.embed.text. Tests will fail unless nomic is mocked. Add a conftest.py fixture or monkeypatch in the test file that mocks nomic.embed.text globally for all KB tests, returning random vectors.

Task D: HyPE questions¶

Files: picobot/agent/knowledge_embeddings.py, tests/test_knowledge_embeddings.py Depends on: Task B Can run in parallel with: Task C (but merge after C since both touch knowledge_embeddings.py)

Changes to knowledge_embeddings.py: 1. Add generate_hype_questions(title, body) -> list[str] async method:

async def generate_hype_questions(self, title: str, body: str, model: str = None) -> list[str]:
    from litellm import acompletion
    prompt = f"""Given this knowledge base entry, generate 5-10 questions an AI agent
might ask when it needs this information. Focus on:
- Task-oriented: "How do I...", "What's the best way to..."
- Lookup: "What is...", "Where is..."
- Decision: "Should I...", "When should I..."

Title: {title}
Content: {body[:2000]}

Return only the questions, one per line, no numbering."""

    response = await acompletion(
        model=model or "openai/gpt-5.2-mini",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=500,
        temperature=0.3,
    )
    text = response.choices[0].message.content
    return [q.strip() for q in text.strip().splitlines() if q.strip()]

2. Extend build() and update() to optionally generate and embed HyPE questions: - Add hype_enabled: bool = True to __init__ - In build(), for each entry: generate questions, embed them, map each HyPE vector back to the source doc_id - Store in npz: hype_vectors, hype_doc_ids, hype_questions (for cache) 3. Extend search() to search both document vectors and HyPE vectors, combine results: - Search doc vectors → list of (doc_id, score) - Search HyPE vectors → list of (doc_id, score) (note: multiple HyPE vectors can map to same doc) - Merge: for each doc_id, take max score across doc + HyPE hits 4. Store HyPE questions and their content hashes in the .npz sidecar — only regenerate when content hash changes

Note on async: generate_hype_questions is async because acompletion is async. build() and update() will need to become async too, or use asyncio.run() internally. Check how the call sites work — store() in knowledge.py is sync. Options: - Make build()/update() sync and use litellm.completion() (sync version) for HyPE - Or keep HyPE generation as a separate step triggered by a CLI command (picobot kb reindex)

Simplest: use litellm.completion() (sync) so the whole pipeline stays sync. store() is sync, search() is sync, keep it consistent.

Tests: - Mock litellm.completion() to return canned questions - test_hype_question_generation — verify returns list of strings - test_hype_vectors_in_search — build with HyPE, search with a question, verify found via HyPE vector - test_hype_disabled — set hype_enabled=False, verify no LLM calls made - test_hype_cache_persistence — build, save, load, verify HyPE questions survive roundtrip

Task E: Type weighting + recency decay¶

Files: picobot/agent/knowledge.py, tests/test_knowledge.py Depends on: Task C

Changes to knowledge.py: 1. Add import math, time at top 2. Add recency_weight(updated_timestamp, half_life_days=30) function 3. Extend _rrf_fuse() to accept optional entry metadata and apply type boosts + recency decay:

def _rrf_fuse(self, ranked_lists, k=60, entry_meta=None):
    scores = {}
    for ranked in ranked_lists:
        for rank, doc_id in enumerate(ranked):
            scores[doc_id] = scores.get(doc_id, 0) + 1.0 / (k + rank + 1)

    if entry_meta:
        for doc_id in scores:
            meta = entry_meta.get(doc_id, {})
            # Type boost
            if "SKILL.md" in doc_id:
                scores[doc_id] *= 1.2
            elif doc_id.startswith("journal/"):
                scores[doc_id] *= 0.8
                # Recency decay
                updated = meta.get("updated")
                if updated:
                    scores[doc_id] *= recency_weight(updated)

    return sorted(scores, key=scores.get, reverse=True)

4. In search(), collect entry metadata (updated timestamps) when iterating entries, pass to _rrf_fuse()

Tests: - test_type_boost_skills — verify SKILL.md entries rank higher than equal-scoring KB entries - test_recency_decay_journal — create journal entries with different ages, verify recent ones rank higher - test_no_decay_on_non_journal — verify non-journal entries are not affected by recency decay

---

Verification checklist¶

After all tasks complete: 1. ruff check picobot/ tests/ — no lint errors 2. pytest tests/ -v — all tests pass (existing + new) 3. picobot init still works (creates KB structure) 4. picobot agent -m "search my knowledge base for X" — search uses hybrid retrieval 5. Manual test: store an entry, search for it using conceptual terms (not exact keywords), verify found via embedding