"""
Search functionality for the ModularOllamaAgent.
This module provides search-related functionality for the ModularOllamaAgent,
including natural language detection of search intent, search execution, and
search results processing.
"""
import time
import logging
from typing import Dict, Any, List, Optional, Set
from ollamatools.plugins.tools.web import search_web
from ollamatools.core.errors import WebError
from ollamatools.plugins.integrations.opensearch.tools import opensearch_query
from .utils import print_status, print_success, print_error, print_warning, print_info
# Set up logger
logger = logging.getLogger(__name__)
def detect_factual_intent(message: str) -> bool:
"""
Detect if a message is likely asking for factual information
that would benefit from search.
Args:
message: User message
Returns:
bool: True if message likely needs external information
"""
# Basic topics that often require external information
external_topics = [
"recipe", "weather", "news", "sports", "movie", "book", "author",
"product", "review", "price", "stock", "company", "history",
"person", "celebrity", "politician", "event", "scientific",
"research", "study", "statistic", "data", "definition"
]
# Check if message contains any of these topics
contains_topic = any(topic in message.lower() for topic in external_topics)
# Check if it's a complex question (longer questions tend to be more specific)
is_complex = len(message.split()) > 6
# Simple heuristic: if it contains a topic of interest and is reasonably complex
return contains_topic and is_complex
def extract_search_context(search_results: Dict[str, Any]) -> str:
"""
Extract useful context from search results.
Args:
search_results: Results from opensearch_query
Returns:
str: Formatted context for the agent
"""
context_parts = []
# Add direct answer if available
if search_results.get("answer"):
try:
# Handle both string and dict answers
answer = search_results["answer"]
if isinstance(answer, dict):
answer_text = answer.get("content", str(answer))
else:
answer_text = str(answer)
context_parts.append(f"Direct answer: {answer_text}")
except Exception as e:
logger.warning(f"Error extracting answer for context: {str(e)}")
# Add snippets from results
if search_results.get("results"):
context_parts.append("Information from search results:")
for i, result in enumerate(search_results["results"][:3]): # Limit to 3 results
title = result.get("title", "Untitled")
content = result.get("content", "No content")
url = result.get("url", "")
context_parts.append(f"- {title}: {content} (Source: {url})")
# Join everything with newlines
return "\n".join(context_parts)
def execute_search(
query: str,
opensearch_config: Dict[str, Any],
permission_manager: Any,
execution_history: List[Dict[str, Any]]
) -> Dict[str, Any]:
"""
Execute a search using OpenSearchNG or fall back to standard search.
Args:
query: Search query
opensearch_config: OpenSearchNG configuration
permission_manager: Permission manager instance
execution_history: History of executed functions
Returns:
dict: Search results
"""
try:
print_status(f"Searching for: {query}")
# Try to use OpenSearchNG first
try:
# Execute the search with OpenSearchNG
results = opensearch_query(
query=query,
config=opensearch_config,
permission_manager=permission_manager
)
# Show the number of results
num_results = len(results.get("results", []))
print_success(f"Found {num_results} results using OpenSearchNG")
# Show suggested answer if available
if results.get("answer"):
try:
# Handle both string and dict answers
answer = results["answer"]
if isinstance(answer, dict):
answer_text = answer.get("content", str(answer))
else:
answer_text = str(answer)
print_info(f"Direct answer: {answer_text}")
except Exception as e:
logger.warning(f"Error displaying answer: {str(e)}")
# Register in execution history
execution_history.append({
"function": "opensearch_query",
"args": {"query": query},
"timestamp": time.time(),
"success": True,
"num_results": num_results
})
return results
except (WebError, ImportError, TypeError) as e:
# If OpenSearchNG fails, fall back to standard web search
if "Could not connect to OpenSearchNG server" in str(e) or isinstance(e, ImportError):
print_warning("OpenSearchNG is not available. Using fallback search method.")
# Use the fallback search method from web_tools
search_results = search_web(
query,
num_results=5,
permission_manager=permission_manager
)
# Convert results to the expected format
fallback_results = {
"query": query,
"results": [],
"suggestion": None,
"answer": None,
"engines": ["fallback"],
"timing": {"total": 0}
}
for result in search_results:
fallback_results["results"].append({
"title": result.get("title", ""),
"url": result.get("url", ""),
"content": result.get("snippet", ""),
"engine": "fallback"
})
num_results = len(fallback_results["results"])
print_success(f"Found {num_results} results using fallback search")
# Register in execution history
execution_history.append({
"function": "search_web_fallback",
"args": {"query": query},
"timestamp": time.time(),
"success": True,
"num_results": num_results
})
return fallback_results
else:
# Re-raise other WebError exceptions
raise
except Exception as e:
print_error(f"Search failed: {str(e)}")
# Register in execution history
execution_history.append({
"function": "execute_search",
"args": {"query": query},
"timestamp": time.time(),
"success": False,
"error": str(e)
})
# Return empty results
return {
"query": query,
"results": [],
"error": str(e)
}