Tavily AI搜索API集成开发完全教程

构建生产级AI搜索应用，从API接入到企业级部署

目录

1. 概述与背景
2. Tavily Search API架构与接入
3. 搜索结果结构化处理
4. 与LangChain集成
5. 与LlamaIndex集成
6. Agent搜索链构建
7. 实时信息检索RAG
8. 搜索结果过滤与重排序
9. 成本控制策略
10. 企业级应用场景
11. 与Google/Bing API对比
12. 生产级部署方案

1. 概述与背景

在AI应用开发中，让大语言模型（LLM）获取实时信息是一个核心挑战。传统搜索引擎返回的是面向人类的网页列表，而Tavily Search API专为AI Agent设计，返回的是结构化、可直接被LLM消费的搜索结果。

为什么需要Tavily？

传统搜索引擎的问题：

• 返回结果是HTML页面列表，需要额外解析
• 包含大量广告、SEO垃圾内容
• 不适合LLM直接消费
• 需要处理反爬虫、验证码等

Tavily的解决方案：

• 专为AI设计的搜索结果格式
• 返回摘要、关键信息和可信来源
• 支持搜索深度控制（basic/advanced）
• 提供相关性评分和来源可信度
• 开箱即用的LangChain/LlamaIndex集成

2. Tavily Search API架构与接入

2.1 注册与获取API Key

步骤：
1. 访问 https://tavily.com
2. 注册账号（支持Google/GitHub登录）
3. 进入 Dashboard → API Keys
4. 创建新的API Key
5. 免费套餐：每月1000次搜索调用

2.2 安装SDK

# Python SDK
pip install tavily-python

# Node.js SDK
npm install tavily

# 或直接使用HTTP API（无需SDK）

2.3 Python快速接入

from tavily import TavilyClient

# 初始化客户端
client = TavilyClient(api_key="tvly-YOUR_API_KEY")

# 基础搜索
response = client.search("2025年最新AI发展趋势")

# 查看结果
print(f"找到 {len(response['results'])} 条结果")
for result in response['results']:
    print(f"标题: {result['title']}")
    print(f"URL: {result['url']}")
    print(f"摘要: {result['content'][:200]}...")
    print(f"相关性: {result['score']:.2f}")
    print("---")

2.4 Node.js接入

import { tavily } from 'tavily';

const client = tavily({ apiKey: 'tvly-YOUR_API_KEY' });

async function search() {
  const response = await client.search('最新AI编程工具对比');

  response.results.forEach(result => {
    console.log(`标题: ${result.title}`);
    console.log(`URL: ${result.url}`);
    console.log(`内容: ${result.content.substring(0, 200)}...`);
    console.log('---');
  });
}

search();

2.5 直接HTTP API调用

import requests

def tavily_search(query: str, api_key: str) -> dict:
    """直接调用Tavily HTTP API"""
    url = "https://api.tavily.com/search"
    headers = {
        "Content-Type": "application/json",
    }
    payload = {
        "api_key": api_key,
        "query": query,
        "search_depth": "basic",     # basic 或 advanced
        "max_results": 5,             # 最大结果数
        "include_answer": True,       # 包含AI生成的答案
        "include_raw_content": False, # 包含原始网页内容
        "include_domains": [],        # 限定搜索域名
        "exclude_domains": [],        # 排除域名
    }

    response = requests.post(url, json=payload, headers=headers)
    return response.json()

# 使用
result = tavily_search("LangChain最新版本特性", "tvly-YOUR_API_KEY")
print(result.get("answer", "无答案"))

2.6 API参数详解

3. 搜索结果结构化处理

3.1 返回数据结构

Tavily的返回结构经过精心设计，便于LLM消费：

{
  "query": "2025年AI发展趋势",
  "answer": "2025年AI发展呈现三大趋势...",
  "results": [
    {
      "title": "2025年AI技术展望",
      "url": "https://example.com/ai-2025",
      "content": "AI领域在2025年经历了...",
      "score": 0.95,
      "raw_content": "完整网页文本..."
    }
  ],
  "response_time": 1.23
}

3.2 结果处理管道

from dataclasses import dataclass
from typing import Optional

@dataclass
class ProcessedResult:
    """处理后的搜索结果"""
    title: str
    url: str
    summary: str
    score: float
    domain: str
    relevance: str  # high/medium/low

class SearchResultProcessor:
    """搜索结果处理器"""

    def __init__(self, min_score: float = 0.5):
        self.min_score = min_score

    def process(self, raw_response: dict) -> list[ProcessedResult]:
        """处理原始API响应"""
        results = []

        for item in raw_response.get("results", []):
            # 过滤低分结果
            if item["score"] < self.min_score:
                continue

            # 提取域名
            domain = self._extract_domain(item["url"])

            # 分类相关性
            relevance = self._classify_relevance(item["score"])

            results.append(ProcessedResult(
                title=item["title"],
                url=item["url"],
                summary=item["content"],
                score=item["score"],
                domain=domain,
                relevance=relevance,
            ))

        # 按分数排序
        results.sort(key=lambda x: x.score, reverse=True)
        return results

    def _extract_domain(self, url: str) -> str:
        from urllib.parse import urlparse
        return urlparse(url).netloc

    def _classify_relevance(self, score: float) -> str:
        if score >= 0.8:
            return "high"
        elif score >= 0.6:
            return "medium"
        return "low"

    def format_for_llm(self, results: list[ProcessedResult]) -> str:
        """格式化为LLM友好的文本"""
        if not results:
            return "未找到相关搜索结果。"

        parts = [f"找到 {len(results)} 条相关结果：\n"]

        for i, r in enumerate(results, 1):
            parts.append(f"【{i}】{r.title}（相关性：{r.relevance}）")
            parts.append(f"来源：{r.url}")
            parts.append(f"摘要：{r.summary}")
            parts.append("")

        return "\n".join(parts)

