AI搜索 实战案例分享｜附源码

本文将通过一个可运行的实战案例，带你从零搭建一个“AI搜索”应用：用户输入自然语言问题，系统能够从本地知识库中检索相关内容，并结合大语言模型生成答案。文章包含方案设计、核心流程、代码实现、优化思路与完整源码示例。

一、什么是AI搜索？

传统搜索引擎通常依赖关键词匹配，例如用户搜索“如何提高接口性能”，系统会查找包含“接口”“性能”等关键词的文档。

但在真实业务中，用户的问题往往是自然语言表达，例如：

“我们系统响应慢，有哪些优化方向？”

这句话可能并没有直接出现“接口性能优化”这些关键词，但它和“接口性能优化方案”“缓存设计”“数据库索引优化”等内容高度相关。

这就是AI搜索要解决的问题。

AI搜索通常结合以下能力：

1. 语义理解
   不只看关键词，而是理解用户问题的含义。

2. 向量检索
   将文本转换成向量，通过向量相似度查找相关内容。

3. 混合检索
   同时结合关键词检索和向量检索，提高召回率。

4. 大模型生成答案
   将检索到的内容交给大语言模型，让它生成更自然、更完整的回答。

5. 引用来源
   给出答案依据，降低“幻觉”风险。

二、实战目标

本文实现一个本地AI搜索系统，具备以下能力：

• 支持导入本地Markdown知识库；
• 自动切分文档；
• 使用Embedding模型生成文本向量；
• 将向量存储到本地FAISS索引；
• 支持用户自然语言提问；
• 检索最相关的知识片段；
• 使用大模型基于知识库内容生成回答；
• 返回答案和引用来源。

最终效果类似：

用户问题：
如何提升接口查询性能？

AI回答：
可以从数据库索引、缓存、分页查询、SQL优化和异步化几个方向入手。
首先，应检查高频查询字段是否建立了合适索引……
参考资料：
1. docs/performance.md
2. docs/cache.md

三、技术选型

为了方便大家快速运行，本文采用Python实现。

说明：如果你使用的是国内大模型服务，只要它兼容OpenAI Chat Completions接口，也可以直接替换Base URL和API Key。

四、项目结构

建议项目结构如下：

ai-search-demo/
├── app.py
├── build_index.py
├── search_engine.py
├── requirements.txt
├── config.py
├── data/
│   └── docs/
│       ├── cache.md
│       ├── database.md
│       └── performance.md
└── storage/
    ├── faiss.index
    └── chunks.json

各文件作用说明：

五、安装依赖

创建 requirements.txt：

fastapi==0.115.0
uvicorn==0.30.6
sentence-transformers==3.0.1
faiss-cpu==1.8.0.post1
numpy==1.26.4
openai==1.40.6
python-dotenv==1.0.1

安装依赖：

pip install -r requirements.txt

如果你使用Apple Silicon或某些特殊环境，faiss-cpu安装可能失败，可以尝试：

conda install -c conda-forge faiss-cpu

六、准备知识库文档

在 data/docs 目录下创建几篇Markdown文档。

1. performance.md

# 接口性能优化

接口性能优化可以从多个方向入手。

第一，减少数据库查询次数。对于频繁访问的数据，可以使用缓存，避免每次请求都访问数据库。

第二，优化SQL语句。避免使用 select *，只查询必要字段。对于复杂查询，需要分析执行计划。

第三，增加合适的索引。索引可以显著提升查询性能，但索引过多也会影响写入性能。

第四，接口结果分页。对于大量数据查询，应该使用分页加载，避免一次性返回过多数据。

第五，异步处理耗时任务。例如发送邮件、生成报表、推送通知等操作，可以放入消息队列异步执行。

2. cache.md

# 缓存设计

缓存适合读多写少、数据变化不频繁的场景。

常见缓存策略包括 Cache Aside、Read Through 和 Write Through。

