双重搜索功能,同时使用Tavily和Kimi搜索引擎,提供更全面、准确的搜索结果。
pip install aiohttp python-dotenv
在.env文件中添加:
TAVILY_API_KEY=tvly-xxxxxxxxxxxx
KIMI_API_KEY=your_kimi_api_key_here
或者设置系统环境变量:
export TAVILY_API_KEY="tvly-xxxxxxxxxxxx"
export KIMI_API_KEY="your_kimi_api_key_here"
from double_search import DoubleSearcher
async def search_example():
searcher = DoubleSearcher()
# 执行搜索
results = await searcher.search("人工智能发展趋势")
# 打印结果
print(f"搜索结果: {results}")
# 查看各个搜索引擎的结果
print(f"Tavily结果: {results.get('tavily', [])}")
print(f"Kimi结果: {results.get('kimi', [])}")
return results
searcher = DoubleSearcher()
async def search(
query: str,
merge_results: bool = True,
limit_per_source: int = 5
) -> Dict[str, Any]
参数:
query (str): 搜索查询字符串merge_results (bool): 是否合并结果(默认True)limit_per_source (int): 每个搜索源返回的结果数量(默认5)返回:
{
"query": "搜索内容",
"merged_results": [
{
"title": "结果标题",
"url": "https://example.com",
"snippet": "结果摘要",
"source": "tavily"
}
],
"source_breakdown": {
"tavily": [...],
"kimi": [...]
}
}
results = await searcher.search(
query="技术趋势",
merge_results=False
)
# 结果按搜索源分开
print(results['tavily'])
print(results['kimi'])
results = await searcher.search(
query="市场分析",
limit_per_source=3
)
results = await searcher.search("投资策略")
# 只获取Tavily的结果
tavily_results = results.get('tavily', [])
# 只获取Kimi的结果
kimi_results = results.get('kimi', [])
# 获取合并的结果
all_results = results.get('merged_results', [])
# 必需
TAVILY_API_KEY=tvly-xxxxxxxxxxxx
# 可选(不设置则只使用Tavily)
KIMI_API_KEY=your_kimi_api_key_here
支持.env文件:
# .env文件示例
TAVILY_API_KEY=tvly-xxxxxxxxxxxx
KIMI_API_KEY=your_kimi_api_key_here
async def safe_search(query: str):
searcher = DoubleSearcher()
try:
results = await searcher.search(query)
if not results['merged_results']:
return {"error": "搜索无结果", "query": query}
return results
except Exception as e:
return {"error": f"搜索失败: {str(e)}", "query": query}
from functools import lru_cache
class CachedSearcher(DoubleSearcher):
@lru_cache(maxsize=100)
async def search(self, query: str, ...):
return await super().search(query, ...)
async def research_topic(topic):
searcher = DoubleSearcher()
results = await searcher.search(topic)
# 分析多个来源
insights = analyze_results(results)
return insights
async def financial_analysis(ticker):
searcher = DoubleSearcher()
query = f"{ticker} 财经分析"
results = await searcher.search(query)
# 过滤财经相关结果
financial_news = filter_financial_content(results)
return financial_news
async def code_search(problem):
searcher = DoubleSearcher()
results = await searcher.search(problem)
# 优先获取技术相关结果
technical_results = filter_tech_content(results)
return technical_results
# 检查API keys
echo $TAVILY_API_KEY
echo $KIMI_API_KEY
# 验证API keys有效性
# 确保在正确的工作目录
cd ~/.openclaw/skills/double-search
# 检查Python路径
which python3
# 安装依赖
pip install aiohttp python-dotenv
共 1 个版本