search

Instructions

Implementation Reference

• src/semantic_cache_mcp/cache/search.py:77-157 (handler)

  Core search logic: embeds the query, performs hybrid BM25+vector search via VectorStorage.search_hybrid(), normalizes scores, filters by directory, and returns SearchResult.

  async def semantic_search(
      cache: SemanticCache,
      query: str,
      k: int = 10,
      directory: str | None = None,
  ) -> SearchResult:
      """Search cached files by semantic meaning using hybrid BM25+vector search."""
      # DoS protection
      k = max(1, min(k, MAX_SEARCH_K))
      query = query[:MAX_SEARCH_QUERY_LEN]
  
      # Embed query for vector component of hybrid search.
      # MUST go through executor — ONNX is not thread-safe.
      loop = asyncio.get_running_loop()
      started = time.perf_counter()
      log_marker(logger, "embed.query.begin", chars=len(query))
      query_embedding = await loop.run_in_executor(cache._io_executor, embed_query, query)
      log_marker(
          logger,
          "embed.query.end",
          ok=query_embedding is not None,
          elapsed_ms=round((time.perf_counter() - started) * 1000, 1),
      )
  
      # Resolve directory for post-search filtering (is_relative_to is secure
      # against prefix attacks like /project vs /project_evil)
      resolved_dir: Path | None = None
      if directory:
          resolved_dir = Path(directory).expanduser().resolve()
  
      # Use hybrid search (BM25 + vector) via VectorStorage
      # Request extra results when directory filtering will reduce the set
      storage = cache._storage
      search_k = k * 3 if resolved_dir else k
      results = await storage.search_hybrid(
          query=query,
          embedding=query_embedding,
          k=search_k,
      )
  
      if not results:
          stats = await storage.get_stats()
          total = stats.get("files_cached", 0)
          return SearchResult(query=query, matches=[], files_searched=0, cached_files=int(total))
  
      # Build matches with directory filtering
      filtered: list[tuple[str, str, float]] = []
      for path, preview, score in results:
          if len(filtered) >= k:
              break
          # Secure directory filter: is_relative_to prevents prefix attacks
          if resolved_dir and not Path(path).is_relative_to(resolved_dir):
              continue
          filtered.append((path, preview, score))
  
      # Normalize scores to 0–1 range (best result = 1.0) so LLMs can
      # judge relevance without knowing RRF score internals.
      max_score = filtered[0][2] if filtered else 1.0
      matches: list[SearchMatch] = []
      for path, preview, score in filtered:
          entry = await cache.get(path)
          tokens = entry.tokens if entry else 0
          normalized = round(score / max_score, 4) if max_score > 0 else 0.0
          matches.append(
              SearchMatch(
                  path=path,
                  similarity=normalized,
                  tokens=tokens,
                  preview=preview.replace("\n", " "),
              )
          )
  
      stats = await storage.get_stats()
      total = stats.get("files_cached", 0)
  
      return SearchResult(
          query=query,
          matches=matches,
          files_searched=int(total),
          cached_files=int(total),
      )

• src/semantic_cache_mcp/server/_tool_models.py:186-202 (schema)

  Pydantic response models: SearchMatch (single file result with path/similarity/tokens/preview) and SearchResponse (full output with query, matches, count, metadata).

  class SearchMatch(ToolResponseModel):
      path: str | None = None
      similarity: float | None = None
      tokens: int | None = None
      preview: str | None = None
  
  
  class SearchResponse(ToolResponseModel):
      query: str | None = None
      matches: list[SearchMatch] | None = None
      count: int | None = None
      cached_files: int | None = None
      files_searched: int | None = None
      k: int | None = None
      directory: str | None = None
      show_preview: bool | None = None

• src/semantic_cache_mcp/types.py:78-95 (schema)

  Internal data types: SearchMatch (path, similarity, tokens, preview) and SearchResult (query, matches list, search stats).

  @dataclass(slots=True)
  class SearchMatch:
      """A single search result with similarity score."""
  
      path: str
      similarity: float  # 0.0-1.0
      tokens: int
      preview: str  # First 200 chars
  
  
  @dataclass(slots=True)
  class SearchResult:
      """Result from semantic_search operation."""
  
      query: str
      matches: list[SearchMatch]
      files_searched: int
      cached_files: int

• src/semantic_cache_mcp/storage/vector/__init__.py:766-792 (helper)

  VectorStorage.search_hybrid: delegates to simplevecdb's hybrid_search (BM25 + vector RRF fusion), falls back to keyword-only if no embedding, deduplicates results.

  async def search_hybrid(
      self,
      query: str,
      embedding: EmbeddingVector | None = None,
      k: int = 5,
      filter: dict | None = None,
  ) -> list[tuple[str, str, float]]:
      """Hybrid BM25 + vector search with RRF fusion.
  
      Falls back to keyword-only if no embedding provided.
      """
      if self._closed:
          return []
      query_vector = list(embedding) if embedding is not None else None
  
      try:
          results = await self._collection.hybrid_search(
              query,
              k=k * 2,
              filter=filter,
              query_vector=query_vector,
          )
      except Exception as e:
          logger.warning(f"Hybrid search failed: {e}")
          return []
  
      return self._dedupe_search_results(results, k)