AI搜索引擎与Perplexity实战完全教程

从技术原理到生产实践，全面掌握AI搜索引擎的使用与开发

目录

1. AI搜索引擎概述
2. 核心技术原理：RAG + 实时检索
3. Perplexity AI 深度解析
4. SearchGPT / OpenAI Search 实战
5. 国产AI搜索方案：秘塔、天工等
6. 自建AI搜索引擎：SearXNG + LLM
7. API集成与开发实战
8. AI搜索优化（ASO）策略
9. 企业级AI搜索应用场景
10. 最佳实践与总结

1. AI搜索引擎概述

1.1 从传统搜索到AI搜索的范式转变

传统搜索引擎（Google、Baidu）的工作模式是"关键词匹配 → 链接列表 → 用户自行筛选"。用户需要：

• 构造合适的关键词
• 浏览多个搜索结果
• 自行阅读、对比、总结

AI搜索引擎的核心革新在于：用自然语言提问，直接获得结构化答案。

1.2 AI搜索引擎的核心价值

用户提问 → AI理解意图 → 实时检索网络 → 大模型综合生成 → 带引用的结构化答案

这个流程解决了三个核心痛点：

1. 信息过载：AI代替用户阅读和筛选
2. 知识时效：实时检索突破模型训练截止日期
3. 理解门槛：自然语言降低使用成本

2. 核心技术原理：RAG + 实时检索

2.1 RAG（Retrieval-Augmented Generation）架构

RAG是AI搜索引擎的技术基石，由Meta在2020年提出。其核心思想是：在生成答案之前，先从外部知识库检索相关信息，作为生成的上下文。

┌─────────────────────────────────────────────────┐
│                 AI搜索引擎架构                     │
│                                                   │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐   │
│  │ 用户查询  │───→│ 查询理解  │───→│ 检索引擎  │   │
│  └──────────┘    └──────────┘    └────┬─────┘   │
│                                       │          │
│                                       ▼          │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐   │
│  │ 最终答案  │←───│ 大模型    │←───│ 上下文构建│   │
│  │ + 引用    │    │ 生成答案  │    │ (Prompt)  │   │
│  └──────────┘    └──────────┘    └──────────┘   │
│                                                   │
└─────────────────────────────────────────────────┘

2.2 实时检索流程详解

第一步：查询理解与改写

原始查询往往是口语化的，需要转换为更适合检索的形式：

# 查询改写示例
original_query = "2024年最新的AI手机有哪些，哪个值得买？"

# 改写后的多个检索查询
rewritten_queries = [
    "2024 AI smartphone best models comparison",
    "2024年人工智能手机推荐 旗舰",
    "Galaxy S24 Pixel 8 iPhone 15 AI features review"
]

第二步：多源并行检索

import asyncio
import aiohttp

async def multi_source_search(query: str) -> dict:
    """并行检索多个来源"""
    tasks = [
        search_web(query),          # 网页搜索
        search_news(query),         # 新闻搜索
        search_academic(query),     # 学术搜索
    ]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    return {
        "web": results[0] if not isinstance(results[0], Exception) else [],
        "news": results[1] if not isinstance(results[1], Exception) else [],
        "academic": results[2] if not isinstance(results[2], Exception) else [],
    }

第三步：结果排序与筛选

def rank_and_filter(results: list, query: str, top_k: int = 10) -> list:
    """对检索结果进行相关性排序和去重"""
    # 1. 计算语义相似度
    query_embedding = encode(query)
    for r in results:
        r["score"] = cosine_similarity(query_embedding, encode(r["snippet"]))
    
    # 2. 时间衰减（新闻类内容优先新鲜的）
    for r in results:
        if r.get("is_news"):
            days_old = (datetime.now() - r["published"]).days
            r["score"] *= max(0.5, 1 - days_old / 30)
    
    # 3. 来源可信度加权
    trusted_domains = {"nature.com", "arxiv.org", "reuters.com", ...}
    for r in results:
        if any(domain in r["url"] for domain in trusted_domains):
            r["score"] *= 1.2
    
    # 4. 去重并返回Top K
    seen_urls = set()
    filtered = []
    for r in sorted(results, key=lambda x: x["score"], reverse=True):
        if r["url"] not in seen_urls:
            seen_urls.add(r["url"])
            filtered.append(r)
    return filtered[:top_k]

第四步：上下文构建与生成

