Spaces:
Running
Running
| import os | |
| from typing import List, Dict | |
| import requests | |
| from config.config import Config | |
| class SearchTool: | |
| """Tavily web search wrapper with content extraction.""" | |
| def __init__(self) -> None: | |
| self.api_key = os.getenv("TAVILY_API_KEY") or Config.TAVILY_API_KEY | |
| if not self.api_key: | |
| print("β οΈ WARNING: TAVILY_API_KEY not found!") | |
| # Don't raise error, allow graceful degradation | |
| self.api_key = None | |
| def search(self, query: str, num_results: int = 5) -> List[Dict]: | |
| """ | |
| Search using Tavily API. | |
| Returns results with title, url, content (snippet from Tavily). | |
| """ | |
| if not self.api_key: | |
| print("β Tavily search skipped - no API key") | |
| return [] | |
| url = "https://api.tavily.com/search" | |
| payload = { | |
| "api_key": self.api_key, | |
| "query": query, | |
| "max_results": num_results, | |
| "include_answer": True, # Get Tavily's AI answer | |
| "include_raw_content": False, | |
| "search_depth": "advanced" # Better results | |
| } | |
| print(f"π Tavily searching for: {query[:50]}...") | |
| try: | |
| resp = requests.post(url, json=payload, timeout=30) | |
| resp.raise_for_status() | |
| data = resp.json() | |
| results = data.get("results", []) | |
| # Add Tavily's answer as metadata if available | |
| tavily_answer = data.get("answer", "") | |
| if tavily_answer and results: | |
| results[0]["tavily_answer"] = tavily_answer | |
| print(f" β Tavily AI answer: {tavily_answer[:100]}...") | |
| print(f" β Tavily returned {len(results)} results") | |
| if results: | |
| print(f" π First result: {results[0].get('title', 'N/A')}") | |
| return results | |
| except requests.exceptions.RequestException as e: | |
| print(f"β Tavily search error: {e}") | |
| return [] | |
| except ValueError as e: | |
| print(f"β Tavily JSON error: {e}") | |
| return [] | |