src.dackar.RCA.storage

Stage-6 storage & retrieval for canonical processed_text_record objects.

Public API:

ChromaRecordStore

Per-doc-type Chroma store (dense + in-memory BM25) for processed_text_records.

ProcessedRecordStore

Corpus-level in-memory index for hydrating records by record_id / chunk_id.

LCProcessedRetriever

Cross-doc-type hybrid retriever/orchestrator built on the two stores above.

ProcessedEvidenceStoreAdapter

Thin adapter exposing the retriever through the evidence-store query interface.

Submodules

• src.dackar.RCA.storage.chroma_store
• src.dackar.RCA.storage.lc_retriever_processed
• src.dackar.RCA.storage.multi_vector_fusion
• src.dackar.RCA.storage.processed_evidence_store_adapter
• src.dackar.RCA.storage.processed_record_store

Classes

ChromaRecordStore	Stage-6 store for canonical processed_text_record objects.
LCProcessedRetriever	Light adaptation of the earlier retrieval layer for processed_text_record.
ProcessedEvidenceStoreAdapter
ProcessedRecordStore	Corpus-level in-memory index of canonical processed_text_record objects.

Package Contents

class src.dackar.RCA.storage.ChromaRecordStore(persist_directory, *, embed_model=None, ollama_base_url=None, collection_prefix='processed', bm25_k=20)

Stage-6 store for canonical processed_text_record objects.

One Chroma collection is used per document class (CR, MR, SOP, ECA, …). Every indexed vector corresponds to one validated processed_text_record.

Parameters:

• persist_directory (str)

• embed_model (Optional[str])

• ollama_base_url (Optional[str])

• collection_prefix (str)

• bm25_k (int)

persist_directory

embed_model

ollama_base_url

collection_prefix = 'processed'

bm25_k = 20

embedder

_states: Dict[str, _CollectionState]

_collection_name(doc_type)

Parameters:

doc_type (str)

Return type:

str

get_or_create_collection(doc_type, collection_name=None)

Parameters:

• doc_type (str)

• collection_name (Optional[str])

Return type:

_CollectionState

load_collection(doc_type, collection_name=None)

Open a persisted collection for querying and return its _CollectionState.

The dense (vector) side is fully backed by Chroma’s on-disk index, so it works regardless of which process performed the original ingest.

BM25, however, is an in-memory corpus built only from documents upserted during the current process (see upsert_records()). When a collection is loaded fresh from disk without a matching in-process ingest, state.bm25_docs is empty and hybrid retrieval degrades to dense-only — hybrid_weight then has no effect. This is a deliberate limitation of the current design (the BM25 corpus is not persisted); callers needing hybrid retrieval in a query-only process must re-ingest the source JSONL first. A warning is emitted here to make the degraded mode explicit.

Parameters:

• doc_type (str)

• collection_name (Optional[str])

Return type:

_CollectionState

_get_or_build_bm25(state)

Parameters:

state (_CollectionState)

Return type:

Optional[langchain_community.retrievers.BM25Retriever]

upsert_records(records, *, doc_type=None, collection_name=None)

Upsert a batch of processed_text_records into the collection for their doc type.

All records in a call must share one doc type (a mixed-type batch raises ValueError); the type is taken from doc_type or inferred from the first record. Malformed records and records with empty embedding text are skipped. Each surviving record is upserted into Chroma by record_id (dense side) and added to the collection’s in-memory BM25 corpus (sparse side), so a subsequent query in the same process gets true hybrid retrieval.

Returns:

The number of records actually upserted.

Parameters:

• records (Iterable[Dict[str, Any]])

• doc_type (Optional[str])

• collection_name (Optional[str])

Return type:

int

upsert_jsonl(jsonl_path, *, doc_type_override=None)

Ingest a JSONL file of processed_text_records, one Chroma collection per doc type.

Records are read from jsonl_path (each line may be a bare record or wrapped under a processed_text_record key), grouped by doc type (or forced to doc_type_override), and upserted via upsert_records().

Returns:

Mapping of doc_type -> number of records upserted.

Parameters:

• jsonl_path (str)

• doc_type_override (Optional[str])

Return type:

Dict[str, int]

query_doc_type(doc_type, query_text, *, top_k=8, filter_meta=None, collection_name=None, hybrid_weight=0.5)

Hybrid (dense + BM25) retrieval over a single doc-type collection.

Runs a dense vector search and a BM25 search, then fuses the two ranked lists with Reciprocal Rank Fusion weighted by hybrid_weight (dense) / 1 - hybrid_weight (BM25). The fused _score is written back onto each returned document’s metadata.

Filtering:

