AI 搜索引擎架构解析：从 Perplexity 到垂直领域 AI 搜索【2026】

原始查询: "Perplexity 是怎么做到的"
改写结果:
  → "Perplexity AI architecture technical design"
  → "Perplexity search engine RAG pipeline"
  → "How does Perplexity AI generate answers with citations"

系统提示词(简化版):
你是一个 AI 搜索引擎。基于以下检索到的来源，生成一段全面的答案。
规则:
1. 每个事实性陈述必须标注来源编号 [1], [2] 等
2. 如果来源之间存在矛盾，明确指出分歧
3. 不要编造来源中没有的信息
4. 以 Markdown 格式输出，包含适当的标题和列表

来源:
[1] {source_1_content}
[2] {source_2_content}
...

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
import asyncio
import httpx

# ========== 第一步：查询改写 ==========
query_rewriter = ChatOpenAI(model="gpt-4o-mini", temperature=0)

rewrite_prompt = ChatPromptTemplate.from_messages([
    ("system", """将用户的搜索查询改写为 3 个更精确的子查询。
输出格式：每行一个子查询，不要编号。"""),
    ("user", "{query}")
])

rewrite_chain = rewrite_prompt | query_rewriter | StrOutputParser()

# ========== 第二步：多源并行检索 ==========
TAVILY_API_KEY = "your-tavily-api-key"

async def search_web(query: str, max_results: int = 5) -> list[dict]:
    """调用 Tavily API 进行网页搜索"""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.tavily.com/search",
            json={
                "api_key": TAVILY_API_KEY,
                "query": query,
                "max_results": max_results,
                "include_raw_content": True,
            }
        )
        return response.json().get("results", [])

async def parallel_search(sub_queries: list[str]) -> list[dict]:
    """并行执行多个搜索查询"""
    tasks = [search_web(q) for q in sub_queries]
    results = await asyncio.gather(*tasks)
    # 扁平化并去重
    seen_urls = set()
    unique_results = []
    for batch in results:
        for r in batch:
            if r["url"] not in seen_urls:
                seen_urls.add(r["url"])
                unique_results.append(r)
    return unique_results

# ========== 第三步：来源排序 ==========
def rank_sources(results: list[dict], query: str) -> list[dict]:
    """基于相关性评分排序来源"""
    for r in results:
        score = r.get("score", 0)
        # 新鲜度加权：越近的内容分数越高
        r["final_score"] = score
    results.sort(key=lambda x: x["final_score"], reverse=True)
    return results[:8]  # 保留 Top-8

# ========== 第四步：答案合成 ==========
answer_llm = ChatOpenAI(model="gpt-4o", temperature=0.1, streaming=True)

answer_prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个 AI 搜索引擎。基于以下来源生成带引用的答案。
规则:
- 每个事实性陈述标注来源 [1], [2] 等
- 使用 Markdown 格式
- 在答案末尾列出所有来源的标题和 URL

来源:
{sources}"""),
    ("user", "{query}")
])

async def generate_answer(query: str, sources: list[dict]):
    """流式生成带引用的答案"""
    sources_text = "\n".join([
        f"[{i+1}] {s['title']}\nURL: {s['url']}\n内容: {s['content'][:1000]}"
        for i, s in enumerate(sources)
    ])

    chain = answer_prompt | answer_llm | StrOutputParser()
    async for chunk in chain.astream({
        "query": query,
        "sources": sources_text
    }):
        print(chunk, end="", flush=True)

# ========== 完整管线 ==========
async def ai_search(user_query: str):
    """AI 搜索引擎主函数"""
    # 1. 查询改写
    rewritten = await rewrite_chain.ainvoke({"query": user_query})
    sub_queries = [q.strip() for q in rewritten.strip().split("\n") if q.strip()]

    # 2. 并行检索
    results = await parallel_search(sub_queries)

    # 3. 排序去重
    ranked = rank_sources(results, user_query)

    # 4. 答案合成
    await generate_answer(user_query, ranked)

# 运行示例
asyncio.run(ai_search("Perplexity AI 的技术架构是怎样的"))

import OpenAI from "openai";

const openai = new OpenAI();
const TAVILY_API_KEY = process.env.TAVILY_API_KEY;

// ========== 查询改写 ==========
async function rewriteQuery(query) {
  const response = await openai.chat.completions.create({
    model: "gpt-4o-mini",
    messages: [
      {
        role: "system",
        content: `将用户查询改写为 3 个更精确的搜索子查询。
每行输出一个子查询，不要编号。`,
      },
      { role: "user", content: query },
    ],
    temperature: 0,
  });
  return response.choices[0].message.content
    .split("\n")
    .filter((q) => q.trim());
}

// ========== 并行搜索 ==========
async function searchWeb(query, maxResults = 5) {
  const res = await fetch("https://api.tavily.com/search", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      api_key: TAVILY_API_KEY,
      query,
      max_results: maxResults,
      include_raw_content: true,
    }),
  });
  const data = await res.json();
  return data.results || [];
}

async function parallelSearch(subQueries) {
  const batches = await Promise.all(subQueries.map((q) => searchWeb(q)));
  const seen = new Set();
  return batches.flat().filter((r) => {
    if (seen.has(r.url)) return false;
    seen.add(r.url);
    return true;
  });
}

// ========== 答案合成（流式） ==========
async function generateAnswer(query, sources) {
  const sourcesText = sources
    .slice(0, 8)
    .map(
      (s, i) =>
        `[${i + 1}] ${s.title}\nURL: ${s.url}\n内容: ${s.content?.slice(0, 1000)}`
    )
    .join("\n\n");

  const stream = await openai.chat.completions.create({
    model: "gpt-4o",
    messages: [
      {
        role: "system",
        content: `你是一个 AI 搜索引擎。基于以下来源生成带引用的答案。
每个事实标注 [1], [2] 等引用编号。使用 Markdown 格式。
在末尾列出来源。\n\n来源:\n${sourcesText}`,
      },
      { role: "user", content: query },
    ],
    temperature: 0.1,
    stream: true,
  });

  for await (const chunk of stream) {
    const text = chunk.choices[0]?.delta?.content || "";
    process.stdout.write(text);
  }
}

// ========== 完整管线 ==========
async function aiSearch(userQuery) {
  console.log(`🔍 搜索: ${userQuery}\n`);

  const subQueries = await rewriteQuery(userQuery);
  console.log(`📝 子查询: ${subQueries.join(" | ")}\n`);

  const results = await parallelSearch(subQueries);
  console.log(`📚 检索到 ${results.length} 个来源\n`);

  await generateAnswer(userQuery, results);
}

await aiSearch("Perplexity AI 的技术架构是怎样的");