import asyncio
from langchain_perigon import (
ArticlesRetriever,
WikipediaRetriever,
ArticlesFilter,
WikipediaOptions,
)
async def search_both():
"""Perform concurrent searches across news articles and Wikipedia.
Returns:
tuple: (news_articles, wikipedia_docs) - Results from both retrievers
Raises:
Exception: If either retriever fails or API errors occur
"""
# Initialize retrievers with automatic API key detection
articles_retriever = ArticlesRetriever()
wiki_retriever = WikipediaRetriever()
# Configure search options for targeted results
articles_options: ArticlesFilter = {
"size": 3, # Limit to 3 articles for faster response
"filter": {
"country": "us", # US-based news sources
"category": "tech", # Technology category only
},
}
# Filter Wikipedia results by popularity (pageviews)
wiki_options: WikipediaOptions = {
"size": 3, # Limit to 3 articles
"pageviewsFrom": 50 # Only articles with 50+ daily views
}
try:
# Perform concurrent async searches for better performance
articles_task = articles_retriever.ainvoke(
"climate change", options=articles_options
)
wiki_task = wiki_retriever.ainvoke(
"climate change", options=wiki_options
)
# Wait for both searches to complete simultaneously
articles, wiki_docs = await asyncio.gather(
articles_task, wiki_task, return_exceptions=True
)
# Handle potential exceptions from either retriever
if isinstance(articles, Exception):
print(f"Articles retrieval failed: {articles}")
articles = []
if isinstance(wiki_docs, Exception):
print(f"Wikipedia retrieval failed: {wiki_docs}")
wiki_docs = []
return articles, wiki_docs
except Exception as e:
print(f"Error in concurrent search: {e}")
return [], []
# Run async search with error handling
try:
articles, wiki_docs = asyncio.run(search_both())
# Display results summary
print(f"Found {len(articles)} news articles and {len(wiki_docs)} Wikipedia articles")
# Show sample results if available
if articles:
print(f"Sample article: {articles[0].metadata.get('title', 'N/A')}")
if wiki_docs:
print(f"Sample Wikipedia: {wiki_docs[0].metadata.get('title', 'N/A')}")
except Exception as e:
print(f"Async search failed: {e}")