def build_rag_prompt(query: str, search_results: list) -> str:
    """构建RAG提示词"""
    context_parts = []
    for i, r in enumerate(search_results, 1):
        context_parts.append(
            f"[{i}] {r['title']}\n"
            f"来源: {r['url']}\n"
            f"内容: {r['snippet']}\n"
        )
    
    context = "\n---\n".join(context_parts)
    
    return f"""基于以下检索到的参考资料回答用户问题。
要求：
- 综合多个来源给出全面回答
- 在答案中使用 [1][2] 等标注引用来源
- 如果资料不足以回答，明确说明
- 保持客观，注明不同来源的观点差异

参考资料：
{context}

用户问题：{query}

请给出详细的回答："""

2.3 向量检索与语义理解

现代AI搜索引擎大量使用向量嵌入来实现语义级别的检索：

from sentence_transformers import SentenceTransformer
import numpy as np

# 加载嵌入模型
model = SentenceTransformer('BAAI/bge-base-zh-v1.5')

def semantic_search(query: str, documents: list, top_k: int = 5):
    """语义搜索实现"""
    # 编码查询和文档
    query_vec = model.encode(query)
    doc_vecs = model.encode(documents)
    
    # 计算余弦相似度
    scores = np.dot(doc_vecs, query_vec) / (
        np.linalg.norm(doc_vecs, axis=1) * np.linalg.norm(query_vec)
    )
    
    # 返回最相关的文档
    top_indices = np.argsort(scores)[::-1][:top_k]
    return [(documents[i], scores[i]) for i in top_indices]

# 示例：语义理解超越关键词匹配
docs = [
    "苹果公司发布了最新款iPhone手机",
    "今天的红富士苹果又大又甜",
    "Apple announced the new MacBook Pro",
]

# 查询"苹果新品"能正确匹配到第1条而非第2条
results = semantic_search("苹果新品", docs)

3. Perplexity AI 深度解析

3.1 产品定位与核心能力

Perplexity AI 是目前最成功的AI搜索引擎产品，成立于2022年，由前Google和Meta的AI研究员创立。其核心理念是"Answer Engine"——不是搜索引擎，而是答案引擎。

3.2 三种搜索模式

Quick Search（快速搜索）

适合简单问题，响应速度快（通常2-5秒），使用较轻量的模型。

适用场景：
- "今天北京天气如何？"
- "Python 3.12 有什么新特性？"
- "某某公司的创始人是谁？"

Pro Search（专业搜索）

Perplexity的核心差异化功能。它会：

1. 将问题分解为多个子问题
2. 执行多轮检索
3. 使用更强的推理模型
4. 生成更深入、更全面的回答

用户提问："比较React和Vue在2024年的生态系统发展"

Pro Search 内部流程：
├── 子问题1: "React 2024 新特性 React Server Components 进展"
├── 子问题2: "Vue 2024 Vue 3.4 Vapor Mode 发展"
├── 子问题3: "React vs Vue 2024 developer survey npm downloads"
├── 子问题4: "React Vue 企业采用率 2024"
└── 综合生成：结构化对比报告

Focus 模式

限定搜索范围，提供更精准的结果：

3.3 Perplexity API 实战

import requests

def perplexity_search(query: str, model: str = "sonar-pro") -> dict:
    """使用Perplexity API进行AI搜索"""
    url = "https://api.perplexity.ai/chat/completions"
    
    headers = {
        "Authorization": "Bearer YOUR_API_KEY",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": model,
        "messages": [
            {
                "role": "system",
                "content": "Be precise and cite your sources."
            },
            {
                "role": "user",
                "content": query
            }
        ],
        "return_citations": True,
        "search_recency_filter": "month",  # 限制为最近一个月
    }
    
    response = requests.post(url, json=payload, headers=headers)
    result = response.json()
    
    answer = result["choices"][0]["message"]["content"]
    citations = result.get("citations", [])
    
    return {
        "answer": answer,
        "citations": citations,
        "usage": result["usage"]
    }

# 使用示例
result = perplexity_search("2024年全球AI芯片市场规模和主要玩家")
print(result["answer"])
print("\n引用来源:")
for c in result["citations"]:
    print(f"  - {c}")

3.4 Perplexity Pages：搜索到内容创作

Perplexity Pages是一个将搜索结果转化为长文内容的功能：

