"""Semantic search across message history using vector embeddings."""
from __future__ import annotations
import json
import logging
from typing import Any, TYPE_CHECKING
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NAME = "semantic_search"
TOOL_DESCRIPTION = (
"Perform a semantic search across message history using vector embeddings. "
"Returns messages that are semantically similar to the query."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query text",
},
"top_k": {
"type": "integer",
"description": "Number of results (default 10, max 50)",
},
"channel_id": {
"type": "string",
"description": "Optional channel ID to filter results",
},
},
"required": ["query"],
}
[docs]
async def run(query: str, top_k: int = 10, channel_id: str = "", *, ctx: ToolContext | None = None) -> str:
"""Execute this tool and return the result.
Args:
query (str): Search query or input string.
top_k (int): The top k value.
channel_id (str): Discord/Matrix channel identifier.
Returns:
str: Result string.
"""
top_k = max(1, min(top_k, 50))
if ctx is None or ctx.message_cache is None:
return json.dumps({"error": "Message cache not available for vector search"})
try:
mc = ctx.message_cache
results = await mc.search_messages(
query=query,
limit=top_k,
channel_id=channel_id or None,
)
return json.dumps({
"query": query,
"top_k": top_k,
"channel_id": channel_id or None,
"num_results": len(results),
"results": results,
}, default=str)
except AttributeError:
return json.dumps({
"query": query,
"num_results": 0,
"results": [],
"error": "Message cache does not support vector search",
})
except Exception as e:
logger.error("Semantic search failed: %s", e, exc_info=True)
return json.dumps({
"query": query,
"num_results": 0,
"results": [],
"error": str(e),
})