基于Tavily Search Skill的智能搜索解决方案：从技术选型到生产实践

6次阅读

共计 2628 个字符，预计需要花费 7 分钟才能阅读完成。

在开发搜索功能时，我们常常面临几个核心问题：数据获取成本高、响应速度慢、结果准确性不足。传统方案如自建爬虫需要大量维护工作，而公开 API 往往有调用频率限制或高延迟。比如：

爬虫方案需处理反爬机制、IP 封禁和页面结构变化
第三方 API 如 Google Custom Search 有每日限额（100 次 / 天免费版）
Elasticsearch 需要自行构建和维护索引

对比主流搜索方案的关键指标：

方案	开发成本	响应速度	准确度	扩展性
自建爬虫	高	中	中	低
Elasticsearch	中	高	高	高
Algolia	低	高	高	中
Tavily Search	低	高	高	高

Tavily 的核心优势在于：

无需管理基础设施
聚合多个搜索引擎结果
内置智能排序算法
免费版提供 300 次 / 月调用

以下是 Python 集成示例（需先 pip install tavily-python）：

from tavily import TavilyClient
import asyncio

# 初始化客户端（建议将 API_KEY 放入环境变量）tavily = TavilyClient(api_key="your_api_key")

async def enhanced_search(query: str, max_results=5):
    """
    执行智能搜索并返回结构化结果
    :param query: 搜索关键词
    :param max_results: 最大返回数量
    :return: 结果列表（包含标题、URL、摘要）"""
    try:
        response = await tavily.search_async(
            query=query,
            search_depth="basic",  # 可选 advanced
            include_raw_content=False,
            max_results=max_results
        )
        return [{'title': r['title'],
            'url': r['url'],
            'content': r['content'][:200] + '...'  # 摘要截断
        } for r in response['results']]
    except Exception as e:
        print(f"搜索失败: {str(e)}")
        return []

# 使用示例
results = asyncio.run(enhanced_search("Python 异步编程最佳实践"))
for idx, item in enumerate(results, 1):
    print(f"{idx}. {item['title']}\n   {item['url']}\n   {item['content']}\n")

关键参数说明：

search_depth: basic 适合常规查询（约 0.5s 响应），advance 用于深度检索（2-3s）
include_raw_content: 为 True 时返回完整页面内容（注意隐私合规）
max_results: 控制返回条目数（免费版上限 50）

from datetime import timedelta
from functools import lru_cache

# 内存缓存（适合短期高频相同查询）@lru_cache(maxsize=1024)
def cached_search(query: str):
    return tavily.search(query)

# Redis 缓存示例（需安装 redis-py）import redis
r = redis.Redis(host='localhost')

def redis_cached_search(query: str, ttl=3600):
    cache_key = f"search:{query}"
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)

    result = tavily.search(query)
    r.setex(cache_key, timedelta(seconds=ttl), json.dumps(result))
    return result

当需要同时处理多个搜索请求时：

async def batch_search(queries: list):
    tasks = [enhanced_search(q) for q in queries]
    return await asyncio.gather(*tasks, return_exceptions=True)

Tavily 的免费版限制为 5 次 / 秒，建议：

实现请求队列
添加指数退避重试

import time

def rate_limited_search(query, max_retries=3):
    for attempt in range(max_retries):
        try:
            return tavily.search(query)
        except Exception as e:
            if "rate limit" in str(e).lower():
                wait = 2 ** attempt  # 指数退避
                time.sleep(wait)
                continue
            raise
    raise Exception("超出最大重试次数")

对相似 URL 的合并处理：

from urllib.parse import urlparse

def deduplicate_results(results):
    seen = set()
    unique = []
    for r in results:
        # 标准化 URL（去除查询参数）parsed = urlparse(r['url'])
        netloc = parsed.netloc.replace('www.', '')
        path = parsed.path.rstrip('/')
        key = f"{netloc}{path}"

        if key not in seen:
            seen.add(key)
            unique.append(r)
    return unique