91 lines
2.7 KiB
Python
91 lines
2.7 KiB
Python
|
"""
|
||
|
|
Web Search Tool
|
||
|
|
|
||
|
|
Web search tool with result aggregation.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
from typing import Dict, Any, List, Optional
|
||
|
|
|
||
|
|
|
||
|
|
class WebSearch:
|
||
|
|
"""Web search tool"""
|
||
|
|
|
||
|
|
def __init__(self, config: dict):
|
||
|
|
self.api_key = config.get("api_key")
|
||
|
|
self.max_results = config.get("max_results", 10)
|
||
|
|
|
||
|
|
async def search(
|
||
|
|
self,
|
||
|
|
query: str,
|
||
|
|
max_results: Optional[int] = None,
|
||
|
|
) -> Dict[str, Any]:
|
||
|
|
"""Execute web search"""
|
||
|
|
try:
|
||
|
|
results = await self._do_search(
|
||
|
|
query,
|
||
|
|
max_results or self.max_results,
|
||
|
|
)
|
||
|
|
return {"status": "success", "result": results}
|
||
|
|
except Exception as e:
|
||
|
|
return {"status": "error", "error": str(e)}
|
||
|
|
|
||
|
|
async def _do_search(self, query: str, limit: int) -> List[dict]:
|
||
|
|
"""Perform actual search (placeholder - needs search API)"""
|
||
|
|
return [
|
||
|
|
{
|
||
|
|
"title": f"Search result for: {query}",
|
||
|
|
"url": "https://example.com",
|
||
|
|
"snippet": "This is a placeholder search result. Configure API key for real results.",
|
||
|
|
}
|
||
|
|
]
|
||
|
|
|
||
|
|
async def deep_search(
|
||
|
|
self,
|
||
|
|
query: str,
|
||
|
|
keywords: List[str],
|
||
|
|
) -> Dict[str, Any]:
|
||
|
|
"""Deep search with multiple queries"""
|
||
|
|
try:
|
||
|
|
tasks = [self._do_search(kw, 5) for kw in [query] + keywords]
|
||
|
|
results = await asyncio.gather(*tasks)
|
||
|
|
|
||
|
|
aggregated = self._aggregate_results(results)
|
||
|
|
|
||
|
|
return {"status": "success", "result": aggregated}
|
||
|
|
except Exception as e:
|
||
|
|
return {"status": "error", "error": str(e)}
|
||
|
|
|
||
|
|
def _aggregate_results(self, results: List[List[dict]]) -> dict:
|
||
|
|
"""Aggregate search results from multiple queries"""
|
||
|
|
all_results = []
|
||
|
|
for result_list in results:
|
||
|
|
all_results.extend(result_list)
|
||
|
|
|
||
|
|
unique_results = []
|
||
|
|
seen_urls = set()
|
||
|
|
for r in all_results:
|
||
|
|
if r.get("url") not in seen_urls:
|
||
|
|
seen_urls.add(r.get("url"))
|
||
|
|
unique_results.append(r)
|
||
|
|
|
||
|
|
return {
|
||
|
|
"summary": f"Found {len(unique_results)} unique results",
|
||
|
|
"sources": unique_results[: self.max_results],
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def create_web_search_executor(config: dict):
|
||
|
|
"""Create web search executor"""
|
||
|
|
search = WebSearch(config)
|
||
|
|
|
||
|
|
async def execute(command: str, parameters: dict) -> dict:
|
||
|
|
if command == "search":
|
||
|
|
return await search.search(**parameters)
|
||
|
|
elif command == "deep_search":
|
||
|
|
return await search.deep_search(**parameters)
|
||
|
|
else:
|
||
|
|
return {"status": "error", "error": f"Unknown command: {command}"}
|
||
|
|
|
||
|
|
return execute
|