src.dackar.RCA.orchestrators.evidence_retriever¶

Attributes¶

`LOGGER`
`JsonDict`
`_NEGATION_SINGLE_TRIGGERS`
`_NEGATION_MULTIWORD_TRIGGERS`
`_NEGATABLE_STATE_TERMS`
`_NEGATION_SCOPE_TOKENS`
`_NEGATION_TRAILING_TRIGGERS`
`store`

Classes¶

`_EmbeddingEncoder`	Duck-typed protocol for anything that can embed a list of strings.
`EvidenceStore`	Abstract retrieval backend, e.g. Chroma via LangChain.
`EvidenceRetrieverConfig`
`ChromaEvidenceRetriever`	Deterministic KG-guided evidence retriever.
`InMemoryEvidenceStore`	Development stub that mimics Chroma-style retrieval.

Functions¶

`utcnow_iso`()
`_norm_text`(value)
`_tokenize`(value)
`_overlap_score`(query_terms, text_terms)
`_contains_any`(text, phrases)
`_negation_refutation_hit`(snippet, target_terms)	Return True when a negation trigger is adjacent, within a short window, to a
`_cosine_sim`(a, b)	Cosine similarity between two pre-normalised numpy vectors.

Module Contents¶

src.dackar.RCA.orchestrators.evidence_retriever.LOGGER[source]¶

src.dackar.RCA.orchestrators.evidence_retriever.JsonDict[source]¶

src.dackar.RCA.orchestrators.evidence_retriever.utcnow_iso()[source]¶

Return type:: str

src.dackar.RCA.orchestrators.evidence_retriever._norm_text(value)[source]¶

Parameters:: value (Any)
Return type:: str

src.dackar.RCA.orchestrators.evidence_retriever._tokenize(value)[source]¶

Parameters:: value (Any)
Return type:: List[str]

src.dackar.RCA.orchestrators.evidence_retriever._overlap_score(query_terms, text_terms)[source]¶

Parameters:

query_terms (List[str])
text_terms (List[str])

Return type:

float

src.dackar.RCA.orchestrators.evidence_retriever._contains_any(text, phrases)[source]¶

Parameters:

text (str)
phrases (List[str])

Return type:

bool

src.dackar.RCA.orchestrators.evidence_retriever._NEGATION_SINGLE_TRIGGERS[source]¶