使用缓存时要注意缓存穿透、缓存击穿和缓存雪崩问题。

缓存穿透可以通过布隆过滤器或缓存空值解决。

缓存击穿可以通过互斥锁或逻辑过期解决。

缓存雪崩可以通过随机过期时间、多级缓存和限流降级解决。

3. database.md

# 数据库优化

数据库优化包括表结构设计、索引优化、SQL优化和连接池优化。

索引设计需要遵循最左前缀原则。对于组合索引，需要注意字段顺序。

慢SQL分析可以通过 explain 查看执行计划，判断是否走索引、扫描行数是否过多。

连接池可以复用数据库连接，减少频繁创建连接带来的性能损耗。

对于大表，可以考虑分库分表、冷热数据分离和归档策略。

七、配置文件

创建 config.py：

import os
from dotenv import load_dotenv

load_dotenv()

# 文档目录
DOCS_DIR = os.getenv("DOCS_DIR", "data/docs")

# 索引存储目录
STORAGE_DIR = os.getenv("STORAGE_DIR", "storage")

# FAISS索引文件
FAISS_INDEX_PATH = os.path.join(STORAGE_DIR, "faiss.index")

# 文本切片文件
CHUNKS_PATH = os.path.join(STORAGE_DIR, "chunks.json")

# Embedding模型
EMBEDDING_MODEL_NAME = os.getenv(
    "EMBEDDING_MODEL_NAME",
    "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
)

# OpenAI兼容接口配置
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")
OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini")

如果你使用环境变量，可以创建 .env 文件：

OPENAI_API_KEY=你的API_KEY
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_MODEL=gpt-4o-mini

八、构建索引核心逻辑

创建 build_index.py。

这个脚本主要完成以下事情：

1. 读取Markdown文件；
2. 将长文本切成较小片段；
3. 使用Embedding模型生成向量；
4. 使用FAISS建立索引；
5. 保存索引和文本片段。

import os
import json
import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

from config import (
    DOCS_DIR,
    STORAGE_DIR,
    FAISS_INDEX_PATH,
    CHUNKS_PATH,
    EMBEDDING_MODEL_NAME
)


def read_markdown_files(docs_dir: str):
    """
    读取目录下所有Markdown文件。
    """
    documents = []

    for root, _, files in os.walk(docs_dir):
        for file in files:
            if not file.endswith(".md"):
                continue

            path = os.path.join(root, file)

            with open(path, "r", encoding="utf-8") as f:
                content = f.read()

            documents.append({
                "source": path,
                "content": content
            })

    return documents


def split_text(text: str, chunk_size: int = 500, overlap: int = 80):
    """
    简单文本切分函数。

    chunk_size：每个文本块最大字符数
    overlap：相邻文本块重叠字符数

    为什么需要overlap？
    因为如果严格切分，可能会把一个完整语义拆开。
    重叠可以缓解上下文断裂问题。
    """
    chunks = []
    start = 0
    text = text.strip()

    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end].strip()

        if chunk:
            chunks.append(chunk)

        start = end - overlap

        if start < 0:
            start = 0

        if start >= len(text):
            break

    return chunks


def build_chunks(documents):
    """
    将文档转换成文本块。
    """
    all_chunks = []

    for doc in documents:
        source = doc["source"]
        content = doc["content"]

        chunks = split_text(content)

        for i, chunk in enumerate(chunks):
            all_chunks.append({
                "id": len(all_chunks),
                "source": source,
                "chunk_index": i,
                "text": chunk
            })

    return all_chunks


def normalize_vectors(vectors: np.ndarray):
    """
    归一化向量。

    使用IndexFlatIP时，如果向量已归一化，
    内积就等价于余弦相似度。
    """
    faiss.normalize_L2(vectors)
    return vectors