# 使用示例
processor = SearchResultProcessor(min_score=0.6)
processed = processor.process(raw_response)
llm_text = processor.format_for_llm(processed)
print(llm_text)

3.3 结果去重与合并

from hashlib import md5

class ResultDeduplicator:
    """搜索结果去重器"""

    def __init__(self):
        self.seen_urls: set[str] = set()
        self.seen_content_hashes: set[str] = set()

    def deduplicate(self, results: list[dict]) -> list[dict]:
        """基于URL和内容去重"""
        unique_results = []

        for result in results:
            url = result["url"]

            # URL去重
            if url in self.seen_urls:
                continue

            # 内容去重（处理同一内容不同URL的情况）
            content_hash = md5(
                result["content"][:500].encode()
            ).hexdigest()
            if content_hash in self.seen_content_hashes:
                continue

            self.seen_urls.add(url)
            self.seen_content_hashes.add(content_hash)
            unique_results.append(result)

        return unique_results

4. 与LangChain集成

4.1 安装与基础使用

pip install langchain langchain-community langchain-openai tavily-python

from langchain_community.tools.tavily_search import TavilySearchResults

# 初始化搜索工具
search_tool = TavilySearchResults(
    max_results=5,
    search_depth="advanced",
)

# 直接调用
results = search_tool.invoke("LangChain最新版本特性")
print(results)

4.2 在Agent中使用

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.tools.tavily_search import TavilySearchResults

# 1. 初始化LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 2. 创建搜索工具
search = TavilySearchResults(max_results=3)

# 3. 定义Prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个专业的研究助手。
    当用户询问需要最新信息的问题时，使用搜索工具获取信息。
    基于搜索结果提供准确、有来源的回答。"""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

# 4. 创建Agent
agent = create_tool_calling_agent(llm, [search], prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=[search],
    verbose=True,
    max_iterations=3,
)

# 5. 运行
result = agent_executor.invoke({
    "input": "2025年最值得学习的编程语言是什么？"
})
print(result["output"])

4.3 自定义Tavily工具

from langchain.tools import BaseTool
from langchain.callbacks.manager import CallbackManagerForToolRun
from tavily import TavilyClient
from typing import Optional

class TavilyNewsSearch(BaseTool):
    """专注于新闻搜索的自定义Tavily工具"""

    name: str = "tavily_news_search"
    description: str = "搜索最新新闻和实时信息"
    client: TavilyClient = None

    def __init__(self, api_key: str, **kwargs):
        super().__init__(**kwargs)
        self.client = TavilyClient(api_key=api_key)

    def _run(
        self,
        query: str,
        run_manager: Optional[CallbackManagerForToolRun] = None,
    ) -> str:
        response = self.client.search(
            query=query,
            topic="news",
            days=7,
            max_results=5,
            include_answer=True,
        )

        # 格式化结果
        parts = []
        if response.get("answer"):
            parts.append(f"综合答案：{response['answer']}\n")

        for r in response.get("results", []):
            parts.append(f"• {r['title']}")
            parts.append(f"  {r['content'][:150]}...")
            parts.append(f"  来源：{r['url']}\n")

        return "\n".join(parts)

5. 与LlamaIndex集成

5.1 安装

pip install llama-index llama-index-tools-tavily

5.2 基础使用

from llama_index.tools.tavily import TavilyToolSpec

# 初始化
tavily_tool = TavilyToolSpec(
    api_key="tvly-YOUR_API_KEY",
)

# 获取工具列表
tools = tavily_tool.to_tool_list()
print(f"可用工具数：{len(tools)}")

# 直接搜索
documents = tavily_tool.load_data(
    query="Python异步编程最佳实践",
    max_results=5,
)

for doc in documents:
    print(f"内容：{doc.text[:200]}...")
    print(f"来源：{doc.metadata.get('source', 'unknown')}")
    print("---")

5.3 构建搜索增强的RAG

from llama_index.core import VectorStoreIndex, Settings
from llama_index.llms.openai import OpenAI
from llama_index.tools.tavily import TavilyToolSpec
from llama_index.core.agent import ReActAgent

# 1. 配置LLM
Settings.llm = OpenAI(model="gpt-4o", temperature=0)

# 2. 初始化搜索工具
tavily_tool = TavilyToolSpec(api_key="tvly-YOUR_API_KEY")
tools = tavily_tool.to_tool_list()

# 3. 创建ReAct Agent
agent = ReActAgent.from_tools(
    tools=tools,
    llm=Settings.llm,
    verbose=True,
    max_iterations=5,
)

# 4. 查询
response = agent.chat(
    "对比2025年主流前端框架的性能表现"
)
print(response)

6. Agent搜索链构建

6.1 搜索链架构

构建一个完整的搜索链，包括查询优化、多轮搜索、结果聚合：

from tavily import TavilyClient
from openai import OpenAI
from typing import TypedDict

class SearchChain:
    """多步搜索链"""

    def __init__(self, tavily_key: str, openai_key: str):
        self.tavily = TavilyClient(api_key=tavily_key)
        self.llm = OpenAI(api_key=openai_key)
        self.search_history: list[dict] = []

    def run(self, question: str, max_rounds: int = 3) -> str:
        """执行多轮搜索"""
        # 第一步：生成初始搜索查询
        queries = self._generate_queries(question)
        print(f"生成 {len(queries)} 个搜索查询")

        # 第二步：执行搜索
        all_results = []
        for query in queries[:max_rounds]:
            results = self.tavily.search(
                query=query,
                search_depth="advanced",
                max_results=3,
            )
            all_results.extend(results.get("results", []))
            self.search_history.append({
                "query": query,
                "result_count": len(results.get("results", [])),
            })

        # 第三步：聚合和生成答案
        answer = self._generate_answer(question, all_results)
        return answer

    def _generate_queries(self, question: str) -> list[str]:
        """使用LLM优化搜索查询"""
        prompt = f"""基于以下问题，生成2-3个不同的搜索查询，
        以获取最全面的信息。每个查询一行。

        问题：{question}

        搜索查询："""

        response = self.llm.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
        )

        queries = response.choices[0].message.content.strip().split("\n")
        return [q.strip() for q in queries if q.strip()]

    def _generate_answer(
        self, question: str, results: list[dict]
    ) -> str:
        """基于搜索结果生成答案"""
        # 格式化搜索结果
        context_parts = []
        for i, r in enumerate(results[:10], 1):
            context_parts.append(
                f"[{i}] {r['title']}\n{r['content'][:300]}\n来源：{r['url']}"
            )
        context = "\n\n".join(context_parts)

        prompt = f"""基于以下搜索结果回答问题。引用来源编号。