src.dackar.RCA.orchestrators.evidence_retriever._NEGATION_MULTIWORD_TRIGGERS = ('no evidence of', 'no sign of', 'no signs of', 'no indication of', 'no indications of', 'did...[source]¶

src.dackar.RCA.orchestrators.evidence_retriever._NEGATABLE_STATE_TERMS[source]¶

src.dackar.RCA.orchestrators.evidence_retriever._NEGATION_SCOPE_TOKENS = 3[source]¶

src.dackar.RCA.orchestrators.evidence_retriever._NEGATION_TRAILING_TRIGGERS[source]¶

src.dackar.RCA.orchestrators.evidence_retriever._negation_refutation_hit(snippet, target_terms)[source]¶

Return True when a negation trigger is adjacent, within a short window, to a degradation/failure-state term — i.e. the snippet refutes a degradation claim.

Deterministic and high-precision by design (short scope window; state-term targets only). snippet must already be normalised (lowercased, whitespace-collapsed).

Parameters:

snippet (str)
target_terms (frozenset)

Return type:

bool

src.dackar.RCA.orchestrators.evidence_retriever._cosine_sim(a, b)[source]¶

Cosine similarity between two pre-normalised numpy vectors.

Both a and b must already be unit-norm. Returns a float in [0, 1] (clipped to exclude numeric noise below zero).

Parameters:

a (Any)
b (Any)

Return type:

float

class src.dackar.RCA.orchestrators.evidence_retriever._EmbeddingEncoder[source]¶

Bases: Protocol

Duck-typed protocol for anything that can embed a list of strings.

Compatible with SentenceTransformer, langchain embedders, and any object whose encode method accepts List[str] and returns an array-like of shape (N, D).

encode(texts)[source]¶

Parameters:: texts (List[str])
Return type:: Any

class src.dackar.RCA.orchestrators.evidence_retriever.EvidenceStore[source]¶

Bases: Protocol

Abstract retrieval backend, e.g. Chroma via LangChain.

query(query_text, top_k=10, filters=None)[source]¶

Parameters:

query_text (str)
top_k (int)
filters (Optional[Dict[str, Any]])

Return type:

List[Dict[str, Any]]

class src.dackar.RCA.orchestrators.evidence_retriever.EvidenceRetrieverConfig[source]¶

top_k_total: int = 10[source]¶

top_k_per_query: int = 5[source]¶

include_doc_id_filter: bool = True[source]¶

include_asset_filter: bool = True[source]¶

include_component_filter: bool = True[source]¶

include_doc_type_filter: bool = True[source]¶

score_threshold: float = 0.0[source]¶

score_metric: str = 'kg_guided_semantic_relevance'[source]¶

contradiction_cues: List[str] | None = None[source]¶

structural_contradiction_cues: List[str] | None = None[source]¶

support_cues: List[str] | None = None[source]¶

contextual_cues: List[str] | None = None[source]¶

doc_type_priority: Dict[str, float] | None = None[source]¶

__post_init__()[source]¶

Return type:: None

class src.dackar.RCA.orchestrators.evidence_retriever.ChromaEvidenceRetriever(store, config=None, annotator=None, encoder=None)[source]¶

Deterministic KG-guided evidence retriever.

Parameters:

store (EvidenceStore)
config (Optional[EvidenceRetrieverConfig])
encoder (Optional[_EmbeddingEncoder])

store[source]¶

config[source]¶

annotator = None[source]¶

encoder = None[source]¶

_emb_cache: Dict[str, Any][source]¶

_embed(text)[source]¶

Return a unit-norm embedding vector for text, or None if no encoder.

Results are cached in self._emb_cache (reset at the start of each retrieve() call) so each unique text is encoded at most once per retrieval session.

Parameters:: text (str)
Return type:: Optional[Any]

retrieve(event, kg_context, causality_candidates, operational_context, run_context)[source]¶

Retrieve and rank documentary evidence for the causal candidates.

Builds per-candidate queries from the KG context, runs them against the Chroma store, normalizes and de-duplicates the hits, assesses each hit’s support role against its candidate, and summarizes the evidence per candidate.

Parameters:

event (JsonDict) – Target abnormal event (supplies asset_id and query terms).
kg_context (JsonDict) – KG neighbourhood providing documents and components to query over.
causality_candidates (JsonDict) – Candidate hypotheses whose cause labels seed the queries.
operational_context (Optional[JsonDict]) – Optional operating-state input influencing query planning, or None.
run_context (JsonDict) – Orchestrator run context.

Returns:

Evidence bundle conforming to schemas/evidence_bundle.json. Each entry in results carries top-level snippet_id / support_score and, under metadata, support_role and the linked_candidate_id used by downstream stages.

Return type:

JsonDict

static _component_filter_mode(*, merged_hits, component_ids_requested)[source]¶

Parameters:

merged_hits (Sequence[JsonDict])
component_ids_requested (bool)

Return type:

str

_build_queries(event, kg_context, causality_candidates, operational_context)[source]¶

Parameters:

event (JsonDict)
kg_context (JsonDict)
causality_candidates (JsonDict)
operational_context (Optional[JsonDict])

Return type:

List[JsonDict]

static _candidate_component_ids(candidate, kg_context)[source]¶

Parameters:

candidate (JsonDict)
kg_context (JsonDict)

Return type:

List[str]

_build_operational_context_query(asset_id, operational_context)[source]¶

Parameters:

asset_id (str)
operational_context (Optional[JsonDict])

Return type:

Optional[JsonDict]

_build_filters(asset_id, query_plan, kg_context)[source]¶

Parameters:

asset_id (Optional[str])
query_plan (JsonDict)
kg_context (JsonDict)

Return type:

Dict[str, Any]

_assess_hit_against_candidate(hit, query_plan, cause_label_emb=None)[source]¶

Parameters:

hit (JsonDict)
query_plan (JsonDict)
cause_label_emb (Optional[Any])

Return type:

JsonDict

_normalize_hits(hits, query_plan)[source]¶

Parameters:

hits (Sequence[JsonDict])
query_plan (JsonDict)

Return type:

List[JsonDict]

_build_candidate_evidence_summary(hits)[source]¶

Parameters:: hits (Sequence[JsonDict])
Return type:: List[JsonDict]

static _build_pipeline_health(*, planned_queries, merged_hits, retrieval_mode)[source]¶

Parameters:

planned_queries (Sequence[JsonDict])
merged_hits (Sequence[JsonDict])
retrieval_mode (str)

Return type:

JsonDict

_dedupe_and_rank(hits)[source]¶

Parameters:: hits (Sequence[JsonDict])
Return type:: List[JsonDict]

class src.dackar.RCA.orchestrators.evidence_retriever.InMemoryEvidenceStore(rows=None)[source]¶

Development stub that mimics Chroma-style retrieval.

Parameters:: rows (Optional[List[JsonDict]])

rows = [][source]¶

add(row)[source]¶

Parameters:: row (JsonDict)
Return type:: None

add_documents(docs)[source]¶

Accept CMMS-style Chroma document payloads and normalize into row format.

Parameters:: docs (List[JsonDict])
Return type:: int

query(query_text, top_k=10, filters=None)[source]¶

Parameters:

query_text (str)
top_k (int)
filters (Optional[Dict[str, Any]])

Return type:

List[JsonDict]

src.dackar.RCA.orchestrators.evidence_retriever.store[source]¶