# 构建类似Perplexity Pages的工作流
def generate_research_article(topic: str) -> str:
    """基于多轮搜索生成研究报告"""
    
    # 1. 第一轮：概览搜索
    overview = perplexity_search(f"{topic} 概述 2024年发展现状")
    
    # 2. 第二轮：深度子话题
    subtopics = extract_subtopics(overview["answer"])
    deep_research = []
    for sub in subtopics[:5]:
        result = perplexity_search(f"{topic} {sub} 深度分析")
        deep_research.append(result)
    
    # 3. 第三轮：数据与案例
    data = perplexity_search(f"{topic} 数据统计 市场规模 案例研究")
    
    # 4. 综合生成长文
    article = generate_article(
        overview=overview,
        deep_research=deep_research,
        data=data,
        topic=topic
    )
    
    return article

4. SearchGPT / OpenAI Search 实战

4.1 SearchGPT 概述

OpenAI在2024年推出了SearchGPT（后整合进ChatGPT的搜索功能），其核心特点：

• 深度整合GPT-4的推理能力
• 实时网络搜索
• 对话式搜索体验
• 与ChatGPT生态无缝衔接

4.2 ChatGPT Search 使用技巧

高效提问模式：

❌ 低效: "AI新闻"
✅ 高效: "总结过去一周AI领域最重要的3个突破，附带原始论文链接"

❌ 低效: "怎么学编程"
✅ 高效: "对比2024年最受欢迎的3个Python在线课程平台，包括价格和项目实战内容"

高级技巧：
1. 指定时间范围："最近一个月内..."
2. 要求引用："附上来源链接"
3. 指定格式："用表格对比"
4. 限定领域："从学术角度..."

4.3 OpenAI Responses API with Web Search

OpenAI提供了带搜索功能的API：

from openai import OpenAI

client = OpenAI()

def chatgpt_search(query: str) -> dict:
    """使用OpenAI Responses API进行搜索"""
    response = client.responses.create(
        model="gpt-4o",
        tools=[{"type": "web_search_preview"}],
        input=query
    )
    
    # 提取搜索结果和引用
    result = {
        "answer": response.output_text,
        "sources": []
    }
    
    # 解析输出中的引用
    for item in response.output:
        if item.type == "web_search_call":
            result["search_query"] = item.action.get("query", "")
        elif item.type == "message":
            for content in item.content:
                if hasattr(content, "annotations"):
                    for ann in content.annotations:
                        if ann.type == "url_citation":
                            result["sources"].append({
                                "title": ann.title,
                                "url": ann.url
                            })
    
    return result

# 使用示例
result = chatgpt_search("2024年诺贝尔物理学奖得主及其贡献")
print(result["answer"])

5. 国产AI搜索方案：秘塔、天工等

5.1 秘塔AI搜索

秘塔AI搜索是国内最成熟的AI搜索引擎之一，特点包括：

• 无广告：干净的搜索体验
• 学术搜索：专门的学术论文检索模式
• 思维导图：自动将搜索结果结构化
• 大纲模式：生成结构化的大纲式回答

秘塔AI搜索的优势场景：
1. 中文内容检索质量高
2. 学术论文搜索专业
3. 法律法规搜索（秘塔法律背景）
4. 无广告干扰的纯净体验

使用入口：metaso.cn

5.2 天工AI搜索

天工AI搜索由昆仑万维推出，特点：

• 支持多模态搜索（文字、图片）
• 深度整合天工大模型
• 提供"深入回答"模式
• 支持追问和多轮对话

5.3 其他国产方案对比

5.4 国产AI搜索API调用示例

# 以通用的国产AI搜索API模式为例
import requests

class ChineseAISearch:
    """国产AI搜索统一接口封装"""
    
    def __init__(self, provider: str, api_key: str):
        self.provider = provider
        self.api_key = api_key
        self.endpoints = {
            "kimi": "https://api.moonshot.cn/v1/chat/completions",
            "doubao": "https://ark.cn-beijing.volces.com/api/v3/chat/completions",
        }
    
    def search(self, query: str, search_enabled: bool = True) -> str:
        """执行AI搜索"""
        if self.provider == "kimi":
            return self._kimi_search(query)
        elif self.provider == "doubao":
            return self._doubao_search(query)
    
    def _kimi_search(self, query: str) -> str:
        """Kimi联网搜索"""
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": "moonshot-v1-128k",
            "messages": [
                {
                    "role": "system",
                    "content": "你是Kimi，一个能够联网搜索的AI助手。"
                              "请搜索最新信息来回答用户问题，并标注来源。"
                },
                {"role": "user", "content": query}
            ],
            "tools": [{
                "type": "builtin_function",
                "function": {"name": "$web_search"}
            }]
        }
        
        resp = requests.post(
            self.endpoints["kimi"],
            json=payload,
            headers=headers
        )
        return resp.json()["choices"][0]["message"]["content"]

