In the modern world of AI agents, one of the main challenges is access to up-to-date information. Most language models are trained on data with a specific cutoff date and cannot provide information about recent events, current prices, or news.
In this article, we will explore how to solve this problem using a combination of any-agent (a universal library for working with agents from Mozilla.ai) and Tavily (a specialized search engine for AI).
In This Article
- Part 1: Getting Started with any-agent
- Part 2: Getting Started with Tavily
- Part 3: Integrating any-agent with Tavily
- Part 4: Full-featured Example
- Part 5: Advanced Features
- Part 6: Best Practices and Recommendations
- Part 7: Practical Use Cases
- Conclusion
- Appendix A: Full Code for a Production Agent
- Appendix B: Configuration Files
- Appendix B: Testing
- Final Recommendations
- Concluding Remarks
Part 1: Getting Started with any-agent
What is any-agent?
any-agent is a Python library from Mozilla.ai that provides a unified interface for working with various agent frameworks. The main idea is to allow developers to easily switch between different agent frameworks (AutoGen, CrewAI, LangGraph, etc.) without rewriting code.
Key advantages:
- Unified API: one interface for all frameworks
- Standardized logging: consistent traces regardless of the framework
- Easy comparison: ability to test different agent configurations
- Extensibility: support for adding new frameworks
System requirements and installation
<em># System requirements</em>
<em># Python 3.11+</em>
<em># Basic installation (TinyAgent only)</em>
pip install any-agent
<em># Installation with support for all frameworks</em>
pip install any-agent[all]
<em># Or with a specific framework</em>
pip install any-agent[langchain]
First agent example
from any_agent import AnyAgent
from any_agent.frameworks import TinyAgent
<em># Creating a simple agent</em>
agent = AnyAgent(
framework=TinyAgent,
model="gpt-4",
tools=[],
system_prompt="Вы полезный ассистент"
)
<em># Running the agent</em>
result = agent.run("Привет! Расскажи о себе.")
print(result)
Important for Jupyter Notebook: If running in Jupyter Notebook you will need to add the following two lines before running AnyAgent, otherwise you may see the error RuntimeError: This event loop is already running. This is a known limitation of Jupyter Notebooks, see Github Issue
import nest_asyncio
nest_asyncio.apply()
<em># Now you can use any-agent</em>
Part 2: Getting Started with Tavily
What is Tavily?
Tavily is a search engine created specifically for AI agents and RAG systems. Unlike traditional search engines, Tavily is optimized to provide structured, factual results that are easily processed by language models.
Key features of Tavily:
- AI Optimization: results adapted for LLMs
- Focus on facts: emphasis on accuracy and verifiability
- High speed: fast real-time responses
- Structured output: JSON format with metadata
Registration and obtaining an API key
- Go to https://app.tavily.com/
- Create an account
- Get your API key in your personal account
Basic Tavily usage
<em># Client installation</em>
pip install tavily-python
from tavily import TavilyClient
<em># Client initialization</em>
client = TavilyClient(api_key="tvly-YOUR_API_KEY")
<em># Simple search</em>
response = client.search("Последние новости о Python 3.12")
print(f"Запрос: {response['query']}")
print(f"Ответ: {response['answer']}")
print(f"Источники: {len(response['results'])} результатов")
Tavily response structure
{
"query": "Ваш поисковый запрос",
"answer": "Краткий ответ на основе найденной информации",
"images": [],
"results": [
{
"title": "Заголовок источника",
"url": "https://example.com",
"content": "Содержимое...",
"score": 0.81025416,
"favicon": "https://example.com/favicon.png"
}
],
"auto_parameters": {
"topic": "general",
"search_depth": "basic"
},
"response_time": "1.67"
}
Part 3: Integrating any-agent with Tavily
Approach 1: Creating a Custom Tool
The most flexible way is to create your own search tool:
import os
from any_agent import AnyAgent
from any_agent.frameworks import TinyAgent
from tavily import TavilyClient
from typing import Dict, Optional
class TavilySearchTool:
"""Custom tool for searching via Tavily"""
def __init__(self, api_key: str):
self.client = TavilyClient(api_key=api_key)
self.name = "tavily_search"
self.description = "Search for up-to-date information on the internet"
def __call__(self, query: str, search_depth: str = "basic") -> Dict:
"""Perform search"""
try:
response = self.client.search(
query=query,
search_depth=search_depth
)
return {
"success": True,
"data": response
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
<em># Creating an agent with Tavily</em>
tavily_tool = TavilySearchTool(api_key="tvly-YOUR_API_KEY")
agent = AnyAgent(
framework=TinyAgent,
model="gpt-4",
tools=[tavily_tool],
system_prompt="""
You are an AI assistant with access to up-to-date information from the internet.
Use the tavily_search tool to get fresh data.
Always cite sources of information in your answers.
"""
)
<em># Usage</em>
result = agent.run("Find the latest news on AI development from OpenAI")
print(result)
Approach 2: Integration via LangChain
If you are using a LangChain-compatible framework:
from langchain_tavily import TavilySearchAPIWrapper
from any_agent import AnyAgent
<em># Setting up search via LangChain</em>
search_tool = TavilySearchAPIWrapper(
tavily_api_key="tvly-YOUR_API_KEY"
)
<em># Creating an agent</em>
agent = AnyAgent(
framework=YourLangChainFramework,
model="gpt-4",
tools=[search_tool],
system_prompt="Assistant with real-time search access"
)
Part 4: Full-featured Example
Let’s create a complete application for working with search agents:
<em>## \file src/agents/smart_search_agent.py</em>
<em># -*- coding: utf-8 -*-</em>
<em>#! .pyenv/bin/python3</em>
"""
Smart Search Agent Module
==============================
Integration of any-agent with Tavily to create an agent
with access to up-to-date information from the internet.
"""
from typing import Dict, List, Optional
from dataclasses import dataclass
import json
from datetime import datetime
from tavily import TavilyClient
from any_agent import AnyAgent
from any_agent.frameworks import TinyAgent
@dataclass
class SearchResult:
"""Search result structure"""
query: str
answer: str
sources: List[Dict]
timestamp: str
response_time: float
class SmartSearchAgent:
"""
Smart Search Agent with Tavily Integration
The agent is capable of searching for up-to-date information,
analyzing results, and providing structured answers.
"""
def __init__(
self,
tavily_api_key: str,
model: str = "gpt-4",
search_depth: str = "basic"
):
"""
Agent initialization
Args:
tavily_api_key: API key for Tavily
model: Model for the agent
search_depth: Search depth ('basic' or 'advanced')
"""
self.tavily_client = TavilyClient(api_key=tavily_api_key)
self.search_depth = search_depth
self.agent = self._create_agent(model)
self.search_history: List[SearchResult] = []
def _create_agent(self, model: str) -> AnyAgent:
"""Create an agent with configured tools"""
return AnyAgent(
framework=TinyAgent,
model=model,
tools=[self._search_tool],
system_prompt=self._get_system_prompt()
)
def _get_system_prompt(self) -> str:
"""System prompt for the agent"""
return """
You are an expert AI assistant with access to up-to-date information from the internet.
Your capabilities:
- Search for up-to-date information via Tavily
- Analyze and synthesize data from multiple sources
- Provide factual answers with sources
Operating principles:
1. Always use search to get up-to-date information
2. Cite sources for all factual statements
3. If information is contradictory, point it out
4. Be accurate and objective
5. Structure answers for better comprehension
"""
def _search_tool(self, query: str) -> Dict:
"""
Search tool for the agent
Args:
query: Search query
Returns:
Search results in a structured format
"""
try:
start_time = datetime.now()
response = self.tavily_client.search(
query=query,
search_depth=self.search_depth
)
end_time = datetime.now()
response_time = (end_time - start_time).total_seconds()
<em># Saving the result to history</em>
search_result = SearchResult(
query=query,
answer=response.get('answer', ''),
sources=response.get('results', []),
timestamp=datetime.now().isoformat(),
response_time=response_time
)
self.search_history.append(search_result)
return {
"success": True,
"query": query,
"answer": response.get('answer', ''),
"sources": response.get('results', []),
"source_count": len(response.get('results', [])),
"response_time": response.get('response_time', response_time)
}
except Exception as e:
return {
"success": False,
"error": f"Search error: {str(e)}",
"query": query
}
def search(self, query: str) -> str:
"""
Execute a search query via the agent
Args:
query: User query
Returns:
Agent's response with up-to-date information
"""
try:
result = self.agent.run(query)
return result
except Exception as e:
return f"An error occurred while processing the request: {str(e)}"
def get_search_history(self) -> List[SearchResult]:
"""Get search query history"""
return self.search_history.copy()
def clear_history(self):
"""Clear search history"""
self.search_history.clear()
def export_history(self, filename: str):
"""
Export search history to a JSON file
Args:
filename: File name for export
"""
try:
history_data = [
{
"query": query,
"answer": answer,
"sources": sources,
"timestamp": timestamp,
"response_time": response_time
}
for query, answer, sources, timestamp, response_time in self.search_history
]
with open(filename, 'w', encoding='utf-8') as f:
json.dump(history_data, f, ensure_ascii=False, indent=2)
print(f"History exported to {filename}")
except Exception as e:
print(f"Export error: {str(e)}")
<em># Example usage</em>
def main():
"""Demonstration of the smart search agent"""
<em># Agent initialization</em>
agent = SmartSearchAgent(
tavily_api_key="tvly-YOUR_API_KEY",
model="gpt-4",
search_depth="basic"
)
<em># Example queries</em>
queries = [
"Latest news on GPT-5 development",
"Current prices for Bitcoin and Ethereum",
"What's new in Python 3.12?",
"Latest achievements in quantum computing"
]
print("=== Smart Search Agent Demonstration ===\n")
for i, query in enumerate(queries, 1):
print(f"Query {i}: {query}")
print("-" * 50)
result = agent.search(query)
print(result)
print("\n" + "="*80 + "\n")
<em># Export history</em>
agent.export_history("search_history.json")
<em># Statistics</em>
history = agent.get_search_history()
print(f"Total queries executed: {len(history)}")
if history:
avg_response_time = sum(r.response_time for r in history) / len(history)
print(f"Average response time: {avg_response_time:.2f} sec")
if __name__ == "__main__":
main()
Part 5: Advanced Features
Multi-Agent Systems
any-agent supports creating Multi-Agent systems where agents can interact with each other:
class ResearchTeam:
"""Team of research agents"""
def __init__(self, tavily_api_key: str):
<em># Researcher agent</em>
self.researcher = SmartSearchAgent(
tavily_api_key=tavily_api_key,
model="gpt-4"
)
<em># Analyst agent</em>
self.analyst = AnyAgent(
framework=TinyAgent,
model="gpt-4",
system_prompt="""
You are an expert analyst. Your task is to analyze
the information received and draw conclusions.
"""
)
<em># Writer agent</em>
self.writer = AnyAgent(
framework=TinyAgent,
model="gpt-4",
system_prompt="""
You are a professional writer. Create
structured reports based on analysis.
"""
)
def research_topic(self, topic: str) -> str:
"""Comprehensive topic research"""
<em># Stage 1: Information search</em>
research_data = self.researcher.search(f"Detailed information about {topic}")
<em># Stage 2: Data analysis</em>
analysis = self.analyst.run(f"Analyze this information: {research_data}")
<em># Stage 3: Report writing</em>
report = self.writer.run(f"Create a structured report based on the analysis: {analysis}")
return report
<em># Usage</em>
team = ResearchTeam("tvly-YOUR_API_KEY")
report = team.research_topic("The impact of AI on the labor market in 2024")
print(report)
Configuring Search Depth
Tavily offers different levels of search depth:
<em># Basic search</em>
basic_agent = SmartSearchAgent(
tavily_api_key="your-key",
search_depth="basic" <em># Faster, fewer sources</em>
)
<em># Advanced search</em>
advanced_agent = SmartSearchAgent(
tavily_api_key="your-key",
search_depth="advanced" <em># Slower, more sources</em>
)
Filtering Results
def create_specialized_search_tool(domains: List[str] = None):
"""Create a specialized search tool"""
def search_tool(query: str) -> Dict:
client = TavilyClient(api_key="your-key")
search_params = {
"query": query,
"search_depth": "advanced"
}
<em># Adding domain filters</em>
if domains:
search_params["include_domains"] = domains
response = client.search(**search_params)
return response
return search_tool
<em># Creating an agent for searching only scientific sources</em>
science_agent = AnyAgent(
framework=TinyAgent,
model="gpt-4",
tools=[create_specialized_search_tool([
"arxiv.org",
"nature.com",
"science.org"
])],
system_prompt="Expert in scientific information"
)
Part 6: Best Practices and Recommendations
Safety and Limitations
⚠️ Important warnings:
- Computational resources: Agent systems require significantly more resources than simple LLM calls
- Security: Agents can perform unpredictable actions
- Cost: Each search query and LLM call is charged
Performance Optimization
class OptimizedSearchAgent:
"""Optimized search agent"""
def __init__(self, tavily_api_key: str):
self.client = TavilyClient(api_key=tavily_api_key)
self.cache = {} <em># Cache for repetitive queries</em>
self.rate_limit_delay = 1 <em># Delay between requests</em>
def search_with_cache(self, query: str) -> Dict:
"""Search with result caching"""
<em># Check cache</em>
if query in self.cache:
print(f"Result taken from cache for: {query}")
return self.cache[query]
<em># Perform search</em>
time.sleep(self.rate_limit_delay) <em># Observe rate limits</em>
result = self.client.search(query)
<em># Save to cache</em>
self.cache[query] = result
return result
Error Handling
class RobustSearchAgent:
"""Error-resilient search agent"""
def __init__(self, tavily_api_key: str, max_retries: int = 3):
self.client = TavilyClient(api_key=tavily_api_key)
self.max_retries = max_retries
def search_with_retry(self, query: str) -> Dict:
"""Search with retries"""
for attempt in range(self.max_retries):
try:
return self.client.search(query)
except Exception as e:
print(f"Attempt {attempt + 1} failed: {str(e)}")
if attempt == self.max_retries - 1:
return {
"error": f"Search failed after {self.max_retries} attempts",
"query": query
}
time.sleep(2 ** attempt) <em># Exponential backoff</em>
Monitoring and Logging
import logging
from datetime import datetime
class MonitoredSearchAgent:
"""Agent with monitoring and logging"""
def __init__(self, tavily_api_key: str):
self.client = TavilyClient(api_key=tavily_api_key)
self.setup_logging()
self.stats = {
"total_searches": 0,
"successful_searches": 0,
"failed_searches": 0,
"total_response_time": 0
}
def setup_logging(self):
"""Configure logging system"""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('search_agent.log'),
logging.StreamHandler()
]
)
self.logger = logging.getLogger('SearchAgent')
def search_with_monitoring(self, query: str) -> Dict:
"""Search with full monitoring"""
start_time = datetime.now()
self.stats["total_searches"] += 1
try:
self.logger.info(f"Starting search: {query}")
result = self.client.search(query)
end_time = datetime.now()
response_time = (end_time - start_time).total_seconds()
self.stats["successful_searches"] += 1
self.stats["total_response_time"] += response_time
self.logger.info(f"Search successful in {response_time:.2f}s: {query}")
return result
except Exception as e:
self.stats["failed_searches"] += 1
self.logger.error(f"Search error {query}: {str(e)}")
return {"error": str(e), "query": query}
def get_statistics(self) -> Dict:
"""Get operational statistics"""
success_rate = (
self.stats["successful_searches"] / self.stats["total_searches"] * 100
if self.stats["total_searches"] > 0 else 0
)
avg_response_time = (
self.stats["total_response_time"] / self.stats["successful_searches"]
if self.stats["successful_searches"] > 0 else 0
)
return {
**self.stats,
"success_rate": f"{success_rate:.1f}%",
"average_response_time": f"{avg_response_time:.2f}s"
}
Part 7: Practical Use Cases
Case 1: News Aggregator
class NewsAggregatorAgent:
"""Agent for aggregating news by topics"""
def __init__(self, tavily_api_key: str):
self.search_agent = SmartSearchAgent(tavily_api_key)
def get_daily_digest(self, topics: List[str]) -> str:
"""Create a daily news digest"""
digest_parts = []
for topic in topics:
query = f"Latest news today: {topic}"
news = self.search_agent.search(query)
digest_parts.append(f"## {topic}\n{news}\n")
return "\n".join(digest_parts)
<em># Usage</em>
news_agent = NewsAggregatorAgent("your-api-key")
digest = news_agent.get_daily_digest([
"AI Technologies",
"Cryptocurrencies",
"Space Exploration"
])
Case 2: Research Assistant
class ResearchAssistant:
"""Assistant for scientific research"""
def __init__(self, tavily_api_key: str):
self.agent = SmartSearchAgent(tavily_api_key)
def literature_review(self, topic: str, num_sources: int = 10) -> Dict:
"""Literature review on a topic"""
queries = [
f"{topic} recent research 2024",
f"{topic} scientific papers",
f"{topic} review articles",
f"{topic} latest findings"
]
all_sources = []
for query in queries:
result = self.agent.search(query)
<em># Extract sources from result</em>
<em># ... processing logic</em>
return {
"topic": topic,
"sources_found": len(all_sources),
"summary": "...",
"key_findings": "...",
"sources": all_sources
}
Case 3: Business Analyst
class BusinessAnalyst:
"""Agent for business analysis"""
def __init__(self, tavily_api_key: str):
self.agent = SmartSearchAgent(tavily_api_key)
def market_analysis(self, company: str, competitors: List[str]) -> str:
"""Market and competitor analysis"""
<em># Analysis of the main company</em>
company_analysis = self.agent.search(
f"Financial results and news {company} 2024"
)
<em># Competitor analysis</em>
competitor_analyses = []
for competitor in competitors:
analysis = self.agent.search(
f"Financial results {competitor} comparison with {company}"
)
competitor_analyses.append(analysis)
<em># Report synthesis</em>
report_query = f"""
Create a business report based on the data:
Company: {company_analysis}
Competitors: {competitor_analyses}
"""
return self.agent.search(report_query)
Conclusion
The integration of any-agent with Tavily opens up broad possibilities for creating smart agents with access to up-to-date information. This combination is especially useful for:
Key advantages:
- Timeliness: access to fresh information in real-time
- Flexibility: a unified interface for different agent frameworks
- Reliability: specialized search optimized for AI
- Scalability: ability to create complex multi-agent systems
Usage recommendations:
- Start simple: master basic integration before moving to complex systems
- Monitor costs: keep track of API usage and computational resources
- Test thoroughly: agent systems can behave unpredictably
- Use caching: avoid repetitive queries for the same information
- Plan error handling: always foresee fallback scenarios
Next steps:
- Explore additional frameworks supported by any-agent
- Experiment with different Tavily search depth settings
- Consider integration with other tools (databases, external service APIs)
- Explore opportunities to create specialized agents for your domain
This technological combination represents a powerful tool for developing modern AI applications capable of working with current data and providing users with relevant information.
Appendix A: Full Code for a Production Agent
<em>## \file src/agents/production_search_agent.py</em>
<em># -*- coding: utf-8 -*-</em>
<em>#! .pyenv/bin/python3</em>
"""
Production Search Agent
===============================
A full-featured agent for production use with a complete
set of capabilities: monitoring, caching, error handling.
"""
import asyncio
import json
import time
import hashlib
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, List, Optional, Any, Union
from dataclasses import dataclass, asdict
from enum import Enum
import logging
from tavily import TavilyClient
from any_agent import AnyAgent
from any_agent.frameworks import TinyAgent
class SearchDepth(Enum):
"""Search depth"""
BASIC = "basic"
ADVANCED = "advanced"
class SearchStatus(Enum):
"""Search request status"""
SUCCESS = "success"
ERROR = "error"
CACHED = "cached"
RATE_LIMITED = "rate_limited"
@dataclass
class SearchRequest:
"""Search request structure"""
query: str
depth: SearchDepth = SearchDepth.BASIC
timestamp: str = None
user_id: Optional[str] = None
session_id: Optional[str] = None
def __post_init__(self):
if not self.timestamp:
self.timestamp = datetime.now().isoformat()
def get_cache_key(self) -> str:
"""Generate key for caching"""
content = f"{self.query}:{self.depth.value}"
return hashlib.md5(content.encode()).hexdigest()
@dataclass
class SearchResponse:
"""Search response structure"""
request: SearchRequest
status: SearchStatus
answer: str = ""
sources: List[Dict] = None
error_message: str = ""
response_time: float = 0.0
from_cache: bool = False
def __post_init__(self):
if self.sources is None:
self.sources = []
class CacheManager:
"""Cache manager for search queries"""
def __init__(self, cache_dir: str = "./cache", ttl_hours: int = 24):
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True)
self.ttl = timedelta(hours=ttl_hours)
def _get_cache_file(self, cache_key: str) -> Path:
"""Get path to cache file"""
return self.cache_dir / f"{cache_key}.json"
def get(self, cache_key: str) -> Optional[Dict]:
"""Get data from cache"""
cache_file = self._get_cache_file(cache_key)
if not cache_file.exists():
return None
try:
with open(cache_file, 'r', encoding='utf-8') as f:
data = json.load(f)
<em># Check TTL</em>
cached_time = datetime.fromisoformat(data['cached_at'])
if datetime.now() - cached_time > self.ttl:
cache_file.unlink() <em># Delete expired cache</em>
return None
return data['content']
except Exception:
return None
def set(self, cache_key: str, content: Dict):
"""Save data to cache"""
cache_file = self._get_cache_file(cache_key)
try:
cache_data = {
'cached_at': datetime.now().isoformat(),
'content': content
}
with open(cache_file, 'w', encoding='utf-8') as f:
json.dump(cache_data, f, ensure_ascii=False, indent=2)
except Exception as e:
logging.warning(f"Failed to save to cache: {e}")
def clear_expired(self):
"""Clear expired cache"""
for cache_file in self.cache_dir.glob("*.json"):
try:
with open(cache_file, 'r', encoding='utf-8') as f:
data = json.load(f)
cached_time = datetime.fromisoformat(data['cached_at'])
if datetime.now() - cached_time > self.ttl:
cache_file.unlink()
except Exception:
continue
class RateLimiter:
"""Request rate limiter"""
def __init__(self, max_requests: int = 60, time_window: int = 60):
self.max_requests = max_requests
self.time_window = time_window
self.requests = []
def can_make_request(self) -> bool:
"""Check if a request can be made"""
now = time.time()
<em># Clear expired requests</em>
self.requests = [req_time for req_time in self.requests
if now - req_time < self.time_window]
return len(self.requests) < self.max_requests
def register_request(self):
"""Register a completed request"""
self.requests.append(time.time())
def get_wait_time(self) -> float:
"""Get wait time until the next request"""
if not self.requests:
return 0.0
oldest_request = min(self.requests)
wait_time = self.time_window - (time.time() - oldest_request)
return max(0.0, wait_time)
class ProductionSearchAgent:
"""
Production Search Agent
A full-featured agent for production use with:
- Result caching
- Rate limiting
- Monitoring and logging
- Error handling
- Usage statistics
"""
def __init__(
self,
tavily_api_key: str,
model: str = "gpt-4",
cache_ttl_hours: int = 24,
rate_limit_per_minute: int = 60,
max_retries: int = 3,
log_level: str = "INFO"
):
<em># Component initialization</em>
self.tavily_client = TavilyClient(api_key=tavily_api_key)
self.cache_manager = CacheManager(ttl_hours=cache_ttl_hours)
self.rate_limiter = RateLimiter(max_requests=rate_limit_per_minute)
self.max_retries = max_retries
<em># Logging configuration</em>
self.setup_logging(log_level)
<em># Agent creation</em>
self.agent = self._create_agent(model)
<em># Statistics</em>
self.stats = {
"total_requests": 0,
"successful_requests": 0,
"cached_requests": 0,
"failed_requests": 0,
"rate_limited_requests": 0,
"total_response_time": 0.0,
"start_time": datetime.now().isoformat()
}
def setup_logging(self, log_level: str):
"""Configure logging system"""
logging.basicConfig(
level=getattr(logging, log_level.upper()),
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('production_search_agent.log'),
logging.StreamHandler()
]
)
self.logger = logging.getLogger('ProductionSearchAgent')
def _create_agent(self, model: str) -> AnyAgent:
"""Create an agent with production settings"""
return AnyAgent(
framework=TinyAgent,
model=model,
tools=[self._search_tool],
system_prompt=self._get_production_system_prompt()
)
def _get_production_system_prompt(self) -> str:
"""System prompt for production"""
return """
You are a professional enterprise-level AI assistant with access
to up-to-date information from the internet.
Your operating principles:
1. ACCURACY: Always verify facts through search
2. SOURCES: Cite sources for all statements
3. OBJECTIVITY: Present different viewpoints
4. STRUCTURE: Organize information logically
5. TIMELINESS: Use only fresh data
Response format:
- Brief summary (2-3 sentences)
- Key information with sources
- Additional sources for further study
If information is contradictory or unreliable - be sure to point it out.
"""
def _search_tool(self, query: str, depth: str = "basic") -> Dict:
"""Search tool with full error handling"""
search_request = SearchRequest(
query=query,
depth=SearchDepth(depth)
)
response = self._execute_search(search_request)
if response.status == SearchStatus.SUCCESS:
return {
"success": True,
"query": query,
"answer": response.answer,
"sources": response.sources,
"from_cache": response.from_cache,
"response_time": response.response_time
}
else:
return {
"success": False,
"query": query,
"error": response.error_message,
"status": response.status.value
}
def _execute_search(self, request: SearchRequest) -> SearchResponse:
"""Perform search with full handling"""
start_time = time.time()
self.stats["total_requests"] += 1
try:
<em># Check cache</em>
cache_key = request.get_cache_key()
cached_result = self.cache_manager.get(cache_key)
if cached_result:
self.stats["cached_requests"] += 1
self.logger.info(f"Result obtained from cache: {request.query}")
return SearchResponse(
request=request,
status=SearchStatus.CACHED,
answer=cached_result.get('answer', ''),
sources=cached_result.get('sources', []),
response_time=time.time() - start_time,
from_cache=True
)
<em># Check rate limit</em>
if not self.rate_limiter.can_make_request():
wait_time = self.rate_limiter.get_wait_time()
self.stats["rate_limited_requests"] += 1
self.logger.warning(f"Rate limit exceeded. Wait time: {wait_time:.1f}s")
return SearchResponse(
request=request,
status=SearchStatus.RATE_LIMITED,
error_message=f"Rate limit exceeded. Try again in {wait_time:.1f} seconds",
response_time=time.time() - start_time
)
<em># Perform search with retries</em>
result = None
last_error = None
for attempt in range(self.max_retries):
try:
self.rate_limiter.register_request()
result = self.tavily_client.search(
query=request.query,
search_depth=request.depth.value
)
break
except Exception as e:
last_error = e
if attempt < self.max_retries - 1:
wait_time = (2 ** attempt) * 1 <em># Exponential backoff</em>
self.logger.warning(f"Attempt {attempt + 1} failed: {e}. Retrying in {wait_time}s...")
time.sleep(wait_time)
if result is None:
self.stats["failed_requests"] += 1
self.logger.error(f"All search attempts failed for: {request.query}")
return SearchResponse(
request=request,
status=SearchStatus.ERROR,
error_message=f"Search failed after {self.max_retries} attempts: {last_error}",
response_time=time.time() - start_time
)
<em># Handle successful result</em>
response_time = time.time() - start_time
self.stats["successful_requests"] += 1
self.stats["total_response_time"] += response_time
<em># Save to cache</em>
cache_data = {
'answer': result.get('answer', ''),
'sources': result.get('results', [])
}
self.cache_manager.set(cache_key, cache_data)
self.logger.info(f"Search successful in {response_time:.2f}s: {request.query}")
return SearchResponse(
request=request,
status=SearchStatus.SUCCESS,
answer=result.get('answer', ''),
sources=result.get('results', []),
response_time=response_time,
from_cache=False
)
except Exception as e:
self.stats["failed_requests"] += 1
self.logger.error(f"Unexpected error during search: {e}", exc_info=True)
return SearchResponse(
request=request,
status=SearchStatus.ERROR,
error_message=f"Unexpected error: {str(e)}",
response_time=time.time() - start_time
)
def search(self, query: str, depth: SearchDepth = SearchDepth.BASIC) -> str:
"""
Main search method for users
Args:
query: Search query
depth: Search depth (basic/advanced)
Returns:
Agent's response with found information
"""
try:
self.logger.info(f"Processing user query: {query}")
<em># Create context for the agent</em>
search_context = f"Perform a search and provide a detailed answer to the query: {query}"
if depth == SearchDepth.ADVANCED:
search_context += " (use advanced search for deeper analysis)"
result = self.agent.run(search_context)
return result
except Exception as e:
self.logger.error(f"Error processing query: {e}", exc_info=True)
return f"Sorry, an error occurred while processing the request: {str(e)}"
def get_statistics(self) -> Dict[str, Any]:
"""Get detailed operational statistics"""
uptime = datetime.now() - datetime.fromisoformat(self.stats["start_time"])
success_rate = (
self.stats["successful_requests"] / self.stats["total_requests"] * 100
if self.stats["total_requests"] > 0 else 0
)
cache_hit_rate = (
self.stats["cached_requests"] / self.stats["total_requests"] * 100
if self.stats["total_requests"] > 0 else 0
)
avg_response_time = (
self.stats["total_response_time"] / self.stats["successful_requests"]
if self.stats["successful_requests"] > 0 else 0
)
return {
**self.stats,
"uptime": str(uptime),
"success_rate": f"{success_rate:.1f}%",
"cache_hit_rate": f"{cache_hit_rate:.1f}%",
"average_response_time": f"{avg_response_time:.3f}s",
"requests_per_hour": self.stats["total_requests"] / (uptime.total_seconds() / 3600) if uptime.total_seconds() > 0 else 0
}
def health_check(self) -> Dict[str, Any]:
"""System health check"""
try:
<em># Check basic functionality</em>
test_response = self._execute_search(SearchRequest("test query"))
return {
"status": "healthy",
"timestamp": datetime.now().isoformat(),
"tavily_api": "connected",
"cache": "operational",
"rate_limiter": "operational",
"last_test_status": test_response.status.value
}
except Exception as e:
return {
"status": "unhealthy",
"timestamp": datetime.now().isoformat(),
"error": str(e)
}
def cleanup(self):
"""Clean up resources upon shutdown"""
self.logger.info("Starting cleanup process...")
<em># Clear expired cache</em>
self.cache_manager.clear_expired()
<em># Save statistics</em>
stats_file = Path("agent_statistics.json")
try:
with open(stats_file, 'w', encoding='utf-8') as f:
json.dump(self.get_statistics(), f, ensure_ascii=False, indent=2)
self.logger.info(f"Statistics saved to {stats_file}")
except Exception as e:
self.logger.error(f"Failed to save statistics: {e}")
self.logger.info("Cleanup completed")
<em># Example usage in production</em>
async def main():
"""Demonstration of the production agent"""
<em># Agent initialization</em>
agent = ProductionSearchAgent(
tavily_api_key="tvly-YOUR_API_KEY",
model="mistral/mistral-small-latest",
cache_ttl_hours=12,
rate_limit_per_minute=30,
log_level="INFO"
)
try:
<em># Health check</em>
health = agent.health_check()
print(f"Agent health: {health['status']}")
<em># Test queries</em>
test_queries = [
"Latest achievements in quantum computing",
"Current trends in AI development 2024",
"Impact of climate change on the economy",
]
print("\n=== Production Agent Demonstration ===\n")
for i, query in enumerate(test_queries, 1):
print(f"Query {i}: {query}")
print("-" * 60)
result = agent.search(query, SearchDepth.BASIC)
print(result[:500] + "..." if len(result) > 500 else result)
print(f"\nStatistics:")
stats = agent.get_statistics()
print(f"- Total requests: {stats['total_requests']}")
print(f"- Successful: {stats['success_rate']}")
print(f"- From cache: {stats['cache_hit_rate']}")
print(f"- Average response time: {stats['average_response_time']}")
print("\n" + "="*80 + "\n")
<em># Final statistics</em>
print("Final statistics:")
final_stats = agent.get_statistics()
for key, value in final_stats.items():
print(f"{key}: {value}")
finally:
<em># Resource cleanup</em>
agent.cleanup()
if __name__ == "__main__":
asyncio.run(main())
Appendix B: Configuration Files
config/agent_config.json
{
"tavily": {
"api_key": "tvly-YOUR_API_KEY",
"default_search_depth": "basic",
"timeout_seconds": 30
},
"agent": {
"model": "gpt-4",
"framework": "TinyAgent",
"max_tokens": 4000,
"temperature": 0.1
},
"cache": {
"enabled": true,
"ttl_hours": 24,
"max_size_mb": 100,
"directory": "./cache"
},
"rate_limiting": {
"enabled": true,
"requests_per_minute": 60,
"burst_allowance": 10
},
"logging": {
"level": "INFO",
"file": "agent.log",
"max_file_size_mb": 10,
"backup_count": 5
},
"monitoring": {
"enabled": true,
"metrics_interval_seconds": 300,
"health_check_interval_seconds": 60
}
}
docker/Dockerfile
FROM python:3.11-slim
WORKDIR /app
<em># Install system dependencies</em>
RUN apt-get update && apt-get install -y \
gcc \
&& rm -rf /var/lib/apt/lists/*
<em># Copy dependency files</em>
COPY requirements.txt .
<em># Install Python dependencies</em>
RUN pip install --no-cache-dir -r requirements.txt
<em># Copy application code</em>
COPY . .
<em># Create necessary directories</em>
RUN mkdir -p /app/cache /app/logs
<em># Configure non-root user</em>
RUN useradd -m -u 1000 agent && \
chown -R agent:agent /app
USER agent
<em># Expose ports</em>
EXPOSE 8000
<em># Startup command</em>
CMD ["python", "-m", "src.agents.production_search_agent"]
docker-compose.yml
version: '3.8'
services:
search-agent:
build: .
container_name: production-search-agent
environment:
- TAVILY_API_KEY=${TAVILY_API_KEY}
- LOG_LEVEL=INFO
volumes:
- ./cache:/app/cache
- ./logs:/app/logs
- ./config:/app/config
ports:
- "8000:8000"
restart: unless-stopped
healthcheck:
test: ["CMD", "python", "-c", "import requests; requests.get('http://localhost:8000/health')"]
interval: 30s
timeout: 10s
retries: 3
redis:
image: redis:7-alpine
container_name: agent-redis
ports:
- "6379:6379"
volumes:
- redis_data:/data
restart: unless-stopped
prometheus:
image: prom/prometheus:latest
container_name: agent-prometheus
ports:
- "9090:9090"
volumes:
- ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
restart: unless-stopped
volumes:
redis_data:
Appendix B: Testing
tests/test_production_agent.py
import pytest
import asyncio
from unittest.mock import Mock, patch
from src.agents.production_search_agent import (
ProductionSearchAgent,
SearchRequest,
SearchDepth,
SearchStatus
)
@pytest.fixture
def mock_agent():
"""Fixture for creating a mock agent"""
with patch('src.agents.production_search_agent.TavilyClient'):
agent = ProductionSearchAgent(
tavily_api_key="test-key",
model="gpt-4"
)
return agent
class TestProductionSearchAgent:
def test_agent_initialization(self, mock_agent):
"""Test agent initialization"""
assert mock_agent is not None
assert mock_agent.stats["total_requests"] == 0
assert mock_agent.max_retries == 3
def test_search_request_creation(self):
"""Test search request creation"""
request = SearchRequest(
query="test query",
depth=SearchDepth.BASIC
)
assert request.query == "test query"
assert request.depth == SearchDepth.BASIC
assert request.timestamp is not None
cache_key = request.get_cache_key()
assert len(cache_key) == 32 <em># MD5 hash</em>
@patch('src.agents.production_search_agent.TavilyClient')
def test_successful_search(self, mock_tavily, mock_agent):
"""Test successful search"""
<em># Mock setup</em>
mock_tavily_instance = Mock()
mock_tavily.return_value = mock_tavily_instance
mock_tavily_instance.search.return_value = {
'answer': 'Test answer',
'results': [{'title': 'Test', 'url': 'http://test.com'}]
}
<em># Perform search</em>
request = SearchRequest("test query")
response = mock_agent._execute_search(request)
<em># Assertions</em>
assert response.status == SearchStatus.SUCCESS
assert response.answer == 'Test answer'
assert len(response.sources) == 1
assert not response.from_cache
def test_cache_functionality(self, mock_agent):
"""Test cache functionality"""
cache_key = "test_key"
test_data = {"answer": "cached answer", "sources": []}
<em># Save to cache</em>
mock_agent.cache_manager.set(cache_key, test_data)
<em># Get from cache</em>
cached_data = mock_agent.cache_manager.get(cache_key)
assert cached_data is not None
assert cached_data["answer"] == "cached answer"
def test_rate_limiting(self, mock_agent):
"""Test request rate limiting"""
rate_limiter = mock_agent.rate_limiter
<em># Fill limit</em>
for _ in range(rate_limiter.max_requests):
rate_limiter.register_request()
<em># Check blocking</em>
assert not rate_limiter.can_make_request()
<em># Check wait time</em>
wait_time = rate_limiter.get_wait_time()
assert wait_time > 0
def test_statistics_tracking(self, mock_agent):
"""Test statistics tracking"""
initial_stats = mock_agent.get_statistics()
assert initial_stats["total_requests"] == 0
assert initial_stats["success_rate"] == "0.0%"
<em># Simulate requests</em>
mock_agent.stats["total_requests"] = 10
mock_agent.stats["successful_requests"] = 8
mock_agent.stats["cached_requests"] = 2
updated_stats = mock_agent.get_statistics()
assert updated_stats["total_requests"] == 10
assert updated_stats["success_rate"] == "80.0%"
assert updated_stats["cache_hit_rate"] == "20.0%"
def test_health_check(self, mock_agent):
"""Test system health check"""
health = mock_agent.health_check()
assert "status" in health
assert "timestamp" in health
assert health["status"] in ["healthy", "unhealthy"]
@pytest.mark.asyncio
async def test_async_functionality():
"""Test asynchronous functionality"""
<em># Asynchronous operations can be tested here</em>
<em># if they are added in the future</em>
pass
if __name__ == "__main__":
pytest.main([__file__, "-v"])
Final Recommendations
1. Production Deployment
Containerization: Use Docker for dependency isolation and simplified deployment.
Monitoring: Set up a monitoring system (Prometheus + Grafana) to track performance.
Logging: Use centralized logging (ELK stack or similar).
2. Security
<em># Secure storage of API keys</em>
import os
from cryptography.fernet import Fernet
class SecureConfig:
"""Secure configuration management"""
def __init__(self, key_file: str = ".encryption_key"):
self.key_file = key_file
self.key = self._load_or_create_key()
self.cipher = Fernet(self.key)
def _load_or_create_key(self) -> bytes:
"""Load or create encryption key"""
if os.path.exists(self.key_file):
with open(self.key_file, 'rb') as f:
return f.read()
else:
key = Fernet.generate_key()
with open(self.key_file, 'wb') as f:
f.write(key)
return key
def encrypt_api_key(self, api_key: str) -> str:
"""Encrypt API key"""
return self.cipher.encrypt(api_key.encode()).decode()
def decrypt_api_key(self, encrypted_key: str) -> str:
"""Decrypt API key"""
return self.cipher.decrypt(encrypted_key.encode()).decode()
<em># Usage</em>
config = SecureConfig()
encrypted_key = config.encrypt_api_key("tvly-YOUR_API_KEY")
<em># Save encrypted_key in config</em>
<em># When using:</em>
api_key = config.decrypt_api_key(encrypted_key)
3. Scaling
<em># Example microservice architecture</em>
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
import uvicorn
app = FastAPI(title="Search Agent API", version="1.0.0")
class SearchQuery(BaseModel):
query: str
depth: str = "basic"
user_id: str = None
class SearchResponse(BaseModel):
query: str
answer: str
sources: list
response_time: float
from_cache: bool
<em># Global agent instance</em>
agent = ProductionSearchAgent(tavily_api_key=os.getenv("TAVILY_API_KEY"))
@app.post("/search", response_model=SearchResponse)
async def search_endpoint(query: SearchQuery, background_tasks: BackgroundTasks):
"""Endpoint for search query"""
try:
result = agent.search(query.query, SearchDepth(query.depth))
<em># Asynchronous logging</em>
background_tasks.add_task(log_search_request, query, result)
return SearchResponse(
query=query.query,
answer=result,
sources=[], <em># Populate with actual sources</em>
response_time=0.0, <em># Populate with actual time</em>
from_cache=False <em># Populate with actual status</em>
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/health")
async def health_check():
"""Service health check"""
return agent.health_check()
@app.get("/stats")
async def get_statistics():
"""Get statistics"""
return agent.get_statistics()
def log_search_request(query: SearchQuery, result: str):
"""Asynchronous logging of request"""
<em># Log to database or external system</em>
pass
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
4. Monitoring and Alerting
<em># monitoring/prometheus.yml</em>
global:
scrape_interval: 15s
scrape_configs:
- job_name: 'search-agent'
static_configs:
- targets: ['search-agent:8000']
metrics_path: '/metrics'
scrape_interval: 30s
rule_files:
- "alert_rules.yml"
alerting:
alertmanagers:
- static_configs:
- targets:
- alertmanager:9093
<em># monitoring/alert_rules.yml</em>
groups:
- name: search_agent_alerts
rules:
- alert: HighErrorRate
expr: (rate(search_agent_failed_requests_total[5m]) / rate(search_agent_total_requests_total[5m])) > 0.1
for: 2m
labels:
severity: warning
annotations:
summary: "High error rate detected"
description: "Error rate is {{ $value | humanizePercentage }}"
- alert: SlowResponseTime
expr: search_agent_avg_response_time_seconds > 5
for: 5m
labels:
severity: critical
annotations:
summary: "Slow response time"
description: "Average response time is {{ $value }}s"
5. Continuous Integration
<em># .github/workflows/ci.yml</em>
name: CI/CD Pipeline
on:
push:
branches: [main, develop]
pull_request:
branches: [main]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
pip install -r requirements.txt
pip install -r requirements-test.txt
- name: Run tests
run: |
pytest tests/ --cov=src --cov-report=xml
- name: Upload coverage
uses: codecov/codecov-action@v3
security:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Run security scan
run: |
pip install bandit safety
bandit -r src/
safety check
build:
needs: [test, security]
runs-on: ubuntu-latest
if: github.ref == 'refs/heads/main'
steps:
- uses: actions/checkout@v3
- name: Build Docker image
run: |
docker build -t search-agent:latest .
- name: Push to registry
run: |
echo ${{ secrets.DOCKER_PASSWORD }} | docker login -u ${{ secrets.DOCKER_USERNAME }} --password-stdin
docker push search-agent:latest
6. API Documentation
<em># Extended FastAPI documentation</em>
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
openapi_schema = get_openapi(
title="Smart Search Agent API",
version="2.0.0",
description="""
## Smart Search Agent with Tavily Integration
This API provides access to a smart search agent,
which uses up-to-date information from the internet.
### Capabilities:
- Real-time search for up-to-date information
- Result caching for improved performance
- Request rate limiting
- Detailed usage statistics
- System health monitoring
### Authentication:
Use the API key in the `X-API-Key` header
""",
routes=app.routes,
)
openapi_schema["info"]["x-logo"] = {
"url": "https://example.com/logo.png"
}
app.openapi_schema = openapi_schema
return app.openapi_schema
app.openapi = custom_openapi
Concluding Remarks
The integration of any-agent with Tavily creates a powerful platform for developing modern AI applications. This technological combination solves the critical problem of access to up-to-date information and opens up new possibilities for creating intelligent systems.
Key Achievements:
🎯 Versatility: A unified interface for working with various agent frameworks
⚡ Timeliness: Access to fresh information in real-time through a specialized search engine
🔧 Flexibility: Ability to create both simple agents and complex multi-agent systems
📊 Reliability: A complete set of tools for production: monitoring, caching, error handling
🚀 Scalability: Ready for deployment in an enterprise environment
Application in Real Projects:
- Corporate assistants with access to up-to-date information
- Market analysis systems for financial organizations
- Research platforms for scientific institutions
- News aggregators and analytical systems
- Educational assistants with current knowledge
Future Development:
AI agent technologies are rapidly evolving. The combination of any-agent + Tavily creates a strong foundation for adapting to new frameworks and capabilities, ensuring long-term value for development investments.
This guide provides a complete development cycle – from basic examples to production solutions. Use it as a reference for creating your own smart agents with access to up-to-date information.
Good luck with your development! 🤖✨