principle  : no decay (MBSE design principles, math/algorithms)
paper      : re-evaluated roughly every half year (papers, technical articles)
news       : decays significantly over weeks to months (news, model-release info)
experiment : decays according to project duration (experiment logs, run records)

pip install -r requirements.txt

pip install -r requirements-dev.txt

python mcp_server/server.py

HYBRID_RAG_DB_PATH=my_memory.db HYBRID_RAG_INDEX_PATH=my_indices python mcp_server/server.py

{
  "mcpServers": {
    "hybrid-rag-memory": {
      "type": "stdio",
      "command": "python",
      "args": ["mcp_server/server.py"],
      "env": {
        "HYBRID_RAG_DB_PATH": "hybrid_rag.db",
        "HYBRID_RAG_INDEX_PATH": "indices"
      }
    }
  }
}

1. ingest(["design_doc.md"], metadata={"knowledge_type": "principle", "tags": ["mbse"]})
2. hybrid_search("about consistency between requirements and architecture", top_k=5)
   -> [{"doc_id": ..., "chunk_index": ..., "content": ..., "knowledge_type": "principle",
        "importance": null, "access_count": 0, "score": 0.87}, ...]
3. set_chunk_tags(doc_id, chunk_index, importance=0.9)
4. rerank(chunks, time_weight=0.5, freq_weight=0.1, importance_threshold=0.3)
   -> chunks reordered along the memory axis (staleness, frequency, importance)

Use the rag-memory agent to look into past design decisions

from hybrid_rag import HybridRAGSystem

rag = HybridRAGSystem(db_path="hybrid_rag.db", index_path="indices")

rag.ingest_documents(
    ["design_doc.md"],
    metadata={"knowledge_type": "principle", "importance": 0.9, "tags": ["mbse"]},
)

result = rag.query(
    "about consistency between requirements and architecture",
    top_k=5,
    enable_memory_rerank=True,   # enable the memory mechanism's staged reranking
    memory_time_weight=0.5,
    memory_freq_weight=0.1,
    memory_importance_threshold=0.3,
)
print(result["context"])

# assign an importance tag after the fact (no vector rebuild needed)
rag.set_chunk_tags(doc_id="design_doc_xxxx", chunk_index=0, importance=0.9)

# forgetting/archival batch (normally run infrequently)
report = rag.run_forgetting_batch(score_threshold=0.05, dry_run=True)

python scripts/run_forgetting_batch.py --dry-run
python scripts/run_forgetting_batch.py --score-threshold 0.1 --time-weight 0.8

cp eval/golden_queries.example.yaml eval/golden_queries.yaml  # once, at first use — rewrite the doc_ids for your own corpus
python scripts/run_evaluation.py --db-path mcp_server/hybrid_rag.db --index-path mcp_server/hybrid_rag_indices

python scripts/find_near_duplicates.py --db-path mcp_server/hybrid_rag.db
python scripts/find_near_duplicates.py --db-path mcp_server/hybrid_rag.db --output eval/duplicates_report.json

pytest tests/ -v

from hybrid_rag import create_rag_system

rag = create_rag_system(backend="faiss")   # "qdrant" / "chroma" / "postgres" are also available
rag.ingest_documents(["document1.pdf", "document2.md"])
result = rag.query("What is machine learning?", top_k=5)

rag = HybridRAGSystem(
    dense_model="paraphrase-multilingual-MiniLM-L12-v2",
    rerank_model="BAAI/bge-reranker-v2-m3",
    max_chunk_size=512,
    index_type="hnsw",           # "flat" / "ivf" / "hnsw"
    enable_mmr=True, mmr_lambda=0.6,
    enable_cache=True, cache_ttl_seconds=3600,
    query_expander=None,          # pass a QueryExpander instance for LLM-based query expansion
    memory_half_life_overrides=None,  # override the half-life (days) per knowledge_type
    enable_guaranteed_candidates=True,  # always add principle/high-importance chunks to the candidate pool (default True)
    guaranteed_knowledge_types=None,    # defaults to ["principle"]
    guaranteed_importance_threshold=0.7,
    guaranteed_candidates_limit=50,
)