# 使用
searcher = ChineseAISearch("kimi", "your-api-key")
result = searcher.search("2024年中国新能源汽车出口数据")
print(result)

6. 自建AI搜索引擎：SearXNG + LLM

6.1 为什么自建？

• 隐私保护：搜索数据不经过第三方
• 定制化：可针对特定领域优化
• 成本控制：长期使用成本更低
• 数据主权：完全掌控搜索数据

6.2 架构设计

┌─────────────────────────────────────────────────────┐
│              自建AI搜索引擎架构                        │
│                                                       │
│  用户查询                                             │
│    │                                                  │
│    ▼                                                  │
│  ┌─────────────────┐                                 │
│  │  Query Processor │  查询理解、改写、扩展             │
│  │  (LLM-powered)  │                                 │
│  └────────┬────────┘                                 │
│           │                                           │
│           ▼                                           │
│  ┌─────────────────┐                                 │
│  │    SearXNG       │  元搜索引擎，聚合多个搜索源       │
│  │  (Self-hosted)  │  Google, Bing, DuckDuckGo...    │
│  └────────┬────────┘                                 │
│           │                                           │
│           ▼                                           │
│  ┌─────────────────┐                                 │
│  │ Result Processor │  去重、排序、过滤、摘要           │
│  └────────┬────────┘                                 │
│           │                                           │
│           ▼                                           │
│  ┌─────────────────┐                                 │
│  │   Answer Engine  │  LLM综合生成答案                 │
│  │  (Local LLM /   │  带引用、可追溯                   │
│  │   API LLM)      │                                 │
│  └────────┬────────┘                                 │
│           │                                           │
│           ▼                                           │
│       最终答案 + 引用                                  │
└─────────────────────────────────────────────────────┘

6.3 SearXNG 部署

# 使用Docker部署SearXNG
docker run -d \
  --name searxng \
  -p 8080:8080 \
  -v ./searxng:/etc/searxng \
  -e SEARXNG_BASE_URL=http://localhost:8080/ \
  searxng/searxng:latest

# 验证部署
curl "http://localhost:8080/search?q=test&format=json"

SearXNG配置文件 settings.yml：

# searxng/settings.yml
use_default_settings: true

general:
  instance_name: "My AI Search"
  
search:
  safe_search: 0
  autocomplete: "google"
  default_lang: "zh-CN"

server:
  secret_key: "your-secret-key-here"
  limiter: false
  
engines:
  - name: google
    engine: google
    shortcut: g
    disabled: false
    
  - name: bing
    engine: bing
    shortcut: b
    disabled: false
    
  - name: duckduckgo
    engine: duckduckgo
    shortcut: ddg
    disabled: false
    
  - name: wikipedia
    engine: wikipedia
    shortcut: w
    disabled: false
    
  - name: arxiv
    engine: arxiv
    shortcut: ar
    disabled: false

6.4 完整的自建AI搜索引擎代码

"""
自建AI搜索引擎：SearXNG + LLM
完整实现一个类似Perplexity的AI搜索引擎
"""

import asyncio
import aiohttp
import json
from dataclasses import dataclass
from typing import Optional
from openai import OpenAI

@dataclass
class SearchResult:
    title: str
    url: str
    snippet: str
    engine: str
    score: float = 0.0

@dataclass
class AIAnswer:
    answer: str
    sources: list[SearchResult]
    sub_queries: list[str]
    model: str