搜索结果：
{context}

问题：{question}

请提供详细、准确的回答，并列出参考来源："""

        response = self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
        )

        return response.choices[0].message.content

# 使用
chain = SearchChain(
    tavily_key="tvly-YOUR_KEY",
    openai_key="sk-YOUR_KEY",
)
answer = chain.run("2025年量子计算最新进展")
print(answer)

6.2 并行搜索优化

import asyncio
from tavily import AsyncTavilyClient

class ParallelSearchChain:
    """并行搜索链"""

    def __init__(self, api_key: str):
        self.client = AsyncTavilyClient(api_key=api_key)

    async def search_multiple(
        self, queries: list[str]
    ) -> list[dict]:
        """并行执行多个搜索"""
        tasks = [
            self.client.search(
                query=q,
                max_results=3,
                search_depth="basic",
            )
            for q in queries
        ]

        results = await asyncio.gather(*tasks, return_exceptions=True)

        # 合并成功的结果
        all_results = []
        for result in results:
            if isinstance(result, dict):
                all_results.extend(result.get("results", []))

        return all_results

# 使用
async def main():
    chain = ParallelSearchChain("tvly-YOUR_KEY")
    queries = ["AI Agent框架对比", "LLM应用开发趋势", "RAG技术最新进展"]
    results = await chain.search_multiple(queries)
    print(f"并行搜索获得 {len(results)} 条结果")

asyncio.run(main())

7. 实时信息检索RAG

7.1 搜索增强RAG架构

from tavily import TavilyClient
from openai import OpenAI
from typing import Optional

class SearchAugmentedRAG:
    """搜索增强的RAG系统"""

    def __init__(
        self,
        tavily_key: str,
        openai_key: str,
        model: str = "gpt-4o",
    ):
        self.search = TavilyClient(api_key=tavily_key)
        self.llm = OpenAI(api_key=openai_key)
        self.model = model
        self.cache: dict[str, str] = {}

    def query(
        self,
        question: str,
        use_cache: bool = True,
        search_depth: str = "basic",
    ) -> dict:
        """执行RAG查询"""
        # 检查缓存
        if use_cache and question in self.cache:
            return {"answer": self.cache[question], "cached": True}

        # 判断是否需要搜索
        needs_search = self._should_search(question)

        if needs_search:
            # 获取搜索结果
            search_results = self.search.search(
                query=question,
                search_depth=search_depth,
                max_results=5,
                include_answer=True,
            )
            context = self._format_context(search_results)
        else:
            context = ""

        # 生成答案
        answer = self._generate_answer(question, context, needs_search)

        # 缓存结果
        if use_cache:
            self.cache[question] = answer

        return {
            "answer": answer,
            "searched": needs_search,
            "sources": self._extract_sources(search_results) if needs_search else [],
        }

    def _should_search(self, question: str) -> bool:
        """判断是否需要搜索"""
        prompt = f"""判断以下问题是否需要搜索最新信息才能准确回答。
        只回答 "yes" 或 "no"。

        问题：{question}

        需要搜索："""

        response = self.llm.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
            max_tokens=5,
        )

        return "yes" in response.choices[0].message.content.lower()

    def _format_context(self, search_results: dict) -> str:
        """格式化搜索上下文"""
        parts = []

        # AI生成的答案
        if search_results.get("answer"):
            parts.append(f"综合摘要：{search_results['answer']}\n")

        # 详细结果
        for i, r in enumerate(search_results.get("results", []), 1):
            parts.append(f"[来源{i}] {r['title']}")
            parts.append(f"URL: {r['url']}")
            parts.append(f"内容: {r['content']}")
            parts.append("")

        return "\n".join(parts)

    def _generate_answer(
        self, question: str, context: str, has_context: bool
    ) -> str:
        """生成最终答案"""
        if has_context:
            system_prompt = """你是一个基于搜索结果回答问题的助手。
请基于提供的搜索结果给出准确、详细的回答。
引用来源编号（如 [1][2]）来支持你的观点。
如果搜索结果不足以回答问题，请明确说明。"""
        else:
            system_prompt = """你是一个知识渊博的助手。
