"""``node_search_step`` — node-level digest search for dream Phase 2. Specialized for dream's recall needs; **NOT** a drop-in replacement for the general-purpose ``search`` step (which serves external RAG agents). Five differences vs ``search``: 1. **Node-level results** — same-path chunks aggregated by max score; one row per digest node, not per chunk. 2. **Frontmatter included** — returns ``name + description`` inline so the caller can triage without a follow-up ``frontmatter_read`` per hit. 3. **Digest-only filter** — hardcoded to ``/`` prefix; dream never wants daily / resource hits as recall candidates. 4. **No expand_links** — dream's synapse recall is looking for nodes that *don't* yet have wikilinks; expansion would surface already-linked neighbors (anti-pattern for synapse construction). 5. **No body / chunk text in response** — caller follows up with ``read`` only on the few candidates that need deep inspection, not all. A single hybrid (vector + BM25 RRF) recall serves **both** the dedup judgment (`same_abstraction` label — is any candidate the same as the new unit?) and the synapse judgment (`related` label — which candidates should be woven as wikilinks?). They are two LLM-internal labels over the **same candidate pool**; there is no need to split into separate recall passes (the previous ``mode={hybrid,vector_only}`` toggle was spurious — both judgments operate on the same recall output). Used in dream Phase 2 (dream.yaml ``integrate_system_prompt_*``). External agents must keep using ``search`` for chunk-level retrieval + link expansion. """ import asyncio from collections import defaultdict from ..base_step import BaseStep from ...components import R _RRF_K = 60 _MAX_CANDIDATES = 200 def _rrf_merge_nodes( vector_paths: list[str], keyword_paths: list[str], vector_weight: float, ) -> dict[str, float]: """RRF fuse two ranked path lists into node-level scores.""" scores: dict[str, float] = defaultdict(float) text_weight = 1.0 - vector_weight for rank, path in enumerate(vector_paths, start=1): scores[path] += vector_weight / (_RRF_K + rank) for rank, path in enumerate(keyword_paths, start=1): scores[path] += text_weight / (_RRF_K + rank) return scores @R.register("node_search_step") class NodeSearchStep(BaseStep): """Node-level digest-only hybrid search for dream Phase 2 recall.""" async def execute(self): assert self.context is not None query: str = (self.context.get("query", "") or "").strip() limit: int = int(self.context.get("limit") or 20) vector_weight: float = float(self.kwargs.get("vector_weight", 0.7)) candidate_multiplier: float = float(self.kwargs.get("candidate_multiplier", 5.0)) if not query: self.context.response.success = False self.context.response.answer = "Error: query cannot be empty" return self.context.response assert limit > 0, f"limit must be positive, got {limit}" digest_dir = self.config_value("digest_dir") digest_prefix = digest_dir.rstrip("/") + "/" # Over-fetch — digest filter drops a lot of raw hits. candidates = min(_MAX_CANDIDATES, max(1, int(limit * candidate_multiplier))) vector_chunks, keyword_chunks = await asyncio.gather( self.file_store.vector_search(query, candidates, {}), self.file_store.keyword_search(query, candidates, {}), ) def _node_dedup(chunks: list) -> list[str]: """Keep first occurrence per path; respect digest prefix only. Self-exclusion (e.g. UPDATE target) is the LLM's job: frontmatter inlining lets the agent recognize self from the candidate list. No need to push it into a mechanical parameter. """ seen: set[str] = set() out: list[str] = [] for c in chunks: if c.path in seen: continue if not c.path.startswith(digest_prefix): continue seen.add(c.path) out.append(c.path) return out vector_paths = _node_dedup(vector_chunks) keyword_paths = _node_dedup(keyword_chunks) path_to_score = _rrf_merge_nodes(vector_paths, keyword_paths, vector_weight) ranked = sorted(path_to_score.items(), key=lambda kv: -kv[1])[:limit] # Attach frontmatter from in-memory FileNode metadata (no extra IO). node_paths = [p for p, _ in ranked] nodes = await self.file_store.get_nodes(node_paths) if node_paths else [] path_to_fm: dict[str, dict[str, str]] = {} for n in nodes: fm = n.front_matter path_to_fm[n.path] = { "name": (getattr(fm, "name", "") or "").strip(), "description": (getattr(fm, "description", "") or "").strip(), } hits: list[dict] = [] for path, score in ranked: fm = path_to_fm.get(path, {}) hits.append( { "path": path, "score": round(float(score), 4), "name": fm.get("name", ""), "description": fm.get("description", ""), }, ) self.logger.info( f"[{self.name}] query={query!r} candidates={candidates} " f"vector_hits={len(vector_chunks)} keyword_hits={len(keyword_chunks)} " f"returned={len(hits)}", ) lines = [ f"=== node_search query={query!r} hits={len(hits)}/{candidates} ===", ] if not hits: lines.append("(no digest hits)") else: for h in hits: lines.append( f"[{h['score']:.4f}] {h['path']}\n" f" name: {h['name']}\n" f" description: {h['description']}", ) self.context.response.success = True self.context.response.answer = "\n".join(lines) self.context.response.metadata["hits"] = hits self.context.response.metadata["counts"] = { "vector_raw": len(vector_chunks), "keyword_raw": len(keyword_chunks), "returned": len(hits), } return self.context.response