class AISearchEngine:
    """自建AI搜索引擎"""
    
    def __init__(
        self,
        searxng_url: str = "http://localhost:8080",
        llm_api_key: str = "",
        llm_base_url: str = "https://api.openai.com/v1",
        llm_model: str = "gpt-4o-mini",
    ):
        self.searxng_url = searxng_url
        self.client = OpenAI(api_key=llm_api_key, base_url=llm_base_url)
        self.llm_model = llm_model
    
    async def search(self, query: str, max_results: int = 10) -> AIAnswer:
        """主搜索入口"""
        
        # Step 1: 查询理解与改写
        sub_queries = await self._rewrite_query(query)
        
        # Step 2: 并行检索
        all_results = []
        async with aiohttp.ClientSession() as session:
            tasks = [
                self._searxng_search(session, q, max_results=max_results)
                for q in [query] + sub_queries
            ]
            results_per_query = await asyncio.gather(*tasks)
            for results in results_per_query:
                all_results.extend(results)
        
        # Step 3: 去重与排序
        ranked_results = self._rank_results(all_results, query)
        
        # Step 4: 生成答案
        answer = await self._generate_answer(query, ranked_results)
        
        return AIAnswer(
            answer=answer,
            sources=ranked_results[:max_results],
            sub_queries=sub_queries,
            model=self.llm_model,
        )
    
    async def _rewrite_query(self, query: str) -> list[str]:
        """使用LLM改写查询，生成多个子问题"""
        response = self.client.chat.completions.create(
            model=self.llm_model,
            messages=[
                {
                    "role": "system",
                    "content": "你是一个搜索查询优化助手。"
                              "将用户问题分解为2-4个更精确的搜索子查询。"
                              "只返回JSON数组，不要其他内容。"
                },
                {"role": "user", "content": f"原始问题：{query}"}
            ],
            response_format={"type": "json_object"},
            temperature=0.3,
        )
        
        try:
            result = json.loads(response.choices[0].message.content)
            if isinstance(result, list):
                return result
            return result.get("queries", [])
        except json.JSONDecodeError:
            return []
    
    async def _searxng_search(
        self,
        session: aiohttp.ClientSession,
        query: str,
        max_results: int = 10
    ) -> list[SearchResult]:
        """调用SearXNG搜索"""
        params = {
            "q": query,
            "format": "json",
            "language": "zh-CN",
            "pageno": 1,
        }
        
        try:
            async with session.get(
                f"{self.searxng_url}/search",
                params=params,
                timeout=aiohttp.ClientTimeout(total=10)
            ) as resp:
                data = await resp.json()
                
                results = []
                for item in data.get("results", [])[:max_results]:
                    results.append(SearchResult(
                        title=item.get("title", ""),
                        url=item.get("url", ""),
                        snippet=item.get("content", ""),
                        engine=item.get("engine", "unknown"),
                    ))
                return results
        except Exception as e:
            print(f"Search error: {e}")
            return []
    
    def _rank_results(
        self,
        results: list[SearchResult],
        query: str
    ) -> list[SearchResult]:
        """对结果去重并排序"""
        seen_urls = set()
        unique = []
        
        for r in results:
            # 基础URL去重（去除尾部斜杠等）
            normalized_url = r.url.rstrip("/").lower()
            if normalized_url not in seen_urls:
                seen_urls.add(normalized_url)
                unique.append(r)
        
        # 简单的相关性评分（可替换为向量相似度）
        query_words = set(query.lower().split())
        for r in unique:
            title_words = set(r.title.lower().split())
            snippet_words = set(r.snippet.lower().split())
            overlap = len(query_words & (title_words | snippet_words))
            r.score = overlap / max(len(query_words), 1)
        
        unique.sort(key=lambda x: x.score, reverse=True)
        return unique
    
    async def _generate_answer(
        self,
        query: str,
        sources: list[SearchResult]
    ) -> str:
        """基于检索结果生成答案"""
        context_parts = []
        for i, s in enumerate(sources[:8], 1):
            context_parts.append(
                f"[{i}] {s.title}\n"
                f"URL: {s.url}\n"
                f"摘要: {s.snippet}\n"
            )
        
        context = "\n---\n".join(context_parts)
        
        response = self.client.chat.completions.create(
            model=self.llm_model,
            messages=[
                {
                    "role": "system",
                    "content": "你是一个专业的AI搜索引擎助手。"
                              "基于提供的参考资料回答用户问题。\n"
                              "要求：\n"
                              "1. 综合多个来源给出全面、准确的回答\n"
                              "2. 使用[1][2]等标注引用来源\n"
                              "3. 如果信息不足，明确说明\n"
                              "4. 保持客观中立"
                },
                {
                    "role": "user",
                    "content": f"参考资料：\n{context}\n\n问题：{query}"
                }
            ],
            temperature=0.3,
            max_tokens=2000,
        )
        
        return response.choices[0].message.content

# 使用示例
async def main():
    engine = AISearchEngine(
        searxng_url="http://localhost:8080",
        llm_api_key="your-api-key",
        llm_model="gpt-4o-mini",
    )
    
    result = await engine.search("2024年最值得关注的AI技术趋势")
    
    print("=== AI搜索结果 ===\n")
    print(result.answer)
    print(f"\n=== 引用来源 ({len(result.sources)}) ===")
    for i, s in enumerate(result.sources, 1):
        print(f"[{i}] {s.title}")
        print(f"    {s.url}")