def main():
    os.makedirs(STORAGE_DIR, exist_ok=True)

    print("正在读取文档...")
    documents = read_markdown_files(DOCS_DIR)

    if not documents:
        raise RuntimeError(f"未找到Markdown文档，请检查目录：{DOCS_DIR}")

    print(f"读取到 {len(documents)} 个文档")

    print("正在切分文档...")
    chunks = build_chunks(documents)
    print(f"生成 {len(chunks)} 个文本块")

    print("正在加载Embedding模型...")
    model = SentenceTransformer(EMBEDDING_MODEL_NAME)

    texts = [chunk["text"] for chunk in chunks]

    print("正在生成向量...")
    embeddings = model.encode(
        texts,
        batch_size=32,
        show_progress_bar=True,
        convert_to_numpy=True
    )

    embeddings = embeddings.astype("float32")
    embeddings = normalize_vectors(embeddings)

    dim = embeddings.shape[1]

    print("正在构建FAISS索引...")
    index = faiss.IndexFlatIP(dim)
    index.add(embeddings)

    print("正在保存索引...")
    faiss.write_index(index, FAISS_INDEX_PATH)

    with open(CHUNKS_PATH, "w", encoding="utf-8") as f:
        json.dump(chunks, f, ensure_ascii=False, indent=2)

    print("索引构建完成")
    print(f"FAISS索引路径：{FAISS_INDEX_PATH}")
    print(f"文本块路径：{CHUNKS_PATH}")


if __name__ == "__main__":
    main()

运行构建索引：

python build_index.py

成功后会看到类似输出：

正在读取文档...
读取到 3 个文档
正在切分文档...
生成 3 个文本块
正在加载Embedding模型...
正在生成向量...
正在构建FAISS索引...
正在保存索引...
索引构建完成

九、实现AI搜索引擎

创建 search_engine.py。

该文件负责：

• 加载FAISS索引；
• 加载文本块；
• 将用户问题转换成向量；
• 检索Top K相关内容；
• 构造Prompt；
• 调用大模型生成答案。

import json
import faiss
import numpy as np
from typing import List, Dict, Any
from sentence_transformers import SentenceTransformer
from openai import OpenAI

from config import (
    FAISS_INDEX_PATH,
    CHUNKS_PATH,
    EMBEDDING_MODEL_NAME,
    OPENAI_API_KEY,
    OPENAI_BASE_URL,
    OPENAI_MODEL
)


class AISearchEngine:
    def __init__(self):
        self.embedding_model = SentenceTransformer(EMBEDDING_MODEL_NAME)
        self.index = faiss.read_index(FAISS_INDEX_PATH)

        with open(CHUNKS_PATH, "r", encoding="utf-8") as f:
            self.chunks = json.load(f)

        self.client = OpenAI(
            api_key=OPENAI_API_KEY,
            base_url=OPENAI_BASE_URL
        )

    def _embed_query(self, query: str):
        """
        将用户问题转换为向量。
        """
        vector = self.embedding_model.encode(
            [query],
            convert_to_numpy=True
        ).astype("float32")

        faiss.normalize_L2(vector)
        return vector

    def search(self, query: str, top_k: int = 5) -> List[Dict[str, Any]]:
        """
        执行向量检索。
        """
        query_vector = self._embed_query(query)

        scores, ids = self.index.search(query_vector, top_k)

        results = []

        for score, idx in zip(scores[0], ids[0]):
            if idx == -1:
                continue

            chunk = self.chunks[idx]

            results.append({
                "score": float(score),
                "source": chunk["source"],
                "chunk_index": chunk["chunk_index"],
                "text": chunk["text"]
            })

        return results

    def build_prompt(self, query: str, contexts: List[Dict[str, Any]]) -> str:
        """
        构造给大模型的Prompt。
        """
        context_text = ""

        for i, item in enumerate(contexts, start=1):
            context_text += (
                f"【资料{i}】\n"
                f"来源：{item['source']}\n"
                f"内容：{item['text']}\n\n"
            )

        prompt = f"""
你是一个严谨的企业知识库问答助手。
请你只根据下面提供的资料回答用户问题。

要求：
1. 如果资料中有明确答案，请进行结构化总结。
2. 如果资料不足以回答，请直接说明“根据现有资料无法确定”。
3. 不要编造资料中没有的信息。
4. 回答最后请列出参考来源。

用户问题：
{query}

可用资料：
{context_text}

请输出中文回答。
"""
        return prompt.strip()

    def answer(self, query: str, top_k: int = 5) -> Dict[str, Any]:
        """
        检索并生成回答。
        """
        contexts = self.search(query, top_k=top_k)

        if not contexts:
            return {
                "answer": "未检索到相关资料。",
                "references": []
            }

        prompt = self.build_prompt(query, contexts)

        completion = self.client.chat.completions.create(
            model=OPENAI_MODEL,
            messages=[
                {
                    "role": "system",
                    "content": "你是一个专业、严谨、可靠的AI搜索助手。"
                },
                {
                    "role": "user",
                    "content": prompt
                }
            ],
            temperature=0.2
        )

        answer = completion.choices[0].message.content

        references = []
        seen = set()

        for item in contexts:
            key = item["source"]
            if key not in seen:
                seen.add(key)
                references.append({
                    "source": item["source"],
                    "score": item["score"]
                })

        return {
            "answer": answer,
            "references": references,
            "contexts": contexts
        }

