Coverage for astrocyte/pipeline/recall_stage.py: 86%
251 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-10-29 22:10 +0000
1from __future__ import annotations
3import asyncio
4import logging
5from datetime import datetime, timezone
6from typing import TYPE_CHECKING, Any
8from astrocyte.pipeline.embedding import generate_embeddings
9from astrocyte.pipeline.entity_extraction import extract_entities
10from astrocyte.pipeline.fusion import (
11 ScoredItem,
12 memory_hits_as_scored,
13 rrf_fusion,
14 weighted_rrf_fusion,
15)
16from astrocyte.pipeline.hyde import generate_hyde_vector
17from astrocyte.pipeline.link_expansion import link_expansion
18from astrocyte.pipeline.query_intent import (
19 QueryIntent,
20 classify_query_intent,
21 weights_for_intent,
22)
23from astrocyte.pipeline.query_plan import build_query_plan
24from astrocyte.pipeline.reranking import (
25 apply_context_diversity,
26 basic_rerank,
27 llm_pairwise_rerank,
28)
29from astrocyte.pipeline.retrieval import parallel_retrieve
30from astrocyte.policy.homeostasis import enforce_token_budget
31from astrocyte.types import (
32 MemoryHit,
33 RecallRequest,
34 RecallResult,
35 RecallTrace,
36 VectorFilters,
37 VectorItem,
38)
40if TYPE_CHECKING:
41 pass
45from astrocyte.pipeline._orchestrator_common import (
46 _deterministic_names,
47 _warn_on_version_drift,
48)
49from astrocyte.pipeline.provenance import stored_source
51_logger = logging.getLogger("astrocyte.mip")
54class RecallStageMixin:
55 """Recall pipeline: parallel retrieve → fuse → expand → rerank.
57 A behavior slice of PipelineOrchestrator, composed into it as a mixin.
58 ``self`` is the full orchestrator at runtime (attributes come from its
59 ``__init__`` / ``apply_config``); shared helpers live on sibling mixins.
60 """
62 async def _expand_via_sibling_chunks(
63 self,
64 fused: list,
65 bank_id: str,
66 ) -> list:
67 """M10: extend ``fused`` with vectors from sibling chunks of each top-K hit.
69 Walks the top-K hits, asks the SourceStore for the sibling chunks
70 of each hit's ``chunk_id`` (capped at ``expansion_max_per_hit``),
71 then asks the vector store for the vectors backing those chunks.
72 Each expanded hit's score is multiplied by
73 ``expansion_score_multiplier`` so the seed hits stay ranked
74 higher (the multiplier is < 1.0 by default).
76 Returns the merged list, sorted by score descending. Existing
77 memory ids are NOT duplicated — siblings whose vector is already
78 in ``fused`` are skipped.
79 """
80 if not fused:
81 return fused
82 store = self.source_store
83 if store is None:
84 return fused
86 # 1. Collect (chunk_id, doc_id?) for the top-K seeds. We process
87 # only the top ``source_expansion_max_per_hit * 4`` because the
88 # multiplier already shrinks tail-hit gains and we want to keep
89 # the per-recall fan-out bounded.
90 seed_cap = max(1, self.source_expansion_max_per_hit * 4)
91 seed_chunk_ids = [getattr(h, "chunk_id", None) for h in fused[:seed_cap] if getattr(h, "chunk_id", None)]
92 if not seed_chunk_ids:
93 return fused
95 # 2. Resolve each seed chunk → its document, then list siblings
96 # of that document. Run lookups in parallel; tolerate failures.
97 async def _siblings_for(chunk_id: str) -> list[str]:
98 try:
99 chunk = await store.get_chunk(chunk_id, bank_id) # type: ignore[union-attr]
100 if chunk is None:
101 return []
102 siblings = await store.list_chunks(chunk.document_id, bank_id) # type: ignore[union-attr]
103 # Drop the seed chunk itself; cap at expansion_max_per_hit.
104 ids = [c.id for c in siblings if c.id != chunk_id][: self.source_expansion_max_per_hit]
105 return ids
106 except Exception as exc:
107 _logger.debug("source_store sibling lookup failed for %s: %s", chunk_id, exc)
108 return []
110 sibling_lists = await asyncio.gather(*(_siblings_for(cid) for cid in seed_chunk_ids))
111 sibling_chunk_ids: set[str] = set()
112 for ids in sibling_lists:
113 sibling_chunk_ids.update(ids)
114 # Drop any chunk_ids already represented in ``fused`` so we don't
115 # re-rank the same memory twice.
116 already_present = {getattr(h, "chunk_id", None) for h in fused if getattr(h, "chunk_id", None)}
117 sibling_chunk_ids -= already_present
118 if not sibling_chunk_ids:
119 return fused
121 # 3. Pull the sibling vectors. The vector store returns score=1.0;
122 # we apply the configured multiplier so seeds stay ranked higher.
123 try:
124 sibling_hits = await self.vector_store.get_by_chunk_ids( # type: ignore[attr-defined]
125 list(sibling_chunk_ids),
126 bank_id,
127 )
128 except Exception as exc:
129 _logger.warning("vector_store.get_by_chunk_ids failed; skipping chunk expansion: %s", exc)
130 return fused
132 if not sibling_hits:
133 return fused
135 existing_ids = {getattr(h, "id", None) for h in fused}
136 boost = self.source_expansion_score_multiplier
137 for sh in sibling_hits:
138 if sh.id in existing_ids:
139 continue
140 sh.score = max(0.0, min(1.0, sh.score * boost))
141 fused.append(sh)
143 # Re-sort: seeds keep their fused score, expansions slot in by
144 # the boosted score. The downstream rerank stage gets the final say.
145 fused.sort(key=lambda h: getattr(h, "score", 0.0), reverse=True)
146 return fused
148 async def recall(self, request: RecallRequest) -> RecallResult:
149 """Recall pipeline: embed query → parallel retrieve → fuse → rerank → budget."""
150 # 1. Embed query
151 query_embeddings = await generate_embeddings([request.query], self.llm_provider)
152 query_vector = query_embeddings[0]
154 # 1b. Wiki-tier precedence (M8 W5) — search compiled wiki pages first.
155 # If the top hit scores above wiki_confidence_threshold, return wiki hits
156 # + raw-memory citations and skip the full parallel-retrieve pipeline.
157 # Caller can bypass this tier by setting fact_types (which implies they
158 # want raw memories, not compiled wiki pages).
159 if self.wiki_store is not None and not request.fact_types:
160 wiki_result = await self._try_wiki_tier(request, query_vector)
161 if wiki_result is not None:
162 return wiki_result
164 # 2. Build filters
165 filters = VectorFilters(
166 bank_id=request.bank_id,
167 tags=request.tags,
168 fact_types=request.fact_types,
169 time_range=request.time_range,
170 as_of=request.as_of, # M9: time-travel filter
171 session_id=request.session_id, # M31 Fix 2
172 )
174 # 2a. Query-level temporal constraint extraction. The analyzer
175 # parses expressions like "what happened in March 2024?" or
176 # "last week's events" into a (start, end) range that becomes
177 # an extra time_range filter on top of any caller-supplied one.
178 # When the request already supplies a time_range, the
179 # analyzer's hit is the more restrictive of the two — the
180 # caller's filter remains the floor.
181 if (
182 self.query_analyzer_enabled and request.time_range is None # caller-supplied range wins
183 ):
184 try:
185 from astrocyte.pipeline.query_analyzer import analyze_query
187 # Resolution order for the relative-phrase anchor:
188 # 1. ``query_reference_date`` if explicitly set
189 # (preferred — see types.py for rationale)
190 # 2. ``as_of`` if set (back-compat: M9 audit-replay
191 # callers got temporal anchoring "for free" before
192 # the split — preserve that)
193 # 3. ``None`` → analyze_query falls back to ``now()``
194 anchor = request.query_reference_date or request.as_of
195 analysis = await analyze_query(
196 request.query,
197 reference_date=anchor,
198 llm_provider=self.llm_provider,
199 allow_llm_fallback=self.query_analyzer_allow_llm_fallback,
200 allow_temporal_expansion=self.query_analyzer_enable_temporal_expansion,
201 )
202 if analysis.temporal_constraint and analysis.temporal_constraint.is_bounded():
203 c = analysis.temporal_constraint
204 # VectorFilters.time_range is (start, end) tuple,
205 # both required for the SQL/in-memory adapters.
206 # Use bench-bank min/max as defaults for open
207 # ranges so the adapter contract is satisfied.
208 far_past = datetime(1900, 1, 1, tzinfo=timezone.utc)
209 far_future = datetime(2200, 1, 1, tzinfo=timezone.utc)
210 filters = VectorFilters(
211 bank_id=filters.bank_id,
212 tags=filters.tags,
213 fact_types=filters.fact_types,
214 time_range=(
215 c.start_date or far_past,
216 c.end_date or far_future,
217 ),
218 as_of=filters.as_of,
219 session_id=filters.session_id, # M31 Fix 2 — preserve
220 )
221 _logger.info(
222 "query_analyzer extracted temporal range %s",
223 analysis.temporal_constraint,
224 )
225 except Exception as exc: # pragma: no cover — defensive
226 _logger.warning(
227 "query_analyzer failed (%s); continuing without temporal filter.",
228 exc,
229 )
231 # 2b. Extract entities from query for graph search
232 entity_ids: list[str] | None = None
233 if self.graph_store:
234 query_entities = await extract_entities(request.query, self.llm_provider)
235 if query_entities:
236 # Look up entity IDs in the graph store
237 found: list[str] = []
238 for ent in query_entities:
239 matches = await self.graph_store.query_entities(ent.name, request.bank_id, limit=3)
240 found.extend(m.id for m in matches)
241 entity_ids = found or None
243 # Resolve per-bank MIP PipelineSpec once (reused for temporal
244 # half-life override at retrieval time AND for rerank + version
245 # check further down). Previously resolved after retrieval — moved
246 # earlier so temporal_half_life_days can flow into parallel_retrieve.
247 bank_pipeline = None
248 if self.mip_router is not None:
249 bank_pipeline = self.mip_router.resolve_pipeline_for_bank(request.bank_id)
251 # Per-bank half-life override beats the orchestrator default.
252 # Long-term knowledge banks (e.g. LongMemEval-style corpora with
253 # months-old answers) set this to 90+; chat banks leave it at 7.
254 effective_half_life = (
255 bank_pipeline.temporal_half_life_days
256 if bank_pipeline is not None and bank_pipeline.temporal_half_life_days is not None
257 else self.temporal_half_life_days
258 )
260 # 2c. HyDE (R1) — generate hypothetical document embedding.
261 # Runs concurrently with the retrieval step below via a separate task.
262 # Failures return None; parallel_retrieve treats None as "HyDE disabled".
263 hyde_vec: list[float] | None = None
264 if self.enable_hyde:
265 hyde_vec = await generate_hyde_vector(request.query, self.llm_provider)
267 # 3. Parallel retrieval
268 overfetch_limit = request.max_results * self.semantic_overfetch
269 strategy_timings_ms: dict[str, float] = {}
270 strategy_candidate_counts: dict[str, int] = {}
271 strategy_results = await parallel_retrieve(
272 query_vector=query_vector,
273 query_text=request.query,
274 bank_id=request.bank_id,
275 vector_store=self.vector_store,
276 graph_store=self.graph_store,
277 document_store=self.document_store,
278 entity_ids=entity_ids,
279 limit=overfetch_limit,
280 filters=filters,
281 enable_temporal=self.enable_temporal_retrieval,
282 temporal_scan_cap=self.temporal_scan_cap,
283 temporal_half_life_days=effective_half_life,
284 hyde_vector=hyde_vec,
285 strategy_timings_ms=strategy_timings_ms,
286 strategy_candidate_counts=strategy_candidate_counts,
287 use_bm25_idf=self.bm25_idf_enabled,
288 )
289 query_plan = build_query_plan(request.query)
290 if not request.fact_types and query_plan.needs_multi_hop_synthesis:
291 entity_path_hits = await self._retrieve_entity_path_fallback(
292 request.query,
293 request.bank_id,
294 limit=overfetch_limit,
295 )
296 if entity_path_hits:
297 strategy_results["entity_path"] = entity_path_hits
299 # 3b. Intent classification — hoisted here so both observation injection
300 # (step 3b) and RRF weighting (step 4) share the same result without
301 # computing it twice. UNKNOWN → neutral 1.0 weights everywhere.
302 query_intent = QueryIntent.UNKNOWN
303 if self.enable_intent_aware_recall:
304 query_intent = classify_query_intent(request.query).intent
305 intent_weights = weights_for_intent(query_intent)
307 # 3c. Observation strategy — intent-gated injection.
308 # The ::obs bank holds distilled, multi-evidence facts synthesised from
309 # raw memories. Injecting it for *every* recall degrades factual
310 # precision (abstract summaries displace verbatim answers). Instead,
311 # inject only for EXPLORATORY and RELATIONAL queries — the two intents
312 # where synthesised behavioural patterns add value over raw memories:
313 # EXPLORATORY: "What are Alice's hobbies?" / "Describe Bob's personality"
314 # RELATIONAL: "How does Alice's role relate to her projects?"
315 # For all other intents, effective_obs_weight falls back to the
316 # configured observation_weight (default 0.0 = disabled).
317 from astrocyte.pipeline.reflect import _auto_prompt_variant
319 _OBS_INJECTION_INTENTS = {QueryIntent.EXPLORATORY, QueryIntent.RELATIONAL}
320 prompt_variant = _auto_prompt_variant(request.query)
321 effective_obs_weight = (
322 self.observation_injection_weight
323 if query_intent in _OBS_INJECTION_INTENTS or prompt_variant == "evidence_inference"
324 else self.observation_weight
325 )
326 if self._observation_consolidator is not None and not request.fact_types and effective_obs_weight > 0.0:
327 obs_results = await self._retrieve_observations(
328 query_vector, request.bank_id, overfetch_limit, request.as_of
329 )
330 if obs_results:
331 strategy_results["observation"] = obs_results
333 # 4. RRF fusion (local strategies + optional federated / manual external_context)
334 weighted_inputs: list[tuple[list[Any], float]] = []
335 for strategy, results in strategy_results.items():
336 if not results:
337 continue
338 # Observation strategy uses the effective intent-gated weight;
339 # other strategies use intent-derived weights.
340 if strategy == "observation":
341 weight = effective_obs_weight
342 else:
343 # Map strategy name → weight. External/proxy results use
344 # semantic weight (they're typically semantic fusions upstream).
345 weight = getattr(intent_weights, strategy, 1.0)
346 weighted_inputs.append((results, weight))
347 if request.external_context:
348 weighted_inputs.append(
349 (memory_hits_as_scored(request.external_context), intent_weights.semantic),
350 )
352 # Short-circuit to plain RRF when all weights are 1.0 — keeps the
353 # baseline path bit-identical for deployments that disable
354 # intent-aware recall.
355 if all(w == 1.0 for _, w in weighted_inputs):
356 fused = rrf_fusion([r for r, _ in weighted_inputs], k=self.rrf_k)
357 else:
358 fused = weighted_rrf_fusion(weighted_inputs, k=self.rrf_k)
360 # 4a. Link expansion (Hindsight parity, C3). After initial RRF
361 # surfaces direct hits, query the three first-class memory-link
362 # signals — entity overlap, semantic kNN, causal — and merge
363 # the resulting candidates back into the fused set. Replaces
364 # the previous BFS-hop spreading-activation path.
365 if self.link_expansion_params is not None and self.graph_store is not None and fused:
366 try:
367 expansion_hits = await link_expansion(
368 fused[: self.link_expansion_params.expansion_limit],
369 bank_id=request.bank_id,
370 vector_store=self.vector_store,
371 graph_store=self.graph_store,
372 params=self.link_expansion_params,
373 tags=request.tags,
374 )
375 except Exception as exc: # pragma: no cover — defensive
376 _logger.warning(
377 "link expansion failed (%s); continuing with direct fused hits only.",
378 exc,
379 )
380 expansion_hits = []
381 if expansion_hits:
382 # Re-fuse via RRF so direct evidence keeps its precedence
383 # (top-1 direct beats top-1 expansion by construction).
384 fused = rrf_fusion([fused, expansion_hits], k=self.rrf_k)
386 # 4b. Multi-query expansion: decompose the question into sub-questions,
387 # recall for each independently (parallel), and merge all fused lists
388 # via a final RRF pass. Only runs when enabled and the LLM judges the
389 # question as multi-hop (len > 1). The original query's fused result is
390 # always included so the expansion never discards the baseline recall.
391 #
392 # Two guards:
393 # (a) Empty-bank guard — if initial retrieval returned nothing,
394 # decomposition cannot help and we avoid a wasted LLM call.
395 # (b) Confidence gate — peek at the top raw semantic score *before*
396 # fusion. Cosine similarity is the cleanest signal for "did we
397 # already find the answer?": if the top hit already exceeds
398 # multi_query_confidence_threshold we skip decomposition entirely,
399 # preventing broad sub-queries from displacing the precise answer.
400 # RRF scores are rank-based and not used here.
401 _top_semantic_score: float = 0.0
402 if "semantic" in strategy_results and strategy_results["semantic"]:
403 _top_semantic_score = strategy_results["semantic"][0].score
405 if self.enable_multi_query_expansion and fused and _top_semantic_score < self.multi_query_confidence_threshold:
406 from astrocyte.pipeline.multi_query import decompose_query
408 sub_queries = await decompose_query(request.query, self.llm_provider)
409 if len(sub_queries) > 1:
411 async def _fuse_sub_query(sq: str) -> list[Any]:
412 sq_vec = (await generate_embeddings([sq], self.llm_provider))[0]
413 sq_strategy_results = await parallel_retrieve(
414 query_vector=sq_vec,
415 query_text=sq,
416 bank_id=request.bank_id,
417 vector_store=self.vector_store,
418 graph_store=self.graph_store,
419 document_store=self.document_store,
420 entity_ids=entity_ids,
421 limit=overfetch_limit,
422 filters=filters,
423 enable_temporal=self.enable_temporal_retrieval,
424 temporal_scan_cap=self.temporal_scan_cap,
425 temporal_half_life_days=effective_half_life,
426 use_bm25_idf=self.bm25_idf_enabled,
427 )
428 sq_intent = (
429 classify_query_intent(sq).intent if self.enable_intent_aware_recall else QueryIntent.UNKNOWN
430 )
431 sq_weights = weights_for_intent(sq_intent)
432 sq_weighted = [
433 (res, getattr(sq_weights, strat, 1.0)) for strat, res in sq_strategy_results.items() if res
434 ]
435 if not sq_weighted:
436 return []
437 if all(w == 1.0 for _, w in sq_weighted):
438 return rrf_fusion([r for r, _ in sq_weighted], k=self.rrf_k)
439 return weighted_rrf_fusion(sq_weighted, k=self.rrf_k)
441 # sub_queries[0] is the original (already in fused); expand the rest
442 sub_fused_lists = await asyncio.gather(*[_fuse_sub_query(sq) for sq in sub_queries[1:]])
443 non_empty = [sf for sf in sub_fused_lists if sf]
444 if non_empty:
445 fused = rrf_fusion([fused, *non_empty], k=self.rrf_k)
447 # 4b. M10 chunk expansion — for each top-K hit with a chunk_id,
448 # fetch sibling chunks (other vectors from the same SourceDocument)
449 # and merge them into the candidate pool. Helps multi-hop / split-
450 # evidence questions where the answer key is in chunk N±1 of a
451 # chunk that hit. No-op when source_store is unwired or the flag
452 # is off, or when the vector store doesn't expose
453 # ``get_by_chunk_ids`` (older adapters degrade gracefully).
454 if (
455 self.source_chunk_expansion
456 and self.source_store is not None
457 and hasattr(self.vector_store, "get_by_chunk_ids")
458 ):
459 fused = await self._expand_via_sibling_chunks(fused, request.bank_id)
461 # 5. Reranking — apply per-bank MIP RerankSpec when a rule targets this bank (P3)
462 # (bank_pipeline already resolved above for temporal half-life override)
463 mip_rerank = bank_pipeline.rerank if bank_pipeline is not None else None
464 reranked = basic_rerank(fused, request.query, mip_rerank=mip_rerank)
465 if self.final_rerank_mode == "llm_pairwise":
466 reranked = await llm_pairwise_rerank(
467 reranked,
468 request.query,
469 self.llm_provider,
470 top_n=self.final_rerank_top_n,
471 keep_n=self.final_rerank_keep_n or len(reranked),
472 )
473 else:
474 reranked = apply_context_diversity(reranked, request.query)
476 # 6. Trim to max_results
477 trimmed = reranked[: request.max_results]
479 # 7. Convert to MemoryHit
480 hits = [
481 MemoryHit(
482 text=item.text,
483 score=item.score,
484 fact_type=item.fact_type,
485 metadata=item.metadata,
486 tags=item.tags,
487 memory_id=item.id,
488 bank_id=request.bank_id,
489 occurred_at=getattr(item, "occurred_at", None),
490 retained_at=getattr(item, "retained_at", None), # M9
491 chunk_id=getattr(item, "chunk_id", None), # M10
492 source=stored_source(item.metadata),
493 )
494 for item in trimmed
495 ]
497 # 7b. Version-drift warning (Phase 2, Step 10)
498 # Compare each hit's persisted ``_mip.pipeline_version`` against the
499 # version currently configured for this bank's rule. Stale hits indicate
500 # rule changes since retain — operator should re-index or accept drift.
501 _warn_on_version_drift(bank_pipeline, hits, request.bank_id)
503 # 8. Token budget
504 truncated = False
505 if request.max_tokens:
506 hits, truncated = enforce_token_budget(hits, request.max_tokens)
508 ext_n = len(request.external_context) if request.external_context else 0
509 strategies_used = list(strategy_results.keys())
510 if ext_n:
511 strategies_used.append("proxy")
512 total_candidates = sum(len(r) for r in strategy_results.values()) + ext_n
514 return RecallResult(
515 hits=hits,
516 total_available=len(fused),
517 truncated=truncated,
518 trace=RecallTrace(
519 strategies_used=strategies_used,
520 total_candidates=total_candidates,
521 fusion_method="rrf",
522 strategy_timings_ms=strategy_timings_ms or None,
523 strategy_candidate_counts=strategy_candidate_counts or None,
524 ),
525 top_semantic_score=_top_semantic_score,
526 )
528 async def _retrieve_observations(
529 self,
530 query_vector: list[float],
531 bank_id: str,
532 limit: int,
533 as_of: Any | None,
534 ) -> list[Any]:
535 """Retrieve observation-layer hits for RRF fusion.
537 Observations are stored in a dedicated bank (``{bank_id}::obs``) that
538 is completely separate from the raw memory bank. This prevents
539 double-counting: the main semantic/keyword/temporal strategies only
540 search the raw bank, while this method exclusively searches the obs
541 bank. Converts ``VectorHit`` objects to ``ScoredItem`` to match the
542 format expected by the weighted RRF fusion step.
543 """
544 from astrocyte.pipeline.fusion import ScoredItem
545 from astrocyte.pipeline.observation import obs_bank_id
547 obs_bank = obs_bank_id(bank_id)
548 obs_filters = VectorFilters(
549 bank_id=obs_bank,
550 as_of=as_of,
551 )
552 try:
553 hits = await self.vector_store.search_similar(query_vector, obs_bank, limit=limit, filters=obs_filters)
554 except Exception as exc:
555 _logger.warning("Observation retrieval failed for bank %s: %s", bank_id, exc)
556 return []
558 return [
559 ScoredItem(
560 id=h.id,
561 text=h.text,
562 score=h.score,
563 fact_type=h.fact_type,
564 metadata=h.metadata,
565 tags=h.tags,
566 memory_layer="observation",
567 occurred_at=getattr(h, "occurred_at", None),
568 retained_at=getattr(h, "retained_at", None),
569 )
570 for h in hits
571 ]
573 async def _retrieve_entity_path_fallback(
574 self,
575 query: str,
576 bank_id: str,
577 *,
578 limit: int,
579 ) -> list[ScoredItem]:
580 """In-memory entity-path recall when no graph backend is configured."""
582 query_names = _deterministic_names(query)
583 if not query_names:
584 return []
585 try:
586 items = await self.vector_store.list_vectors(bank_id, offset=0, limit=max(limit * 10, 200))
587 except Exception:
588 return []
590 hits: list[ScoredItem] = []
591 for item in items:
592 metadata = dict(item.metadata or {})
593 text_names = _deterministic_names(item.text)
594 metadata_names = {
595 part.strip().lower()
596 for key in ("locomo_persons", "locomo_speakers", "person")
597 for part in str(metadata.get(key) or "").replace("|", ",").split(",")
598 if part.strip()
599 }
600 matched = query_names & (text_names | metadata_names)
601 if not matched:
602 continue
603 metadata["_entity_path"] = " -> ".join(sorted(matched))
604 metadata["_entity_path_kind"] = "metadata_fallback"
605 hits.append(
606 ScoredItem(
607 id=item.id,
608 text=item.text,
609 score=0.65 + min(len(matched), 3) * 0.05,
610 fact_type=item.fact_type,
611 metadata=metadata,
612 tags=item.tags,
613 memory_layer=item.memory_layer,
614 occurred_at=item.occurred_at,
615 retained_at=item.retained_at,
616 )
617 )
618 return sorted(hits, key=lambda hit: hit.score, reverse=True)[:limit]
620 async def _try_wiki_tier(
621 self,
622 request: RecallRequest,
623 query_vector: list[float],
624 ) -> RecallResult | None:
625 """Search compiled wiki pages and return hits if the top score meets the threshold.
627 Returns ``None`` when:
628 - No wiki pages exist in the bank.
629 - The top wiki score is below ``wiki_confidence_threshold``.
630 - The vector store raises (wiki tier is non-fatal; full recall continues).
632 When a wiki hit is returned the result also includes raw-memory citations
633 derived from ``_wiki_source_ids`` stored in VectorItem metadata during
634 compile. The citations are returned as additional MemoryHits with
635 ``memory_layer="raw"`` so callers can distinguish synthesised wiki
636 content from the underlying evidence.
637 """
638 try:
639 wiki_filters = VectorFilters(
640 bank_id=request.bank_id,
641 tags=request.tags,
642 fact_types=["wiki"],
643 )
644 wiki_hits = await self.vector_store.search_similar(
645 query_vector,
646 request.bank_id,
647 limit=request.max_results,
648 filters=wiki_filters,
649 )
650 except Exception:
651 _logger.debug("Wiki-tier search failed; falling back to standard recall", exc_info=True)
652 return None
654 if not wiki_hits or wiki_hits[0].score < self.wiki_confidence_threshold:
655 return None
657 # Convert wiki VectorHits → MemoryHits
658 hits: list[MemoryHit] = [
659 MemoryHit(
660 text=h.text,
661 score=h.score,
662 fact_type=h.fact_type,
663 metadata=h.metadata,
664 tags=h.tags,
665 memory_id=h.id,
666 bank_id=request.bank_id,
667 memory_layer="compiled",
668 source=stored_source(h.metadata),
669 )
670 for h in wiki_hits
671 ]
673 # Append raw-memory citations from source_ids stored in metadata
674 citation_ids: list[str] = []
675 for h in wiki_hits:
676 raw_ids_str = (h.metadata or {}).get("_wiki_source_ids", "")
677 if raw_ids_str:
678 citation_ids.extend(raw_ids_str.split(","))
680 if citation_ids:
681 # Fetch raw memories by scanning the vector store.
682 # We use list_vectors (paginated) to find matches by ID — the
683 # VectorStore SPI has no get_by_ids, so we scan once and filter.
684 raw_map: dict[str, VectorItem] = {}
685 offset = 0
686 batch = 100
687 target = set(citation_ids)
688 while target:
689 chunk = await self.vector_store.list_vectors(request.bank_id, offset=offset, limit=batch)
690 if not chunk:
691 break
692 for item in chunk:
693 if item.id in target:
694 raw_map[item.id] = item
695 target.discard(item.id)
696 if len(chunk) < batch:
697 break
698 offset += batch
700 for cid in citation_ids:
701 item = raw_map.get(cid)
702 if item is None:
703 continue
704 hits.append(
705 MemoryHit(
706 text=item.text,
707 score=0.0, # citations are provenance, not ranked hits
708 fact_type=item.fact_type,
709 metadata=item.metadata,
710 tags=item.tags,
711 memory_id=item.id,
712 bank_id=request.bank_id,
713 memory_layer="raw",
714 )
715 )
717 # Apply token budget if requested
718 truncated = False
719 if request.max_tokens:
720 hits, truncated = enforce_token_budget(hits, request.max_tokens)
722 return RecallResult(
723 hits=hits[: request.max_results],
724 total_available=len(wiki_hits),
725 truncated=truncated,
726 trace=RecallTrace(
727 strategies_used=["wiki"],
728 total_candidates=len(wiki_hits),
729 fusion_method="wiki_tier",
730 wiki_tier_used=True,
731 ),
732 )