if __name__ == "__main__":
    asyncio.run(main())

7. API集成与开发实战

7.1 构建统一的AI搜索接口

在实际项目中，我们往往需要聚合多个AI搜索服务：

"""
统一AI搜索接口：聚合多个搜索引擎
"""

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
import asyncio

class SearchProvider(Enum):
    PERPLEXITY = "perplexity"
    OPENAI = "openai"
    KIMI = "kimi"
    SERARXNG_LLM = "searxng_llm"

@dataclass
class UnifiedSearchResult:
    answer: str
    sources: list[dict]
    provider: str
    latency_ms: int
    tokens_used: int = 0

class BaseAISearchProvider(ABC):
    """AI搜索提供者基类"""
    
    @abstractmethod
    async def search(
        self,
        query: str,
        max_results: int = 10,
        language: str = "zh-CN",
    ) -> UnifiedSearchResult:
        pass

class PerplexityProvider(BaseAISearchProvider):
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.perplexity.ai"
    
    async def search(self, query: str, **kwargs) -> UnifiedSearchResult:
        import time
        start = time.time()
        
        import aiohttp
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.base_url}/chat/completions",
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json",
                },
                json={
                    "model": "sonar-pro",
                    "messages": [
                        {"role": "user", "content": query}
                    ],
                    "return_citations": True,
                }
            ) as resp:
                data = await resp.json()
        
        answer = data["choices"][0]["message"]["content"]
        citations = data.get("citations", [])
        
        return UnifiedSearchResult(
            answer=answer,
            sources=[{"url": c} for c in citations],
            provider="perplexity",
            latency_ms=int((time.time() - start) * 1000),
            tokens_used=data.get("usage", {}).get("total_tokens", 0),
        )

class SearchAggregator:
    """搜索聚合器：支持多源并发、结果融合"""
    
    def __init__(self):
        self.providers: dict[str, BaseAISearchProvider] = {}
    
    def register(self, name: str, provider: BaseAISearchProvider):
        self.providers[name] = provider
    
    async def search(
        self,
        query: str,
        providers: list[str] | None = None,
        strategy: str = "fastest",  # fastest | all | best
    ) -> UnifiedSearchResult:
        """执行聚合搜索"""
        target_providers = providers or list(self.providers.keys())
        
        if strategy == "fastest":
            # 返回最快的结果
            tasks = [
                self.providers[name].search(query)
                for name in target_providers
                if name in self.providers
            ]
            done, _ = await asyncio.wait(
                tasks, return_when=asyncio.FIRST_COMPLETED
            )
            for task in done:
                return task.result()
        
        elif strategy == "all":
            # 并发获取所有结果
            tasks = {
                name: self.providers[name].search(query)
                for name in target_providers
                if name in self.providers
            }
            results = {}
            for name, task in tasks.items():
                try:
                    results[name] = await task
                except Exception as e:
                    print(f"Provider {name} failed: {e}")
            
            # 返回最快的成功结果（或融合结果）
            if results:
                return list(results.values())[0]
        
        raise RuntimeError("No providers available")

# 使用示例
async def main():
    aggregator = SearchAggregator()
    
    # 注册多个搜索提供者
    aggregator.register("perplexity", PerplexityProvider("your-key"))
    # aggregator.register("openai", OpenAIProvider("your-key"))
    # aggregator.register("kimi", KimiProvider("your-key"))
    
    result = await aggregator.search(
        "什么是量子计算的最新突破",
        strategy="fastest"
    )
    
    print(f"答案来源: {result.provider}")
    print(f"响应时间: {result.latency_ms}ms")
    print(f"\n{result.answer}")

asyncio.run(main())

7.2 构建搜索API服务

"""
将AI搜索引擎封装为REST API
"""

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import asyncio

app = FastAPI(title="AI Search API", version="1.0.0")

class SearchRequest(BaseModel):
    query: str
    max_results: int = 10
    provider: str = "perplexity"
    language: str = "zh-CN"
    search_type: str = "quick"  # quick | pro

class SearchResponse(BaseModel):
    answer: str
    sources: list[dict]
    provider: str
    latency_ms: int
    sub_queries: list[str] = []

# 初始化搜索引擎
engine = AISearchEngine(
    searxng_url="http://localhost:8080",
    llm_api_key="your-key",
)

