src.dackar.RCA.storage.chroma_store

Attributes

LOGGER
LIST_FIELDS

Classes

_CollectionState
ChromaRecordStore	Stage-6 store for canonical processed_text_record objects.

Functions

_iter_jsonl(jsonl_path)
_collapse_ws(s)
_looks_like_processed_text_record(obj)
extract_processed_text_record(obj)
_stable_record_id_from_doc(doc)
_sanitize_meta_value(value)
_normalize_filter_meta(filter_meta)	Normalize high-level retrieval filters into the scalar metadata keys that
_doc_matches_component_ids(doc, wanted)	Return True if doc is associated with at least one of the wanted component IDs.
_doc_matches_filter_sane(doc, filter_sane)	Apply the dense-path Chroma filter semantics to a BM25 hit's metadata.
_sanitize_meta(meta)
_record_component_ids(metadata)
build_chroma_metadata(record)	Flatten a canonical processed_text_record into Chroma-storable metadata.
to_chroma_payload(record)	Convert a processed_text_record into the (id, document, metadata) triple upserted
collection_name_for_doc_type(doc_type[, prefix])	Return the Chroma collection name for a document type, e.g. processed_CR.

Module Contents

src.dackar.RCA.storage.chroma_store.LOGGER

src.dackar.RCA.storage.chroma_store._iter_jsonl(jsonl_path)

Parameters:

jsonl_path (str)

Return type:

Iterable[Dict[str, Any]]

src.dackar.RCA.storage.chroma_store._collapse_ws(s)

Parameters:

s (Optional[str])

Return type:

str

src.dackar.RCA.storage.chroma_store._looks_like_processed_text_record(obj)

Parameters:

obj (Dict[str, Any])

Return type:

bool

src.dackar.RCA.storage.chroma_store.extract_processed_text_record(obj)

Parameters:

obj (Dict[str, Any])

Return type:

Optional[Dict[str, Any]]

src.dackar.RCA.storage.chroma_store._stable_record_id_from_doc(doc)

Parameters:

doc (langchain_core.documents.Document)

Return type:

Optional[str]

src.dackar.RCA.storage.chroma_store.LIST_FIELDS

src.dackar.RCA.storage.chroma_store._sanitize_meta_value(value)

Parameters:

value (Any)

Return type:

Optional[Any]

src.dackar.RCA.storage.chroma_store._normalize_filter_meta(filter_meta)

Normalize high-level retrieval filters into the scalar metadata keys that are actually stored in Chroma.

Examples

• doc_ids -> doc_id

• doc_types -> doc_type

• component_ids is translated in query_doc_type() into primary_component_id

Parameters:

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

Return type:

Dict[str, Any]

src.dackar.RCA.storage.chroma_store._doc_matches_component_ids(doc, wanted)

Return True if doc is associated with at least one of the wanted component IDs.

Checks:

1. The scalar primary_component_id metadata field (index-friendly path).

2. The scalar component_id metadata field (legacy single-component path).

3. The component_ids field. In practice this is a JSON-encoded string on every path, because metadata is passed through _sanitize_meta() (which stringifies lists) before it reaches either Chroma or the in-memory BM25 corpus. The plain-list branch is kept only as a defensive fallback for externally constructed documents.

Parameters:

• doc (langchain_core.documents.Document)

• wanted (set)

Return type:

bool

src.dackar.RCA.storage.chroma_store._doc_matches_filter_sane(doc, filter_sane)

Apply the dense-path Chroma filter semantics to a BM25 hit’s metadata.

The dense path builds {k: {"$in": vals}} for list-valued filters and {k: {"$eq": v}} for scalars. This mirrors that membership-vs-equality logic for the in-memory BM25 post-filter so the two retrieval views agree. Previously BM25 used == even for list-valued filters, which never matched a scalar metadata value and silently dropped every BM25 hit — collapsing hybrid retrieval to dense-only whenever a list filter was set.

Parameters:

• doc (langchain_core.documents.Document)

• filter_sane (Dict[str, Any])

Return type:

bool

src.dackar.RCA.storage.chroma_store._sanitize_meta(meta)

Parameters:

meta (Dict[str, Any])

Return type:

Dict[str, Any]

src.dackar.RCA.storage.chroma_store._record_component_ids(metadata)

Parameters:

metadata (Dict[str, Any])

Return type:

List[str]

src.dackar.RCA.storage.chroma_store.build_chroma_metadata(record)

Flatten a canonical processed_text_record into Chroma-storable metadata.

Chroma metadata values must be primitives (str / int / float / bool). This function projects the nested record structure onto a flat scalar dict that supports both index-level filtering and BM25 text search:

• Copies identity/traceability keys (record_id, doc_id, doc_type, chunk_index, chunk_id, page span, authority level, section role).

• Derives primary_component_id (first component) for index-level component filtering.

• Flattens condition_assessment, eca, oe_metadata, NER entities, and Stage-5 causal statements into ca_* / eca_* / oe_* / *_text scalar keys.

• Sets finding_status from the document type (ECA=confirmed, CR=preliminary, OE=fleet_experience, else observational).

List-valued keys are preserved but JSON-stringified by _sanitize_meta() on return, and companion *_text keys are added for the fields in LIST_FIELDS.

Returns:

A sanitized, primitive-only metadata dict ready to hand to Chroma.

Parameters:

record (Dict[str, Any])

Return type:

Dict[str, Any]

src.dackar.RCA.storage.chroma_store.to_chroma_payload(record)

Convert a processed_text_record into the (id, document, metadata) triple upserted into Chroma.

The vector id is the record’s record_id, the embedded/indexed document text is the whitespace-collapsed embedding_text, and metadata comes from build_chroma_metadata().

Parameters:

record (Dict[str, Any])

Return type:

Dict[str, Any]

src.dackar.RCA.storage.chroma_store.collection_name_for_doc_type(doc_type, prefix='processed')

Return the Chroma collection name for a document type, e.g. processed_CR.

The doc type is upper-cased and sanitized (/ and spaces become _) so it is a valid, stable collection identifier; a falsy doc type maps to OTHER.

Parameters:

• doc_type (str)

• prefix (str)

Return type:

str

class src.dackar.RCA.storage.chroma_store._CollectionState

vectorstore: langchain_chroma.Chroma

bm25_docs: List[langchain_core.documents.Document] = []

bm25: langchain_community.retrievers.BM25Retriever | None = None

class src.dackar.RCA.storage.chroma_store.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]