十、提供API接口

创建 app.py：

from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional

from search_engine import AISearchEngine


app = FastAPI(title="AI Search Demo")

engine = AISearchEngine()


class SearchRequest(BaseModel):
    query: str
    top_k: Optional[int] = 5


@app.get("/")
def health_check():
    return {
        "status": "ok",
        "message": "AI Search Demo is running"
    }


@app.post("/search")
def search(req: SearchRequest):
    results = engine.search(req.query, top_k=req.top_k)

    return {
        "query": req.query,
        "results": results
    }


@app.post("/answer")
def answer(req: SearchRequest):
    result = engine.answer(req.query, top_k=req.top_k)

    return {
        "query": req.query,
        **result
    }

启动服务：

uvicorn app:app --reload --port 8000

浏览器访问：

http://127.0.0.1:8000/docs

可以看到FastAPI自动生成的接口文档。

十一、测试搜索接口

使用curl测试：

curl -X POST "http://127.0.0.1:8000/search" \
  -H "Content-Type: application/json" \
  -d '{"query":"如何提升接口查询性能？","top_k":3}'

返回示例：

{
  "query": "如何提升接口查询性能？",
  "results": [
    {
      "score": 0.7213,
      "source": "data/docs/performance.md",
      "chunk_index": 0,
      "text": "# 接口性能优化\n\n接口性能优化可以从多个方向入手..."
    },
    {
      "score": 0.6845,
      "source": "data/docs/database.md",
      "chunk_index": 0,
      "text": "# 数据库优化\n\n数据库优化包括表结构设计、索引优化..."
    },
    {
      "score": 0.5912,
      "source": "data/docs/cache.md",
      "chunk_index": 0,
      "text": "# 缓存设计\n\n缓存适合读多写少、数据变化不频繁的场景..."
    }
  ]
}

这个结果说明系统并不是简单查关键词，而是理解“提升接口查询性能”和“数据库优化”“缓存设计”之间的语义关系。

十二、测试问答接口

curl -X POST "http://127.0.0.1:8000/answer" \
  -H "Content-Type: application/json" \
  -d '{"query":"如何提升接口查询性能？","top_k":3}'

返回示例：