filter_meta is normalized (see _normalize_filter_meta()) into scalar Chroma $eq/$in clauses. component_ids is handled specially: it becomes an index-level primary_component_id $in filter, with a legacy post-filter fallback for older records that predate primary_component_id (see _doc_matches_component_ids()).

BM25 availability:

BM25 only contributes when this collection was ingested in the current process; on a disk-loaded collection retrieval is dense-only (see load_collection()). _bm25_available is recorded on each returned document’s metadata.

Parameters:

• doc_type (str) – Document type selecting the collection.

• query_text (str) – Natural-language query.

• top_k (int) – Maximum number of fused results to return.

• filter_meta (Optional[Dict[str, Any]]) – Optional high-level metadata filters.

• collection_name (Optional[str]) – Explicit collection override (else derived from doc_type).

• hybrid_weight (float) – Dense-vs-BM25 blend in [0, 1]; 1.0 is dense-only, 0.0 is BM25-only.

Returns:

Up to top_k LangChain Document objects ordered by fused score.

Raises:

ValueError – If the target collection has not been initialised via upsert_jsonl() / upsert_records() / load_collection().

Return type:

List[langchain_core.documents.Document]

class src.dackar.RCA.storage.LCProcessedRetriever(manager, doc_store)

Light adaptation of the earlier retrieval layer for processed_text_record.

Differences from the original lc_retriever.py: - Uses ChromaRecordStore collections keyed by document type. - Treats record_id as the vector/document identity. - Hydrates canonical processed_text_record objects instead of raw mdParser chunks.

Parameters:

• manager (src.dackar.RCA.storage.chroma_store.ChromaRecordStore)

• doc_store (src.dackar.RCA.storage.processed_record_store.ProcessedRecordStore)

manager

store

query_doc_types(*, doc_types, query_text, top_k_per_doc_type=8, k_final=10, filter_meta=None, fusion='rrf', view_weights=None, hybrid_weight=0.5, snippet_preference='raw_text')

Parameters:

• doc_types (List[str])

• query_text (str)

• top_k_per_doc_type (int)

• k_final (int)

• filter_meta (Optional[Dict[str, Any]])

• fusion (str)

• view_weights (Optional[Dict[str, float]])

• hybrid_weight (float)

• snippet_preference (str)

Return type:

ContextPack

_fuse(per_view_hits, *, fusion, k_final, view_weights)

Fuse per-doc-type hit lists into a single ranked list.

Only Reciprocal Rank Fusion (RRF) is valid at this cross-doc-type layer. Each incoming hit’s score is the fused RRF score produced by ChromaRecordStore.query_doc_type (higher = better), not a raw vector distance. weighted_distance_inversion assumes a raw non-negative distance (lower = better), so applying it here would rank matches worst-first. Any non-"rrf" fusion value is therefore mapped to RRF with a one-time warning; weighted_distance_inversion remains available only for the per-view (raw-distance) layer inside the store.

Parameters:

• per_view_hits (Dict[str, List[Dict[str, Any]]])

• fusion (str)

• k_final (int)

• view_weights (Optional[Dict[str, float]])

Return type:

List[Dict[str, Any]]

class src.dackar.RCA.storage.ProcessedEvidenceStoreAdapter

retriever: src.dackar.RCA.storage.lc_retriever_processed.LCProcessedRetriever

default_doc_types: List[str] | None = None

top_k_per_doc_type: int = 8

k_final: int = 10

fusion: str = 'rrf'

hybrid_weight: float = 0.5

snippet_preference: str = 'raw_text'

query(query_text, *, top_k, filters=None)

Parameters:

• query_text (str)

• top_k (int)

• filters (Optional[Dict[str, Any]])

Return type:

List[Dict[str, Any]]

class src.dackar.RCA.storage.ProcessedRecordStore(jsonl_paths=None)

Corpus-level in-memory index of canonical processed_text_record objects.

Records are indexed by record_id, with a secondary index by provenance.chunk_id when available, so downstream components can hydrate either identifier.

Parameters:

jsonl_paths (Optional[List[str]])

_by_record_id: Dict[str, Dict[str, Any]]

_record_id_by_chunk_id: Dict[str, str]

__len__()

Return type:

int

add_jsonl(jsonl_path)

Parameters:

jsonl_path (str)

Return type:

int

add_record(rec)

Parameters:

rec (Dict[str, Any])

Return type:

bool

get(record_id)

Parameters:

record_id (str)

Return type:

Optional[Dict[str, Any]]

get_by_chunk_id(chunk_id)

Parameters:

chunk_id (str)

Return type:

Optional[Dict[str, Any]]

get_many(record_ids)

Parameters:

record_ids (List[str])

Return type:

List[Dict[str, Any]]

all_record_ids()

Return type:

List[str]