请基于你的知识直接回答问题。"""

        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"搜索结果：\n{context}\n\n问题：{question}"},
        ]

        response = self.llm.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=0.3,
        )

        return response.choices[0].message.content

    def _extract_sources(self, search_results: dict) -> list[dict]:
        """提取来源信息"""
        return [
            {"title": r["title"], "url": r["url"]}
            for r in search_results.get("results", [])
        ]

# 使用示例
rag = SearchAugmentedRAG(
    tavily_key="tvly-YOUR_KEY",
    openai_key="sk-YOUR_KEY",
)

# 会触发搜索（需要最新信息）
result1 = rag.query("2025年OpenAI最新发布了什么模型？")
print(result1["answer"])

# 不会触发搜索（通用知识）
result2 = rag.query("什么是Python的装饰器？")
print(result2["answer"])

8. 搜索结果过滤与重排序

8.1 基于规则的过滤

class SearchFilter:
    """搜索结果过滤器"""

    def __init__(self):
        self.blocked_domains: set[str] = set()
        self.required_keywords: list[str] = []
        self.min_content_length: int = 100

    def add_blocked_domain(self, domain: str):
        self.blocked_domains.add(domain)

    def filter(self, results: list[dict]) -> list[dict]:
        """应用所有过滤规则"""
        filtered = []
        for r in results:
            # 域名过滤
            if self._is_blocked(r.get("url", "")):
                continue
            # 内容长度过滤
            if len(r.get("content", "")) < self.min_content_length:
                continue
            # 关键词过滤
            if self.required_keywords and not self._has_keywords(r):
                continue
            filtered.append(r)
        return filtered

    def _is_blocked(self, url: str) -> bool:
        from urllib.parse import urlparse
        domain = urlparse(url).netloc
        return any(blocked in domain for blocked in self.blocked_domains)

    def _has_keywords(self, result: dict) -> bool:
        text = (result.get("title", "") + " " + result.get("content", "")).lower()
        return any(kw.lower() in text for kw in self.required_keywords)

# 使用
filter = SearchFilter()
filter.add_blocked_domain("pinterest.com")
filter.min_content_length = 200
filtered = filter.filter(raw_results)

8.2 基于LLM的重排序

class LLMReranker:
    """使用LLM对搜索结果重排序"""

    def __init__(self, openai_key: str):
        self.llm = OpenAI(api_key=openai_key)

    def rerank(
        self, query: str, results: list[dict], top_k: int = 5
    ) -> list[dict]:
        """基于查询相关性重排序"""
        # 构建评估提示
        results_text = "\n".join(
            f"[{i}] {r['title']}: {r['content'][:100]}..."
            for i, r in enumerate(results)
        )

        prompt = f"""对以下搜索结果按照与查询的相关性排序。
返回结果编号，从最相关到最不相关，用逗号分隔。

查询：{query}

搜索结果：
{results_text}

排序（仅返回编号，如 2,0,4,1,3）："""

        response = self.llm.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
        )

        # 解析排序结果
        try:
            order = [
                int(x.strip())
                for x in response.choices[0].message.content.split(",")
            ]
            reranked = [results[i] for i in order if i < len(results)]
            return reranked[:top_k]
        except (ValueError, IndexError):
            return results[:top_k]

9. 成本控制策略

9.1 Tavily定价方案

9.2 搜索结果缓存

import hashlib
import json
import time
from pathlib import Path

class TavilyCache:
    """搜索结果缓存"""

    def __init__(self, cache_dir: str = ".tavily_cache", ttl: int = 3600):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)
        self.ttl = ttl  # 缓存过期时间（秒）

    def _make_key(self, query: str, params: dict) -> str:
        """生成缓存键"""
        content = json.dumps({"query": query, **params}, sort_keys=True)
        return hashlib.sha256(content.encode()).hexdigest()

    def get(self, query: str, params: dict) -> dict | None:
        """获取缓存结果"""
        key = self._make_key(query, params)
        cache_file = self.cache_dir / f"{key}.json"

        if not cache_file.exists():
            return None

        data = json.loads(cache_file.read_text())

        # 检查是否过期
        if time.time() - data["timestamp"] > self.ttl:
            cache_file.unlink()
            return None

        return data["result"]

    def set(self, query: str, params: dict, result: dict):
        """设置缓存"""
        key = self._make_key(query, params)
        cache_file = self.cache_dir / f"{key}.json"

        data = {
            "timestamp": time.time(),
            "query": query,
            "result": result,
        }
        cache_file.write_text(json.dumps(data, ensure_ascii=False))

# 使用
cache = TavilyCache(ttl=7200)  # 2小时缓存

def cached_search(client, query: str, **params):
    """带缓存的搜索"""
    cached = cache.get(query, params)
    if cached:
        return cached

    result = client.search(query=query, **params)
    cache.set(query, params, result)
    return result

9.3 查询优化减少调用

class QueryOptimizer:
    """查询优化器，减少不必要的API调用"""

    def __init__(self):
        self.recent_queries: dict[str, float] = {}

    def should_search(self, query: str, min_interval: int = 60) -> bool:
        """判断是否需要发起新搜索"""
        normalized = query.lower().strip()

        # 检查是否为重复查询
        if normalized in self.recent_queries:
            elapsed = time.time() - self.recent_queries[normalized]
            if elapsed < min_interval:
                return False

        self.recent_queries[normalized] = time.time()
        return True

    def simplify_query(self, query: str) -> str:
        """简化查询以提高搜索效率"""
        # 移除冗余词汇
        stopwords = {"的", "了", "吗", "呢", "吧", "是", "在", "有"}
        words = query.split()
        simplified = [w for w in words if w not in stopwords]
        return " ".join(simplified) if simplified else query

9.4 成本监控

class CostTracker:
    """API成本追踪器"""

    # Tavily 每次搜索成本（美元）
    COST_PER_SEARCH = {
        "basic": 0.001,
        "advanced": 0.002,
    }

    def __init__(self, monthly_budget: float = 50.0):
        self.monthly_budget = monthly_budget
        self.usage_file = Path(".tavily_usage.json")
        self._load_usage()

    def _load_usage(self):
        if self.usage_file.exists():
            self.usage = json.loads(self.usage_file.read_text())
        else:
            self.usage = {"total_calls": 0, "total_cost": 0.0, "by_month": {}}

    def record_call(self, search_depth: str = "basic"):
        """记录一次API调用"""
        cost = self.COST_PER_SEARCH.get(search_depth, 0.001)
        month_key = time.strftime("%Y-%m")

        self.usage["total_calls"] += 1
        self.usage["total_cost"] += cost

        if month_key not in self.usage["by_month"]:
            self.usage["by_month"][month_key] = {"calls": 0, "cost": 0.0}

        self.usage["by_month"][month_key]["calls"] += 1
        self.usage["by_month"][month_key]["cost"] += cost

        self._save_usage()

        # 检查预算
        monthly_cost = self.usage["by_month"][month_key]["cost"]
        if monthly_cost > self.monthly_budget * 0.8:
            print(f"⚠️ 警告：本月已使用 {monthly_cost:.2f} 美元，"
                  f"接近预算 {self.monthly_budget} 美元")

    def _save_usage(self):
        self.usage_file.write_text(
            json.dumps(self.usage, indent=2, ensure_ascii=False)
        )

    def get_report(self) -> str:
        """生成使用报告"""
        month_key = time.strftime("%Y-%m")
        month_data = self.usage["by_month"].get(month_key, {"calls": 0, "cost": 0.0})

        return f"""Tavily 使用报告