{
  "query": "如何提升接口查询性能？",
  "answer": "可以从以下几个方面提升接口查询性能：\n\n1. 减少数据库查询次数...\n2. 使用缓存...\n3. 优化SQL语句...\n4. 增加合适索引...\n5. 对结果进行分页...\n\n参考来源：\n- data/docs/performance.md\n- data/docs/database.md\n- data/docs/cache.md",
  "references": [
    {
      "source": "data/docs/performance.md",
      "score": 0.7213
    },
    {
      "source": "data/docs/database.md",
      "score": 0.6845
    },
    {
      "source": "data/docs/cache.md",
      "score": 0.5912
    }
  ]
}

十三、核心流程解析

整个AI搜索流程可以概括为：

文档
 ↓
文本切分
 ↓
Embedding向量化
 ↓
FAISS索引
 ↓
用户提问
 ↓
问题向量化
 ↓
向量相似度检索
 ↓
召回相关文本
 ↓
构造Prompt
 ↓
大模型生成答案

这类架构也被称为RAG，即：

Retrieval-Augmented Generation，检索增强生成。

RAG的核心思想是：

大模型本身不直接记住所有企业知识，而是在回答问题前，先从外部知识库检索相关内容，再基于检索结果回答。

这样做有几个明显好处：

1. 降低幻觉
   模型回答基于检索资料，而不是凭空生成。

2. 知识可更新
   更新知识库后重新构建索引即可，不需要重新训练模型。

3. 适合私有数据
   企业内部文档、客服话术、产品手册、技术规范都可以接入。

4. 成本较低
   相比微调模型，RAG实现成本更低，迭代更快。

十四、为什么要做文本切分？

很多初学者会问：

为什么不直接把整篇文档转成向量？

原因主要有三个。

1. 文档太长，语义会被稀释

如果一篇文档包含多个主题，例如既讲缓存，又讲数据库，还讲消息队列，那么整篇文档向量会变得很“平均”，检索时不够精准。

2. 大模型上下文有限

即使检索到整篇文档，也不适合全部塞进Prompt。文档太长会占用大量Token，并且增加成本。

3. 更容易定位引用来源

切分后可以知道答案来自哪一段内容，而不是只知道来自某篇文档。

不过，切分也不能太小。如果切得太碎，单个片段缺少上下文，模型可能无法理解完整语义。

实践中常见参数：

十五、当前方案的不足

本文实现的是一个入门版AI搜索系统，可以跑通完整流程，但生产环境中还需要进一步增强。

1. 缺少关键词检索

纯向量检索适合语义搜索，但对精确词、编号、错误码、专有名词不一定友好。

例如：

ERR_10027是什么意思？

这类问题可能关键词检索更有效。

生产环境中通常会使用：

• BM25关键词检索；
• 向量检索；
• 两路结果融合。

2. 缺少重排序

向量检索召回的Top K结果不一定顺序最优。

可以引入Reranker模型，例如：

• bge-reranker；
• Cohere rerank；
• Jina reranker。

流程变为：

先召回Top 50
 ↓
Reranker重新打分
 ↓
取Top 5给大模型

3. 缺少权限控制

企业知识库往往涉及权限：

• A部门只能看A部门文档；
• 普通员工不能看财务数据；
• 客服只能查询客服知识库。

因此检索时必须加入权限过滤。

4. 缺少增量更新

当前示例每次都重新构建索引。生产环境需要支持：

• 新增文档；
• 修改文档；
• 删除文档；
• 局部重建索引。

5. 缺少质量评估

AI搜索系统上线后，需要持续评估：

• 召回是否准确；
• 回答是否忠于原文；
• 是否存在幻觉；
• 用户是否满意；
• 哪些问题没有答案。

十六、如何升级为混合检索？

如果要增强搜索效果，可以加入BM25。

混合检索大致流程：

用户问题
 ├── BM25关键词检索
 └── 向量语义检索
        ↓
   结果合并
        ↓
   分数归一化
        ↓
   Rerank重排序
        ↓
   大模型回答

简单融合策略可以使用RRF，Reciprocal Rank Fusion。

公式大致为：

score = 1 / (k + rank)

其中：