pip install sphinx sphinx-rtd-theme
python -m sphinx -b html docs/source docs/build

hybrid_rag_agent_spec.md   # design spec for the memory mechanism
.mcp.json                  # MCP server registration for Claude Code
.claude/agents/rag-memory.md  # sub-agent definition for Claude Code

mcp_server/
└── server.py              # the MCP server itself (13 tools, see the table above)

scripts/
├── run_forgetting_batch.py       # CLI for the forgetting/archival batch
├── run_evaluation.py             # CLI that automatically evaluates retrieval accuracy against a golden query set
├── find_near_duplicates.py       # CLI that audits near-duplicate ingests in the existing corpus (report-only, never deletes)
├── backfill_source_date.py       # bulk-backfills source_date on existing chunks
├── list_md_files.py              # lists candidate Markdown files for ingestion
├── manage_ingest_status.py       # tracks ingest progress against list_md_files.py's listing
├── manage_conv_ingest_status.py  # tracks ingest progress against convert_conversations.py's output
└── convert_conversations.py      # converts a Claude.ai export (JSON) into Markdown

hybrid_rag/
├── __init__.py
├── ingestion.py            # document processing
├── chunking.py             # semantic chunking
├── indexing.py             # dense & sparse index (FAISS)
├── indexing_bm25.py        # BM25 index
├── indexing_sparse_tfidf.py  # TF-IDF sparse index (shared by the Chroma/Postgres/Qdrant backends)
├── indexing_qdrant.py / indexing_chroma.py / indexing_postgres.py
├── retrieval.py            # RRF search
├── reranking.py            # Cross-encoder reranking (relevance axis)
├── memory_scoring.py        # staged reranking (memory axis: importance/decay/frequency)
├── archival.py              # forgetting/archival batch processing
├── index_health.py          # index/DB consistency checking (backs the ⑫ index_health tool)
├── caching.py / embedding_cache.py
├── context.py / diversity.py / evaluation.py
├── storage.py               # SQLite database (including memory-mechanism fields)
├── query_expansion.py
├── rag_system.py            # main orchestrator (FAISS version, implements the memory mechanism)
├── _rag_system_indexing.py  # ^ ingest/build/incremental-update/load (mixin)
├── _rag_system_query.py     # ^ query pipeline (mixin)
├── _rag_system_memory.py    # ^ tags/neighboring chunks/forgetting batch (mixin)
├── _rag_system_stats.py     # ^ stats & cache management (mixin)
├── _rag_system_docops.py    # ^ embedding/delete/lightweight search (mixin)
├── rag_system_base.py       # base class shared by the Chroma/Postgres/Qdrant backends
├── rag_system_qdrant.py / rag_system_chroma.py / rag_system_postgres.py
└── rag_system_factory.py

tests/
├── test_metadata_pipeline.py   # metadata regression test across ingest → build_index → query
├── test_memory_scoring.py      # unit tests for staged reranking
├── test_archival.py            # unit tests for the forgetting/archival batch
├── test_incremental_index.py   # unit/integration tests for update_index (incremental updates)
├── test_index_health.py        # unit tests for index_health (index/DB consistency check)
├── test_result_dedup.py        # unit tests for RRF fusion-key stability and search-result dedup
├── test_diversity.py           # unit tests for MMR diversity selection
├── test_reranking.py           # unit tests for Cross-encoder reranking stats
├── test_retriever_shutdown.py  # tests for RRFRetriever resource cleanup (thread leaks)
├── test_indexing_bm25.py       # unit tests for the BM25 index
├── test_storage_concurrency.py # unit tests for concurrent SQLite writes
├── test_source_date.py         # unit tests for source_date derivation (time-decay reference point)
├── test_document_chunks.py     # unit tests for get_document_chunks (fetching neighboring chunks)
├── test_evaluation.py          # unit tests for RAGEvaluator (Precision@k, etc.)
├── test_database_stats.py      # unit tests for get_database_stats / duplicate-ingest detection
├── test_guaranteed_candidates.py  # unit tests for guaranteed candidate-pool merging (the fix for principle burial)
├── test_rag_system_factory.py  # unit tests for create_rag_system (backend switching)
└── conftest.py                 # shared pytest configuration