rag_search
Search the web for queries and retrieve relevant content using local RAG-based similarity ranking to provide context for LLMs.
Instructions
Search the web for a given query. Give back context to the LLM with a RAG-like similarity sort.
Args: query (str): The query to search for. num_results (int): Number of results to return. top_k (int): Use top "k" results for content.
Returns: Dict of strings containing best search based on input query. Formatted in markdown.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | ||
| num_results | No | ||
| top_k | No |
Implementation Reference
- src/mcp_local_rag/main.py:7-37 (handler)The core handler function for the 'rag_search' MCP tool, which includes the @mcp.tool() decorator for registration. It performs a web search using DuckDuckGo (DDGS), computes similarity scores using embeddings, sorts and selects top results, fetches their content concurrently, and returns a dictionary with markdown-formatted content.@mcp.tool() def rag_search(query: str, num_results:int=10, top_k:int=5) -> Dict: """ Search the web for a given query. Give back context to the LLM with a RAG-like similarity sort. Args: query (str): The query to search for. num_results (int): Number of results to return. top_k (int): Use top "k" results for content. Returns: Dict of strings containing best search based on input query. Formatted in markdown. """ # Import heavy dependencies only when tool is invoked from ddgs import DDGS from .utils.fetch import fetch_all_content from .utils.tools import sort_by_score ddgs = DDGS() results = ddgs.text(query, max_results=num_results) scored_results = sort_by_score(add_score_to_dict(query, results)) top_results = scored_results[0:top_k] # fetch content using thread pool md_content = fetch_all_content(top_results) # formatted as dict return { "content": md_content }
- src/mcp_local_rag/main.py:39-55 (helper)Helper function used by rag_search to add cosine similarity scores (using MediaPipe TextEmbedder) to each search result based on the query embedding.def add_score_to_dict(query: str, results: List[Dict]) -> List[Dict]: """Add similarity scores to search results.""" # Import heavy dependencies only when needed (slow import!) from importlib.resources import files from mediapipe.tasks.python import text from .utils.fetch import fetch_embedder, get_path_str path = get_path_str(files('mcp_local_rag.embedder').joinpath('embedder.tflite')) embedder = fetch_embedder(path) query_embedding = embedder.embed(query) for i in results: i['score'] = text.TextEmbedder.cosine_similarity( embedder.embed(i['body']).embeddings[0], query_embedding.embeddings[0]) return results
- Utility function to sort search results by descending similarity score, used in rag_search.def sort_by_score(results: List[Dict]) -> List[Dict]: """Sort results by similarity score.""" return sorted(results, key=lambda x: x['score'], reverse=True)
- Helper function to fetch textual content from top search result URLs concurrently using ThreadPoolExecutor and BeautifulSoup parsing, returns list of content dicts used by rag_search.def fetch_all_content(results: List[Dict]) -> List[str]: """Fetch content from all URLs using a thread pool.""" urls = [site['href'] for site in results if site.get('href')] # parallelize requests with ThreadPoolExecutor(max_workers=5) as executor: # submit fetch tasks to executor future_to_url = {executor.submit(fetch_content, url): url for url in urls} content_list = [] for future in future_to_url: try: content = future.result() if content: content_list.append({ "type": "text", "text": content }) except Exception as e: print(f"Request failed with exception: {e}") return content_list
- Helper function to load and create MediaPipe TextEmbedder from TFLite model path, used in scoring by rag_search.def fetch_embedder(path:str, l2_normalize:bool=True, quantize:bool=False):# ->text.text_embedder.TextEmbedder: base_options = python.BaseOptions(model_asset_path=path) options = text.TextEmbedderOptions( base_options=base_options, l2_normalize=l2_normalize, quantize=quantize) embedder = text.TextEmbedder.create_from_options(options) return embedder