================
本月调用次数: {month_data['calls']}
本月费用: ${month_data['cost']:.2f}
月度预算: ${self.monthly_budget:.2f}
预算使用率: {(month_data['cost'] / self.monthly_budget * 100):.1f}%
"""

10. 企业级应用场景

10.1 竞品监控系统

class CompetitorMonitor:
    """竞品信息监控"""

    def __init__(self, tavily_key: str, competitors: list[str]):
        self.client = TavilyClient(api_key=tavily_key)
        self.competitors = competitors

    def daily_scan(self) -> dict[str, list[dict]]:
        """每日竞品扫描"""
        report = {}

        for competitor in self.competitors:
            results = self.client.search(
                query=f"{competitor} 最新动态 产品发布",
                topic="news",
                days=1,
                max_results=5,
                search_depth="advanced",
            )

            report[competitor] = [
                {
                    "title": r["title"],
                    "summary": r["content"][:200],
                    "url": r["url"],
                    "score": r["score"],
                }
                for r in results.get("results", [])
                if r["score"] > 0.7
            ]

        return report

    def generate_summary(self, report: dict) -> str:
        """生成竞品摘要"""
        parts = ["# 竞品每日监控报告\n"]

        for competitor, items in report.items():
            parts.append(f"## {competitor}")
            if not items:
                parts.append("今日无新动态。\n")
                continue

            for item in items:
                parts.append(f"- **{item['title']}**")
                parts.append(f"  {item['summary']}")
                parts.append(f"  [详情]({item['url']})\n")

        return "\n".join(parts)

10.2 智能客服知识库

class CustomerServiceRAG:
    """智能客服搜索增强"""

    def __init__(self, tavily_key: str):
        self.search = TavilyClient(api_key=tavily_key)
        self.product_docs: dict[str, str] = {}

    def answer(self, question: str, product: str = None) -> dict:
        """回答客户问题"""
        # 首先搜索内部文档
        internal_answer = self._search_internal(question, product)

        # 如果内部文档不足以回答，搜索外部
        if not internal_answer:
            external_results = self.search.search(
                query=f"{product or ''} {question}",
                max_results=3,
                include_answer=True,
            )
            return {
                "answer": external_results.get("answer", "无法找到相关信息"),
                "sources": [r["url"] for r in external_results.get("results", [])],
                "confidence": "low",
            }

        return {
            "answer": internal_answer,
            "sources": ["internal_docs"],
            "confidence": "high",
        }

    def _search_internal(self, question: str, product: str) -> str | None:
        """搜索内部文档"""
        # 这里可以接入向量数据库搜索内部知识库
        # 示例：简单的关键词匹配
        for doc_id, content in self.product_docs.items():
            if any(word in content for word in question.split()):
                return content
        return None

10.3 投资研究助手

class InvestmentResearcher:
    """投资研究助手"""

    def __init__(self, tavily_key: str):
        self.client = TavilyClient(api_key=tavily_key)

    def research_stock(self, ticker: str) -> dict:
        """研究股票信息"""
        queries = [
            f"{ticker} stock analysis 2025",
            f"{ticker} financial results earnings",
            f"{ticker} market news latest",
        ]

        results = {}
        for query in queries:
            response = self.client.search(
                query=query,
                search_depth="advanced",
                max_results=3,
                include_answer=True,
            )
            results[query] = response

        return self._compile_analysis(ticker, results)

    def _compile_analysis(self, ticker: str, results: dict) -> dict:
        """编译分析报告"""
        all_sources = []
        for response in results.values():
            all_sources.extend(response.get("results", []))

        # 去重
        seen_urls = set()
        unique_sources = []
        for source in all_sources:
            if source["url"] not in seen_urls:
                seen_urls.add(source["url"])
                unique_sources.append(source)

        return {
            "ticker": ticker,
            "sources_count": len(unique_sources),
            "sources": unique_sources[:10],
            "answers": {
                k: v.get("answer") for k, v in results.items() if v.get("answer")
            },
        }

11. 与Google/Bing API对比

11.1 功能对比

11.2 代码对比

# Tavily - 3行代码获取结构化结果
from tavily import TavilyClient
client = TavilyClient(api_key="tvly-xxx")
results = client.search("AI trends 2025", include_answer=True)
# results["answer"] 直接获得AI摘要
# results["results"] 结构化结果，带相关性评分

# Google Custom Search - 需要额外处理
from googleapiclient.discovery import build
service = build("customsearch", "v1", developerKey="xxx")
results = service.cse().list(q="AI trends 2025", cx="xxx").execute()
# results["items"] 需要解析HTML片段
# 无AI摘要，无相关性评分

# Bing Search API - 需要额外处理
import requests
headers = {"Ocp-Apim-Subscription-Key": "xxx"}
response = requests.get(
    "https://api.bing.microsoft.com/v7.0/search",
    headers=headers,
    params={"q": "AI trends 2025"},
)
# response.json()["webPages"]["value"] 需要解析
# 无AI摘要，无相关性评分

11.3 选择建议

选择Tavily的场景：
- 构建AI Agent和聊天机器人
- 需要即开即用的搜索结果
- 需要相关性评分和AI摘要
- 快速原型开发
- 预算敏感的小团队

选择Google/Bing的场景：
- 需要最大搜索覆盖范围
- 需要图片/视频等多媒体搜索
- 已有大量搜索API配额
- 需要自定义搜索引擎功能
- 企业级合规要求

12. 生产级部署方案

12.1 环境配置

# config.py - 生产环境配置
import os
from dataclasses import dataclass

@dataclass
class TavilyConfig:
    api_key: str = os.getenv("TAVILY_API_KEY", "")
    max_results: int = 5
    search_depth: str = "basic"
    cache_ttl: int = 3600
    monthly_budget: float = 100.0
    retry_attempts: int = 3
    retry_delay: float = 1.0
    timeout: int = 30

    @classmethod
    def from_env(cls) -> "TavilyConfig":
        return cls(
            api_key=os.getenv("TAVILY_API_KEY", ""),
            max_results=int(os.getenv("TAVILY_MAX_RESULTS", "5")),
            search_depth=os.getenv("TAVILY_SEARCH_DEPTH", "basic"),
            cache_ttl=int(os.getenv("TAVILY_CACHE_TTL", "3600")),
            monthly_budget=float(os.getenv("TAVILY_MONTHLY_BUDGET", "100.0")),
        )

12.2 带重试的生产级客户端

import time
import logging
from typing import Optional
from tavily import TavilyClient

logger = logging.getLogger(__name__)

class ProductionTavilyClient:
    """生产级Tavily客户端"""

    def __init__(self, config: TavilyConfig):
        self.config = config
        self.client = TavilyClient(api_key=config.api_key)
        self.cache = TavilyCache(ttl=config.cache_ttl)
        self.cost_tracker = CostTracker(monthly_budget=config.monthly_budget)

    def search(self, query: str, **kwargs) -> Optional[dict]:
        """带重试和缓存的搜索"""
        # 检查预算
        month_key = time.strftime("%Y-%m")
        monthly_cost = self.cost_tracker.usage["by_month"].get(
            month_key, {"cost": 0.0}
        )["cost"]
        if monthly_cost >= self.config.monthly_budget:
            logger.warning("月度预算已耗尽，跳过搜索")
            return None

        # 检查缓存
        params = {"search_depth": self.config.search_depth, **kwargs}
        cached = self.cache.get(query, params)
        if cached:
            logger.info(f"缓存命中：{query[:50]}")
            return cached

        # 带重试的API调用
        for attempt in range(self.config.retry_attempts):
            try:
                result = self.client.search(
                    query=query,
                    search_depth=kwargs.get("search_depth", self.config.search_depth),
                    max_results=kwargs.get("max_results", self.config.max_results),
                    **{k: v for k, v in kwargs.items()
                       if k not in ("search_depth", "max_results")},
                )

                # 缓存结果
                self.cache.set(query, params, result)

                # 记录成本
                self.cost_tracker.record_call(
                    kwargs.get("search_depth", self.config.search_depth)
                )

                return result

            except Exception as e:
                logger.warning(
                    f"搜索失败（第{attempt + 1}次）：{e}"
                )
                if attempt < self.config.retry_attempts - 1:
                    time.sleep(self.config.retry_delay * (attempt + 1))

        logger.error(f"搜索最终失败：{query[:50]}")
        return None

12.3 Docker部署

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# 创建非root用户
RUN useradd -m appuser
USER appuser

# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
    CMD python -c "from tavily import TavilyClient; print('ok')"

CMD ["python", "main.py"]

# docker-compose.yml
version: "3.8"

services:
  search-service:
    build: .
    environment:
      - TAVILY_API_KEY=${TAVILY_API_KEY}
      - TAVILY_MONTHLY_BUDGET=100.0
      - TAVILY_CACHE_TTL=3600
      - TAVILY_MAX_RESULTS=5
    volumes:
      - cache-data:/app/.tavily_cache
      - usage-data:/app/.tavily_usage.json
    restart: unless-stopped

volumes:
  cache-data:
  usage-data:

12.4 监控与告警

import time
from dataclasses import dataclass, field
from collections import defaultdict

@dataclass
class SearchMetrics:
    """搜索指标监控"""
    total_calls: int = 0
    failed_calls: int = 0
    cache_hits: int = 0
    total_latency: float = 0.0
    latencies: list[float] = field(default_factory=list)

    def record(self, success: bool, latency: float, from_cache: bool):
        self.total_calls += 1
        if not success:
            self.failed_calls += 1
        if from_cache:
            self.cache_hits += 1
        self.total_latency += latency
        self.latencies.append(latency)

    @property
    def success_rate(self) -> float:
        if self.total_calls == 0:
            return 1.0
        return (self.total_calls - self.failed_calls) / self.total_calls

    @property
    def avg_latency(self) -> float:
        if not self.latencies:
            return 0.0
        return sum(self.latencies) / len(self.latencies)

    @property
    def p95_latency(self) -> float:
        if not self.latencies:
            return 0.0
        sorted_lat = sorted(self.latencies)
        idx = int(len(sorted_lat) * 0.95)
        return sorted_lat[min(idx, len(sorted_lat) - 1)]

    def check_alerts(self) -> list[str]:
        """检查告警条件"""
        alerts = []

        if self.success_rate < 0.95:
            alerts.append(
                f"🚨 成功率过低：{self.success_rate:.1%}"
            )

        if self.p95_latency > 5.0:
            alerts.append(
                f"⚠️ P95延迟过高：{self.p95_latency:.1f}s"
            )

        cache_rate = self.cache_hits / max(self.total_calls, 1)
        if cache_rate > 0.8:
            alerts.append(
                f"ℹ️ 缓存命中率高：{cache_rate:.1%}，考虑增加缓存TTL"
            )

        return alerts

12.5 安全最佳实践

生产环境安全清单：

1. API Key管理
   ✅ 使用环境变量存储，不要硬编码
   ✅ 使用密钥管理服务（AWS Secrets Manager, HashiCorp Vault）
   ✅ 定期轮换API Key
   ✅ 限制API Key的IP白名单（如支持）

2. 输入验证
   ✅ 对用户输入的查询进行长度限制
   ✅ 过滤敏感词和注入尝试
   ✅ 限制搜索频率（rate limiting）

3. 输出安全
   ✅ 对搜索结果进行XSS过滤
   ✅ 验证URL的合法性
   ✅ 不信任搜索结果中的可执行内容

4. 日志与审计
   ✅ 记录所有API调用日志
   ✅ 监控异常调用模式
   ✅ 定期审计成本和使用量

13. 高级实战技巧

13.1 搜索结果质量评估

在生产环境中，对搜索结果进行质量评估至关重要。低质量的搜索结果会直接影响下游LLM的输出质量，导致"垃圾进、垃圾出"的问题。

class SearchQualityEvaluator:
    """搜索结果质量评估器"""

    def __init__(self):
        self.metrics = {
            "total_queries": 0,
            "empty_results": 0,
            "low_score_results": 0,
            "avg_score": 0.0,
        }

    def evaluate(self, query: str, results: dict) -> dict:
        """评估单次搜索结果的质量"""
        items = results.get("results", [])
        scores = [r["score"] for r in items]

        evaluation = {
            "query": query,
            "result_count": len(items),
            "has_answer": bool(results.get("answer")),
            "avg_score": sum(scores) / len(scores) if scores else 0,
            "min_score": min(scores) if scores else 0,
            "max_score": max(scores) if scores else 0,
            "score_variance": self._variance(scores),
            "quality_grade": self._grade(scores, results.get("answer")),
        }

        # 更新全局指标
        self.metrics["total_queries"] += 1
        if not items:
            self.metrics["empty_results"] += 1
        elif evaluation["avg_score"] < 0.5:
            self.metrics["low_score_results"] += 1

        return evaluation

    def _variance(self, scores: list[float]) -> float:
        if len(scores) < 2:
            return 0.0
        mean = sum(scores) / len(scores)
        return sum((s - mean) ** 2 for s in scores) / len(scores)

    def _grade(self, scores: list[float], answer: str | None) -> str:
        if not scores:
            return "F"
        avg = sum(scores) / len(scores)
        if avg >= 0.8 and answer:
            return "A"
        elif avg >= 0.6:
            return "B"
        elif avg >= 0.4:
            return "C"
        return "D"

13.2 多源搜索融合

实际项目中，仅依赖单一搜索源往往不够。将Tavily与其他信息源结合，可以获得更全面的搜索结果。

class MultiSourceSearcher:
    """多源搜索融合器"""

    def __init__(self, tavily_key: str):
        self.tavily = TavilyClient(api_key=tavily_key)

    def search(self, query: str) -> list[dict]:
        """融合多个来源的搜索结果"""
        all_results = []

        # Tavily通用搜索
        tavily_results = self.tavily.search(
            query=query, max_results=5, search_depth="advanced"
        )
        for r in tavily_results.get("results", []):
            r["source"] = "tavily"
            all_results.append(r)

        # Tavily新闻搜索（补充时效性信息）
        news_results = self.tavily.search(
            query=query, topic="news", days=7, max_results=3
        )
        for r in news_results.get("results", []):
            r["source"] = "tavily_news"
            all_results.append(r)

        # 去重并按分数排序
        seen_urls = set()
        unique_results = []
        for r in sorted(all_results, key=lambda x: x["score"], reverse=True):
            if r["url"] not in seen_urls:
                seen_urls.add(r["url"])
                unique_results.append(r)

        return unique_results[:10]

13.3 搜索结果持久化与分析

对于需要分析搜索趋势的企业场景，持久化搜索结果并进行长期分析非常有价值。

import json
import sqlite3
from datetime import datetime

class SearchHistoryStore:
    """搜索历史存储"""

    def __init__(self, db_path: str = "search_history.db"):
        self.conn = sqlite3.connect(db_path)
        self._init_db()

    def _init_db(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS searches (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                query TEXT NOT NULL,
                results_json TEXT NOT NULL,
                result_count INTEGER,
                avg_score REAL,
                has_answer BOOLEAN,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        self.conn.execute("""
            CREATE INDEX IF NOT EXISTS idx_created_at
            ON searches(created_at)
        """)
        self.conn.commit()

    def save(self, query: str, results: dict):
        """保存搜索结果"""
        items = results.get("results", [])
        avg_score = (
            sum(r["score"] for r in items) / len(items) if items else 0
        )

        self.conn.execute(
            """INSERT INTO searches
            (query, results_json, result_count, avg_score, has_answer)
            VALUES (?, ?, ?, ?, ?)""",
            (
                query,
                json.dumps(results, ensure_ascii=False),
                len(items),
                avg_score,
                bool(results.get("answer")),
            ),
        )
        self.conn.commit()

    def get_popular_queries(self, days: int = 30, limit: int = 10) -> list:
        """获取热门查询"""
        cursor = self.conn.execute(
            """SELECT query, COUNT(*) as cnt, AVG(avg_score) as score
            FROM searches
            WHERE created_at > datetime('now', ?)
            GROUP BY query
            ORDER BY cnt DESC
            LIMIT ?""",
            (f"-{days} days", limit),
        )
        return cursor.fetchall()

    def get_quality_trend(self, days: int = 30) -> list:
        """获取搜索质量趋势"""
        cursor = self.conn.execute(
            """SELECT DATE(created_at) as day,
            AVG(avg_score) as score,
            COUNT(*) as count
            FROM searches
            WHERE created_at > datetime('now', ?)
            GROUP BY DATE(created_at)
            ORDER BY day""",
            (f"-{days} days",),
        )
        return cursor.fetchall()

13.4 异步批量搜索

在需要处理大量查询的场景中，异步批量搜索可以显著提升吞吐量。

import asyncio
from dataclasses import dataclass

@dataclass
class BatchSearchTask:
    """批量搜索任务"""
    id: str
    query: str
    params: dict

class BatchSearchProcessor:
    """批量搜索处理器"""

    def __init__(self, api_key: str, max_concurrent: int = 5):
        self.client = AsyncTavilyClient(api_key=api_key)
        self.semaphore = asyncio.Semaphore(max_concurrent)

    async def _search_one(self, task: BatchSearchTask) -> dict:
        """执行单个搜索（带并发控制）"""
        async with self.semaphore:
            try:
                result = await self.client.search(
                    query=task.query, **task.params
                )
                return {"task_id": task.id, "success": True, "data": result}
            except Exception as e:
                return {"task_id": task.id, "success": False, "error": str(e)}

    async def process_batch(self, tasks: list[BatchSearchTask]) -> list[dict]:
        """批量处理搜索任务"""
        print(f"开始批量搜索，共 {len(tasks)} 个任务")
        results = await asyncio.gather(
            *[self._search_one(t) for t in tasks]
        )
        success = sum(1 for r in results if r["success"])
        print(f"批量搜索完成：{success}/{len(tasks)} 成功")
        return results

# 使用示例
async def main():
    tasks = [
        BatchSearchTask(id=f"q{i}", query=q, params={"max_results": 3})
        for i, q in enumerate([
            "Python异步编程", "Rust系统编程", "Go微服务架构",
            "TypeScript类型系统", "Kubernetes最佳实践",
        ])
    ]

    processor = BatchSearchProcessor("tvly-YOUR_KEY", max_concurrent=3)
    results = await processor.process_batch(tasks)

    for r in results:
        if r["success"]:
            print(f"[{r['task_id']}] 找到 {len(r['data'].get('results', []))} 条结果")

asyncio.run(main())

14. 测试策略

14.1 单元测试

import pytest
from unittest.mock import patch, MagicMock

class TestSearchResultProcessor:
    """搜索结果处理器测试"""

    def setup_method(self):
        self.processor = SearchResultProcessor(min_score=0.5)

    def test_filter_low_scores(self):
        """测试过滤低分结果"""
        raw = {
            "results": [
                {"title": "A", "url": "http://a.com", "content": "test", "score": 0.9},
                {"title": "B", "url": "http://b.com", "content": "test", "score": 0.3},
                {"title": "C", "url": "http://c.com", "content": "test", "score": 0.7},
            ]
        }
        results = self.processor.process(raw)
        assert len(results) == 2
        assert all(r.score >= 0.5 for r in results)

    def test_sort_by_score(self):
        """测试按分数排序"""
        raw = {
            "results": [
                {"title": "Low", "url": "http://l.com", "content": "test", "score": 0.6},
                {"title": "High", "url": "http://h.com", "content": "test", "score": 0.95},
            ]
        }
        results = self.processor.process(raw)
        assert results[0].score > results[1].score

    def test_relevance_classification(self):
        """测试相关性分类"""
        assert self.processor._classify_relevance(0.9) == "high"
        assert self.processor._classify_relevance(0.7) == "medium"
        assert self.processor._classify_relevance(0.4) == "low"

    def test_empty_results(self):
        """测试空结果处理"""
        raw = {"results": []}
        results = self.processor.process(raw)
        assert len(results) == 0

        formatted = self.processor.format_for_llm(results)
        assert "未找到" in formatted

class TestTavilyCache:
    """缓存测试"""

    def setup_method(self, tmp_path):
        self.cache = TavilyCache(cache_dir=str(tmp_path), ttl=3600)

    def test_cache_hit(self):
        """测试缓存命中"""
        query = "test query"
        params = {"max_results": 5}
        result = {"results": [{"title": "test"}]}

        self.cache.set(query, params, result)
        cached = self.cache.get(query, params)
        assert cached is not None
        assert cached["results"][0]["title"] == "test"

    def test_cache_miss(self):
        """测试缓存未命中"""
        cached = self.cache.get("nonexistent", {})
        assert cached is None

14.2 集成测试

@pytest.mark.integration
class TestTavilyIntegration:
    """Tavily API集成测试（需要真实API Key）"""

    @pytest.fixture
    def client(self):
        import os
        key = os.getenv("TAVILY_API_KEY")
        if not key:
            pytest.skip("需要设置 TAVILY_API_KEY 环境变量")
        return TavilyClient(api_key=key)

    def test_basic_search(self, client):
        """测试基础搜索功能"""
        result = client.search("Python programming")
        assert "results" in result
        assert len(result["results"]) > 0
        assert all("title" in r for r in result["results"])

    def test_search_with_answer(self, client):
        """测试带AI摘要的搜索"""
        result = client.search(
            "什么是机器学习",
            include_answer=True,
        )
        assert result.get("answer") is not None
        assert len(result["answer"]) > 0

    def test_news_search(self, client):
        """测试新闻搜索"""
        result = client.search(
            "AI news",
            topic="news",
            days=7,
        )
        assert "results" in result

    def test_domain_filter(self, client):
        """测试域名过滤"""
        result = client.search(
            "Python tutorial",
            include_domains=["python.org"],
            max_results=3,
        )
        for r in result.get("results", []):
            assert "python.org" in r["url"]

总结

Tavily Search API为AI应用提供了开箱即用的实时搜索能力。关键要点：

1. 专为AI设计：结构化返回、AI摘要、相关性评分，省去大量后处理工作
2. 生态完善：LangChain/LlamaIndex原生集成，快速构建RAG和Agent
3. 成本可控：免费额度充足，付费价格透明，缓存策略灵活
4. 生产就绪：通过缓存、重试、监控等机制，可直接用于企业级应用

快速上手路径：

1. 注册 → 获取API Key（5分钟）
2. pip install tavily-python（1分钟）
3. 3行代码完成第一次搜索（2分钟）
4. 集成到LangChain/LlamaIndex（15分钟）
5. 添加缓存和监控（30分钟）
6. Docker部署上线（1小时）

从原型到生产，Tavily让AI应用的搜索能力不再是一个难题。