Coverage for astrocyte/pipeline/pipeline_config.py: 97%
89 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-10-29 22:10 +0000
1"""Feature-flag configuration for :class:`PipelineOrchestrator`.
3Historically the orchestrator was built with base dependencies and *then*
4``Astrocyte.set_pipeline`` reached in and mutated ~30 feature-flag attributes
5one by one from ``AstrocyteConfig`` blocks — a two-phase construction where a
6foreign class finished wiring the orchestrator by poking its internals.
8``PipelineConfig`` collapses that translation into one place. Build it once
9with :meth:`PipelineConfig.from_config` and hand it to
10:meth:`PipelineOrchestrator.apply_config`; the orchestrator no longer has its
11configuration assembled by an outside method. The flag *storage* still lives on
12the orchestrator (so the 3k-line hot paths read ``self.<flag>`` unchanged and
13direct-construction tests keep their constructor defaults), but the *derivation*
14of those flags from config now has a single, testable owner.
15"""
17from __future__ import annotations
19import os
20from dataclasses import dataclass, field
21from typing import TYPE_CHECKING, Any
23if TYPE_CHECKING:
24 from astrocyte.config import AstrocyteConfig, DedupConfig, ExtractionProfileConfig, RecallAuthorityConfig
25 from astrocyte.pipeline.agentic_reflect import AgenticReflectParams
26 from astrocyte.pipeline.cross_encoder_rerank import CrossEncoderProtocol
27 from astrocyte.pipeline.link_expansion import LinkExpansionParams
30def _temporal_expansion_flag(config_default: bool) -> bool:
31 """Resolve query-analyzer temporal expansion, honouring the env override.
33 ``ASTROCYTE_M18_ENABLE_TEMPORAL_EXPANSION`` gives bench-time control without
34 a code change; otherwise the per-bank config value wins.
35 """
36 env = os.environ.get("ASTROCYTE_M18_ENABLE_TEMPORAL_EXPANSION", "").lower()
37 if env in ("1", "true", "yes"):
38 return True
39 if env in ("0", "false", "no"):
40 return False
41 return config_default
44@dataclass(frozen=True)
45class PipelineConfig:
46 """Everything ``set_pipeline`` used to mutate onto the orchestrator, resolved once.
48 Field names mirror the orchestrator attributes they populate, so
49 :meth:`PipelineOrchestrator.apply_config` is a flat, obviously-correct
50 assignment with no per-field translation.
51 """
53 # Extraction / authority.
54 extraction_profiles: dict[str, ExtractionProfileConfig] | None
55 recall_authority: RecallAuthorityConfig | None
57 # Cross-encoder rerank (Hindsight parity). ``cross_encoder`` is a loaded
58 # model handle or ``None`` (heuristic fallback).
59 cross_encoder: CrossEncoderProtocol | None
60 cross_encoder_top_k: int
62 # Causal-link extraction at retain time.
63 causal_links_enabled: bool
64 causal_max_pairs_per_memory: int
65 causal_min_confidence: float
67 # Semantic-kNN graph at retain time (C3a).
68 semantic_link_graph_enabled: bool
69 semantic_link_graph_top_k: int
70 semantic_link_graph_threshold: float
72 # Structured fact extraction at retain time.
73 structured_fact_extraction_enabled: bool
74 structured_fact_extraction_max_facts: int
75 structured_fact_extraction_mode: str
76 structured_fact_extraction_chunk_strategy: str
77 structured_fact_extraction_chunk_max_size: int | None
78 structured_fact_extraction_parallel_chunks: bool
79 structured_fact_extraction_parallel_chunks_max_concurrency: int
81 # Entity co-occurrence link cap.
82 entity_cooccurrence_enabled: bool
83 entity_cooccurrence_max_entities: int
85 # Query analyzer (temporal constraint extraction at recall time).
86 query_analyzer_enabled: bool
87 query_analyzer_allow_llm_fallback: bool
88 query_analyzer_enable_temporal_expansion: bool
90 # Link expansion (C3) — ``None`` skips the post-fusion 3-signal expansion.
91 link_expansion_params: LinkExpansionParams | None
93 # M9 BM25-IDF keyword strategy.
94 bm25_idf_enabled: bool
96 # M10 source-aware retain + recall.
97 source_store: object | None
98 source_retain_provenance: bool
99 source_chunk_expansion: bool
100 source_expansion_score_multiplier: float
101 source_expansion_max_per_hit: int
103 # Adversarial defense.
104 adversarial_abstention_enabled: bool
105 adversarial_abstention_floor: float
106 adversarial_premise_verification_enabled: bool
107 adversarial_premise_min_confidence: float
108 adversarial_prompt_enabled: bool
110 # Agentic reflect loop — ``None`` = legacy single-shot synthesis.
111 agentic_reflect_params: AgenticReflectParams | None
113 # Mental-model service — wires agentic reflect to the configured store.
114 mental_model_service: object | None
116 # Retain-time dedup (``signal_quality.dedup``). A matched MIP rule's
117 # ``dedup.threshold`` / ``dedup.action`` override these per retain.
118 # Defaulted and last so the dataclass stays positionally compatible.
119 dedup_enabled: bool = True
120 dedup_similarity_threshold: float = 0.95
121 #: In the pipeline's vocabulary (``DedupConfig.pipeline_action``).
122 dedup_action: str = "skip_chunk"
123 # Retain-time dedup against the store, not only this process's cache.
124 dedup_consult_store: bool = True
125 # Banks with their own ``banks.<id>.signal_quality`` block: the resolved
126 # dedup settings (already merged over the top-level block) for that bank.
127 dedup_by_bank: dict[str, DedupConfig] = field(default_factory=dict)
129 @classmethod
130 def from_config(
131 cls,
132 config: AstrocyteConfig,
133 *,
134 source_store: object | None = None,
135 mental_model_store: object | None = None,
136 ) -> PipelineConfig:
137 """Derive the full flag set from an :class:`AstrocyteConfig`.
139 Lazy service construction (cross-encoder model load, agentic-reflect and
140 link-expansion params, mental-model service) happens here exactly as it
141 did in the old ``set_pipeline`` body — only when the corresponding block
142 is enabled, so disabled features stay zero-cost.
143 """
144 from astrocyte.pipeline.extraction import merged_extraction_profiles
146 # Cross-encoder reranker — lazy model load, opt-in.
147 cer_cfg = config.cross_encoder_rerank
148 cross_encoder: CrossEncoderProtocol | None = None
149 cross_encoder_top_k = 30
150 if cer_cfg.enabled:
151 from astrocyte.pipeline.cross_encoder_rerank import get_default_cross_encoder
153 cross_encoder = get_default_cross_encoder(
154 cer_cfg.model_name,
155 force_cpu=cer_cfg.force_cpu,
156 )
157 cross_encoder_top_k = cer_cfg.top_k
159 # Link expansion (reuses the legacy ``spreading_activation:`` block).
160 sa_cfg = config.spreading_activation
161 link_expansion_params: LinkExpansionParams | None = None
162 if sa_cfg.enabled:
163 from astrocyte.pipeline.link_expansion import LinkExpansionParams
165 link_expansion_params = LinkExpansionParams(
166 expansion_limit=sa_cfg.expansion_limit,
167 )
169 # Agentic reflect loop.
170 ad_cfg = config.adversarial_defense
171 ar_cfg = config.agentic_reflect
172 agentic_reflect_params: AgenticReflectParams | None = None
173 if ar_cfg.enabled:
174 from astrocyte.pipeline.agentic_reflect import AgenticReflectParams
176 agentic_reflect_params = AgenticReflectParams(
177 max_iterations=ar_cfg.max_iterations,
178 recall_step_max_results=ar_cfg.recall_step_max_results,
179 max_evidence_pool_size=ar_cfg.max_evidence_pool_size,
180 # Mirrors the separate adversarial_defense block so enabling
181 # adversarial defense also tightens the loop's system prompt.
182 adversarial_defense=ad_cfg.adversarial_prompt_enabled,
183 )
185 # Mental-model service — present only when a store is configured.
186 mental_model_service: object | None = None
187 if mental_model_store is not None:
188 from astrocyte.pipeline.mental_model import MentalModelService
190 mental_model_service = MentalModelService(mental_model_store) # type: ignore[arg-type]
192 cl_cfg = config.causal_links
193 slg_cfg = config.semantic_link_graph
194 sfe_cfg = config.structured_fact_extraction
195 coocc_cfg = config.entity_cooccurrence
196 qa_cfg = config.query_analyzer
197 sar_cfg = config.source_aware_retrieval
198 dedup_cfg = config.signal_quality.dedup
200 return cls(
201 extraction_profiles=merged_extraction_profiles(config),
202 recall_authority=config.recall_authority,
203 cross_encoder=cross_encoder,
204 cross_encoder_top_k=cross_encoder_top_k,
205 causal_links_enabled=cl_cfg.enabled,
206 causal_max_pairs_per_memory=cl_cfg.max_pairs_per_memory,
207 causal_min_confidence=cl_cfg.min_confidence,
208 dedup_enabled=dedup_cfg.enabled,
209 dedup_similarity_threshold=dedup_cfg.similarity_threshold,
210 dedup_action=dedup_cfg.pipeline_action,
211 dedup_consult_store=dedup_cfg.consult_store,
212 dedup_by_bank={
213 bank_id: bank.signal_quality.dedup
214 for bank_id, bank in (config.banks or {}).items()
215 if bank.signal_quality is not None
216 },
217 semantic_link_graph_enabled=slg_cfg.enabled,
218 semantic_link_graph_top_k=slg_cfg.top_k,
219 semantic_link_graph_threshold=slg_cfg.similarity_threshold,
220 structured_fact_extraction_enabled=sfe_cfg.enabled,
221 structured_fact_extraction_max_facts=sfe_cfg.max_facts_per_call,
222 structured_fact_extraction_mode=sfe_cfg.extraction_mode,
223 structured_fact_extraction_chunk_strategy=sfe_cfg.chunk_strategy,
224 structured_fact_extraction_chunk_max_size=sfe_cfg.chunk_max_size,
225 structured_fact_extraction_parallel_chunks=sfe_cfg.parallel_chunks,
226 structured_fact_extraction_parallel_chunks_max_concurrency=sfe_cfg.parallel_chunks_max_concurrency,
227 entity_cooccurrence_enabled=coocc_cfg.enabled,
228 entity_cooccurrence_max_entities=coocc_cfg.max_entities_per_memory,
229 query_analyzer_enabled=qa_cfg.enabled,
230 query_analyzer_allow_llm_fallback=qa_cfg.allow_llm_fallback,
231 query_analyzer_enable_temporal_expansion=_temporal_expansion_flag(qa_cfg.enable_temporal_expansion),
232 link_expansion_params=link_expansion_params,
233 bm25_idf_enabled=config.bm25_idf.enabled,
234 source_store=source_store,
235 source_retain_provenance=sar_cfg.retain_provenance,
236 source_chunk_expansion=sar_cfg.chunk_expansion,
237 source_expansion_score_multiplier=sar_cfg.expansion_score_multiplier,
238 source_expansion_max_per_hit=sar_cfg.expansion_max_per_hit,
239 adversarial_abstention_enabled=ad_cfg.abstention_enabled,
240 adversarial_abstention_floor=ad_cfg.abstention_floor,
241 adversarial_premise_verification_enabled=ad_cfg.premise_verification_enabled,
242 adversarial_premise_min_confidence=ad_cfg.premise_verification_min_confidence,
243 adversarial_prompt_enabled=ad_cfg.adversarial_prompt_enabled,
244 agentic_reflect_params=agentic_reflect_params,
245 mental_model_service=mental_model_service,
246 )
248 def as_orchestrator_attrs(self) -> dict[str, Any]:
249 """The flag set as a flat ``{attr_name: value}`` map for assignment."""
250 return dict(self.__dict__)