@app.post("/api/search", response_model=SearchResponse)
async def search(request: SearchRequest):
    """AI搜索接口"""
    try:
        result = await engine.search(
            query=request.query,
            max_results=request.max_results,
        )
        return SearchResponse(
            answer=result.answer,
            sources=[
                {"title": s.title, "url": s.url, "snippet": s.snippet}
                for s in result.sources
            ],
            provider="custom",
            latency_ms=0,
            sub_queries=result.sub_queries,
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/api/search/suggest")
async def suggest(q: str):
    """搜索建议/自动补全"""
    # 可接入LLM生成搜索建议
    suggestions = [
        f"{q} 最新进展",
        f"{q} 对比分析",
        f"{q} 实战教程",
    ]
    return {"suggestions": suggestions}

8. AI搜索优化（ASO）策略

8.1 什么是ASO？

ASO（AI Search Optimization）是针对AI搜索引擎的内容优化策略，类似于传统SEO但有本质区别：

8.2 ASO核心策略

1. 内容结构化

# 好的内容结构（AI友好）

## 什么是XX技术？
[清晰定义，50-100字]

## XX技术的核心优势
- 优势1：具体说明
- 优势2：具体说明
- 优势3：具体说明

## XX vs YY 对比

| 维度 | XX | YY |
|------|----|----|
| 性能 | 数据 | 数据 |
| 成本 | 数据 | 数据 |

## 实际应用案例
[具体的、有数据支撑的案例]

## 常见问题（FAQ）
**Q: 问题1？**
A: 简洁明确的回答。

**Q: 问题2？**
A: 简洁明确的回答。

2. 提供可引用的事实和数据

# AI搜索引擎更倾向于引用包含以下特征的内容：
CITABLE_FEATURES = {
    "统计数据": "2024年全球AI市场规模达到5000亿美元",
    "明确来源": "根据Gartner 2024年报告...",
    "具体案例": "某公司通过XX方案，效率提升40%",
    "对比数据": "A方案比B方案快3倍，成本低50%",
    "权威引用": "IEEE标准、RFC文档、官方文档",
}

3. 技术优化清单

ASO技术清单:
  内容层面:
    - 使用清晰的标题层级 (H1 > H2 > H3)
    - 在前100字包含核心答案
    - 使用列表、表格等结构化格式
    - 提供明确的数据和来源引用
    - 包含FAQ部分
    - 保持内容更新（标注日期）
  
  技术层面:
    - 使用Schema.org结构化数据标记
    - 确保页面可被爬虫访问
    - 提供RSS/Atom订阅
    - 使用语义化HTML标签
    - 移动端友好
  
  权威性:
    - 作者信息和专业背景
    - 引用权威来源
    - 展示专业资质
    - 用户评价和反馈

8.3 结构化数据标记示例

<!-- Schema.org FAQ结构化标记 -->
<script type="application/ld+json">
{
  "@context": "https://schema.org",
  "@type": "FAQPage",
  "mainEntity": [
    {
      "@type": "Question",
      "name": "什么是RAG技术？",
      "acceptedAnswer": {
        "@type": "Answer",
        "text": "RAG（Retrieval-Augmented Generation）是一种结合信息检索和文本生成的AI技术框架，通过在生成答案前先检索相关文档来提高回答的准确性和时效性。"
      }
    },
    {
      "@type": "Question",
      "name": "AI搜索引擎和传统搜索引擎有什么区别？",
      "acceptedAnswer": {
        "@type": "Answer",
        "text": "AI搜索引擎直接给出结构化答案并附带引用来源，而传统搜索引擎只返回链接列表。AI搜索支持自然语言提问和多轮对话。"
      }
    }
  ]
}
</script>

<!-- Schema.org HowTo结构化标记 -->
<script type="application/ld+json">
{
  "@context": "https://schema.org",
  "@type": "HowTo",
  "name": "如何自建AI搜索引擎",
  "step": [
    {
      "@type": "HowToStep",
      "name": "部署SearXNG",
      "text": "使用Docker部署SearXNG元搜索引擎"
    },
    {
      "@type": "HowToStep",
      "name": "配置LLM",
      "text": "选择并配置大语言模型作为答案生成引擎"
    }
  ]
}
</script>

9. 企业级AI搜索应用场景

9.1 企业知识库搜索

"""
企业内部知识库AI搜索系统
基于RAG架构，整合内部文档
"""

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.chains import RetrievalQA

class EnterpriseKnowledgeSearch:
    """企业知识库AI搜索"""
    
    def __init__(self):
        self.embeddings = OpenAIEmbeddings()
        self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
        )
        self.vectorstore = None
    
    def index_documents(self, documents: list[str]):
        """索引企业文档"""
        chunks = []
        for doc in documents:
            chunks.extend(self.text_splitter.split_text(doc))
        
        self.vectorstore = Chroma.from_texts(
            texts=chunks,
            embedding=self.embeddings,
            persist_directory="./enterprise_vectordb"
        )
        print(f"已索引 {len(chunks)} 个文档片段")
    
    def search(self, query: str, department: str = None) -> dict:
        """搜索企业知识库"""
        if not self.vectorstore:
            return {"error": "知识库未初始化"}
        
        # 可选：按部门过滤
        filter_dict = {"department": department} if department else None
        
        retriever = self.vectorstore.as_retriever(
            search_kwargs={"k": 5, "filter": filter_dict}
        )
        
        chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            retriever=retriever,
            return_source_documents=True,
        )
        
        result = chain.invoke({"query": query})
        
        return {
            "answer": result["result"],
            "sources": [
                {
                    "content": doc.page_content[:200],
                    "metadata": doc.metadata
                }
                for doc in result["source_documents"]
            ]
        }

