brain-mcp

Instructions

    Search conversation summaries with hybrid vector + keyword search.

    Args:
        query: Search query
        extract: What to extract from results:
            - "summary" (default): Full summary with metadata
            - "questions": Open questions from matching conversations
            - "decisions": Decisions made in matching conversations
            - "quotes": Quotable phrases from matching conversations
        limit: Max results (default 10)
        domain: Filter by domain (e.g. "ai-dev", "business-strategy")
        importance: Filter by importance ("breakthrough", "significant", "routine")
        thinking_stage: Filter by stage ("exploring", "crystallizing", "refining", "executing")
        source: Filter by source ("claude-code", "chatgpt", etc.)
        mode: Search mode — "hybrid" (default), "vector", "fts"
    

Implementation Reference

• brain_mcp/server/tools_search.py:86-216 (handler)

  The 'search_summaries' tool definition and implementation, which performs hybrid search (vector + keyword) over conversation summaries. It handles different extraction modes (questions, decisions, quotes) and filtering by domain, importance, thinking_stage, or source.

  def search_summaries(
      query: str,
      extract: str = "summary",
      limit: int = 10,
      domain: str = None,
      importance: str = None,
      thinking_stage: str = None,
      source: str = None,
      mode: str = "hybrid",
  ) -> str:
      """
      Search conversation summaries with hybrid vector + keyword search.
  
      Args:
          query: Search query
          extract: What to extract from results:
              - "summary" (default): Full summary with metadata
              - "questions": Open questions from matching conversations
              - "decisions": Decisions made in matching conversations
              - "quotes": Quotable phrases from matching conversations
          limit: Max results (default 10)
          domain: Filter by domain (e.g. "ai-dev", "business-strategy")
          importance: Filter by importance ("breakthrough", "significant", "routine")
          thinking_stage: Filter by stage ("exploring", "crystallizing", "refining", "executing")
          source: Filter by source ("claude-code", "chatgpt", etc.)
          mode: Search mode — "hybrid" (default), "vector", "fts"
      """
      if extract == "questions":
          return _extract_open_questions(query, domain=domain, importance=importance,
                                         thinking_stage=thinking_stage, source=source,
                                         limit=limit, mode=mode)
      elif extract == "decisions":
          return _extract_decisions(query, domain=domain, importance=importance,
                                    thinking_stage=thinking_stage, source=source,
                                    limit=limit, mode=mode)
      elif extract == "quotes":
          return _extract_quotes(query, domain=domain, importance=importance,
                                 thinking_stage=thinking_stage, source=source,
                                 limit=limit, mode=mode)
  
      lance = get_summaries_lance()
      if not lance:
          return (
              "Summary vector database not found. "
              "To use summary search, run the summarize pipeline:\n"
              "  python -m summarize.summarize\n\n"
              "For basic search, use semantic_search or search_conversations instead."
          )
  
      try:
          tbl = lance.open_table(SUMMARIES_TABLE)
      except Exception as e:
          return f"Summary table not found ({e})."
  
      # Build SQL filter (sanitize user inputs for LanceDB where clauses)
      filters = []
      if domain:
          filters.append(f"domain_primary = '{sanitize_sql_value(domain)}'")
      if importance:
          filters.append(f"importance = '{sanitize_sql_value(importance)}'")
      if thinking_stage:
          filters.append(f"thinking_stage = '{sanitize_sql_value(thinking_stage)}'")
      if source:
          filters.append(f"source = '{sanitize_sql_value(source)}'")
      where_clause = " AND ".join(filters) if filters else None
  
      try:
          if mode == "hybrid":
              search = tbl.search(query, query_type="hybrid").limit(limit * 3)
              if where_clause:
                  search = search.where(where_clause)
              results = search.to_list()
          elif mode == "fts":
              search = tbl.search(query, query_type="fts").limit(limit)
              if where_clause:
                  search = search.where(where_clause)
              results = search.to_list()
          else:
              embedding = get_embedding(f"search_query: {query}")
              if not embedding:
                  return "Could not generate embedding."
              search = tbl.search(embedding).limit(limit)
              if where_clause:
                  search = search.where(where_clause)
              results = search.to_list()
      except Exception:
          # Fallback to vector-only
          embedding = get_embedding(f"search_query: {query}")
          if not embedding:
              return "Could not generate embedding."
          search = tbl.search(embedding).limit(limit)
          if where_clause:
              search = search.where(where_clause)
          results = search.to_list()
  
      if not results:
          return f"No summaries found for: {query}"
  
      # Note: Cross-encoder reranking removed in fastembed migration.
      # Results are returned in vector similarity order.
  
      results = results[:limit]
  
      output = [f"## Summary Search: '{query}'\n"]
      if where_clause:
          output.append(f"_Filters: {where_clause}_\n")
      output.append(f"_Found {len(results)} matching conversation summaries_\n")
  
      for i, r in enumerate(results):
          title = (r.get("title") or "Untitled")[:80]
          summary = (r.get("summary") or "")[:400]
          imp = r.get("importance", "?")
          domain_p = r.get("domain_primary", "?")
          stage = r.get("thinking_stage", "?")
          src = r.get("source", "?")
          concepts_raw = parse_json_field(r.get("concepts"))
          concepts_str = ", ".join(concepts_raw[:8]) if concepts_raw else ""
          rerank = r.get("rerank_score")
          conv_id = r.get("conversation_id", "?")
  
          imp_icon = {"breakthrough": "🔥", "significant": "⭐", "routine": "📝"}.get(imp, "📝")
          output.append(f"### {i+1}. {imp_icon} {title}")
          output.append(f"**Domain**: {domain_p} | **Stage**: {stage} | **Source**: {src} | **Importance**: {imp}")
          if concepts_str:
              output.append(f"**Concepts**: {concepts_str}")
          if rerank is not None:
              output.append(f"**Relevance**: {rerank:.2f} (reranked)")
          output.append(f"> {summary}")
          output.append(f"_Conversation ID: {conv_id}_\n")
  
      return "\n".join(output)