• rank 表示某个结果在某一路检索中的排名；
• k 是平滑参数，常用60。

RRF的优点是不依赖不同检索系统的原始分数尺度，因此非常适合融合BM25和向量检索。

十七、Prompt设计建议

AI搜索的回答质量很大程度取决于Prompt。

一个好的RAG Prompt通常要包含：

1. 角色设定
   告诉模型它是知识库问答助手。

2. 资料边界
   要求只能根据检索资料回答。

3. 不确定性处理
   如果资料不足，明确回答无法确定。

4. 输出格式
   要求分点回答、给出引用来源。

5. 禁止编造
   明确不要使用资料外信息。

示例：

你是企业内部知识库问答助手。
请只基于提供的资料回答问题。
如果资料中没有答案，请回答“根据现有资料无法确定”。
不要编造不存在的制度、流程、数字或负责人。
回答时请使用条目化结构，并在最后列出参考资料。

十八、生产环境落地建议

如果你要在真实业务中落地AI搜索，可以重点关注以下方面。

1. 数据治理优先于模型调参

很多AI搜索效果差，并不是模型不够强，而是知识库本身质量差。

常见问题包括：

• 文档过期；
• 标题不清晰；
• 同一问题多份答案冲突；
• 文档没有结构；
• 图片和表格没有被解析；
• PDF扫描件无法提取文字。

所以，在做AI搜索前，建议先对知识库进行整理。

2. 建立标准文档格式

例如要求每篇文档包含：

# 标题

## 适用范围

## 问题描述

## 解决方案

## 注意事项

## 更新时间

## 负责人

结构越清晰，切分和检索效果越好。

3. 保留原文引用

AI生成的答案不应该成为唯一结果。最好同时展示：

• AI总结；
• 引用片段；
• 原文链接；
• 相似度分数；
• 更新时间。

这样用户可以自行核验。

4. 做好安全控制

尤其是企业内部知识库，要注意：

• API Key安全；
• 用户身份认证；
• 文档权限过滤；
• 敏感信息脱敏；
• 日志合规；
• 防止Prompt Injection。

5. 建立反馈机制

每次回答后，可以让用户点击：

• 有帮助；
• 无帮助；
• 答案错误；
• 没有找到想要内容。

这些反馈可以用于后续优化检索、补充知识库和评估系统质量。

十九、完整运行步骤

汇总一下完整步骤。

1. 创建项目

mkdir ai-search-demo
cd ai-search-demo

2. 安装依赖

pip install -r requirements.txt

3. 准备文档

mkdir -p data/docs

将Markdown文档放入 data/docs。

4. 配置API Key

创建 .env：

OPENAI_API_KEY=你的API_KEY
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_MODEL=gpt-4o-mini

5. 构建索引

python build_index.py

6. 启动服务

uvicorn app:app --reload --port 8000

7. 调用接口

curl -X POST "http://127.0.0.1:8000/answer" \
  -H "Content-Type: application/json" \
  -d '{"query":"如何解决缓存雪崩？","top_k":3}'

二十、总结

本文通过一个完整案例实现了一个基础版AI搜索系统。

它的核心并不复杂：

• 把知识库文档切成片段；
• 用Embedding模型生成向量；
• 用FAISS做相似度检索；
• 把检索结果交给大模型；
• 让大模型基于资料生成答案。

这个方案虽然简单，但已经具备RAG系统的基本形态，可以应用在很多场景中，例如：

• 企业内部知识库；
• 技术文档问答；
• 客服智能助手；
• 产品手册搜索；
• 法务制度查询；
• 运维故障排查。

如果继续增强，可以引入混合检索、Reranker、权限过滤、增量更新、反馈评估等能力，逐步演进为生产级AI搜索平台。

AI搜索的关键不是“让模型记住所有知识”，而是让模型在回答之前，先找到正确的知识。检索越准，答案越可靠；知识库越规范，系统越稳定。