# 使用示例
ks = EnterpriseKnowledgeSearch()

# 索引内部文档
ks.index_documents([
    "公司差旅报销政策：国内出差住宿标准...",
    "产品技术架构文档：系统采用微服务架构...",
    "人力资源手册：员工入职流程...",
])

# 搜索
result = ks.search("出差住宿报销标准是多少？")
print(result["answer"])

9.2 电商产品搜索

class EcommerceAISearch:
    """电商AI搜索引擎"""
    
    def search_products(self, query: str) -> dict:
        """
        用户自然语言查询 → 智能商品匹配
        
        示例：
        用户："给我妈买个生日礼物，预算500左右，她喜欢养花"
        
        AI分析：
        - 收礼人：母亲（中老年女性）
        - 场景：生日礼物
        - 预算：500元
        - 兴趣：园艺/花卉
        
        推荐结果：
        1. 智能花盆自动浇水器 - ¥299
        2. 高端园艺工具套装 - ¥459
        3. 花卉养护书籍礼盒 - ¥168
        """
        pass

9.3 更多应用场景

10. 最佳实践与总结

10.1 选型建议

选择AI搜索引擎的决策树：

你的需求是什么？
│
├── 个人日常搜索
│   └── 推荐：Perplexity / 秘塔AI搜索
│
├── 开发集成
│   ├── 预算充足 → Perplexity API / OpenAI + Web Search
│   └── 预算有限 → 自建SearXNG + 开源LLM
│
├── 企业内部搜索
│   ├── 数据敏感 → 自建方案（本地LLM + 私有向量库）
│   └── 可用云服务 → RAG + 云API
│
└── 特定领域搜索
    ├── 学术搜索 → Perplexity Academic / Semantic Scholar
    ├── 代码搜索 → Sourcegraph Cody / GitHub Copilot
    └── 法律搜索 → 秘塔法律 / Westlaw Edge

10.2 成本优化

# 搜索成本优化策略
COST_OPTIMIZATION = {
    "缓存策略": {
        "描述": "对相似查询缓存结果",
        "节省": "30-60%",
        "实现": "Redis + 语义相似度匹配",
    },
    "分级模型": {
        "描述": "简单问题用小模型，复杂问题用大模型",
        "节省": "40-70%",
        "实现": "查询复杂度分类器",
    },
    "结果复用": {
        "描述": "跨用户复用搜索结果",
        "节省": "20-40%",
        "实现": "共享缓存层",
    },
    "检索优化": {
        "描述": "减少不必要的外部检索调用",
        "节省": "20-30%",
        "实现": "查询意图分类 + 知识库预检索",
    },
}

10.3 关键要点总结

1. RAG是核心：所有AI搜索引擎都基于"检索增强生成"的范式
2. 检索质量决定答案质量：投入更多精力优化检索环节
3. 多源融合优于单一来源：聚合多个搜索引擎提高覆盖面
4. 引用透明是信任基础：始终标注信息来源
5. ASO将成为新SEO：内容需要为AI搜索优化
6. 自建方案可控性强：企业场景优先考虑自建
7. 成本需要精细管理：缓存、模型分级、结果复用是关键

下一篇推荐阅读：AI Agent工具使用与Function Calling完全教程