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

在大模型应用落地过程中，“AI搜索”几乎是最常见、也最容易产生实际业务价值的场景之一。相比传统关键词搜索，AI搜索能够理解用户自然语言意图，支持语义召回、上下文理解、多轮追问，并能基于企业私有知识库生成答案。

本文将通过一个完整实战案例，分享如何从零搭建一个简易版 AI知识库搜索系统。系统支持将本地文档切片、向量化、存入向量数据库，并在用户提问时进行语义检索，最后结合大语言模型生成答案。

本文内容包括：

• AI搜索的核心原理
• 实战案例架构设计
• 文档加载与文本切片
• 向量化与向量数据库存储
• 检索与问答生成
• FastAPI接口封装
• 完整源码示例
• 优化方向与落地建议

一、什么是AI搜索？

传统搜索主要依赖关键词匹配，例如用户搜索“订单退款流程”，系统会查找包含“订单”“退款”“流程”等关键词的文档。它的优点是实现简单、速度快，但缺点也很明显：如果用户换一种表达方式，比如“我想把买的东西退掉，钱怎么回来”，传统搜索很可能无法准确匹配。

AI搜索的核心能力在于“语义理解”。

它会将用户问题和文档内容都转换成向量，也就是一组高维数字表示。语义相近的文本，在向量空间中的距离也会更近。因此，即使用户没有输入精确关键词，只要表达的意思相近，也可以被召回。

一个典型的AI搜索流程如下：

用户问题
   ↓
问题向量化
   ↓
向量数据库相似度检索
   ↓
召回相关文档片段
   ↓
大模型结合上下文生成答案
   ↓
返回用户

这种模式也常被称为 RAG，即 Retrieval-Augmented Generation，中文通常翻译为“检索增强生成”。

二、实战案例背景

假设我们要为一家企业搭建一个内部知识库问答系统。企业员工可以通过自然语言提问，例如：

• “员工报销需要哪些材料？”
• “年假怎么申请？”
• “客户退款审批流程是什么？”
• “新员工入职需要完成哪些步骤？”

系统需要从企业内部文档中找到相关内容，并生成简洁准确的答案。

为了便于演示，本文使用以下技术栈：

说明：本文代码以教学和原型验证为主，生产环境还需要考虑权限、日志、审计、缓存、数据隔离、容灾等问题。

三、项目目录结构

建议项目结构如下：

ai-search-demo/
├── app.py                  # FastAPI服务入口
├── build_index.py           # 构建向量索引
├── config.py                # 配置文件
├── requirements.txt         # 依赖包
├── data/
│   ├── hr_policy.md          # 示例知识文档
│   └── refund_process.md     # 示例知识文档
├── storage/
│   ├── faiss.index           # FAISS索引文件
│   └── documents.json        # 文档片段元数据
└── utils/
    ├── loader.py             # 文档加载
    ├── splitter.py           # 文本切片
    ├── embedding.py          # 向量模型
    └── llm.py                # 大模型调用

四、安装依赖

首先创建虚拟环境：

python -m venv venv

激活虚拟环境：

# macOS / Linux
source venv/bin/activate

# Windows
venv\Scripts\activate

安装依赖：

pip install fastapi uvicorn faiss-cpu sentence-transformers openai python-dotenv

也可以创建 requirements.txt：

fastapi
uvicorn
faiss-cpu
sentence-transformers
openai
python-dotenv

五、准备示例知识库文档

在 data/hr_policy.md 中写入：

# 员工假期制度

员工年假根据入职年限计算。入职满一年不满十年的员工，每年享有5天年假；满十年不满二十年的员工，每年享有10天年假；满二十年以上的员工，每年享有15天年假。

员工申请年假需要提前三个工作日在OA系统提交申请，并选择直属主管作为审批人。审批通过后，员工方可休假。

病假需要提供医院开具的病假证明。连续病假超过三天的，需要同时提交诊断证明和病历记录。

在 data/refund_process.md 中写入：

# 客户退款流程

客户申请退款后，客服人员需要先核实订单状态。如果订单尚未发货，可以直接进入退款审批流程。

如果订单已经发货，客服需要确认客户是否已经拒收或者退回商品。仓库收到退货并完成质检后，财务部门在三个工作日内完成退款。

退款金额超过5000元时，需要部门负责人和财务负责人双重审批。普通退款由客服主管审批即可。

六、配置文件

新建 config.py：

import os
from dotenv import load_dotenv

load_dotenv()

DATA_DIR = "data"
STORAGE_DIR = "storage"

FAISS_INDEX_PATH = os.path.join(STORAGE_DIR, "faiss.index")
DOCUMENTS_PATH = os.path.join(STORAGE_DIR, "documents.json")

EMBEDDING_MODEL_NAME = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"

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")

这里使用了 sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2 作为向量模型。它支持多语言场景，对中文也有一定效果。生产环境中，如果对中文语义召回要求更高，可以换成效果更好的中文Embedding模型。

七、文档加载模块

新建 utils/loader.py：

import os


def load_documents(data_dir: str):
    """
    加载指定目录下的txt和md文件
    """
    documents = []

    for filename in os.listdir(data_dir):
        if not filename.endswith((".txt", ".md")):
            continue

        filepath = os.path.join(data_dir, filename)

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

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

    return documents

这个模块负责读取本地知识库文件。实际项目中，文档来源可能更加复杂，例如：

• PDF
• Word
• Excel
• 企业Wiki
• 飞书文档
• 钉钉知识库
• 数据库记录
• 客服工单
• 产品说明书

不同来源的数据最终都需要转换成统一的文本格式。

八、文本切片模块

AI搜索中非常关键的一步是文本切片。

如果文档太长，直接整体向量化会导致语义信息不够精确；如果切得太碎，又可能丢失上下文。因此需要在召回准确率和上下文完整性之间做平衡。

新建 utils/splitter.py：

def split_text(text: str, chunk_size: int = 300, overlap: int = 50):
    """
    简单文本切片
    chunk_size: 每个片段最大字符数
    overlap: 相邻片段重叠字符数
    """
    chunks = []

    if not text:
        return chunks

    start = 0
    text_length = len(text)

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

        if chunk:
            chunks.append(chunk)

        start += chunk_size - overlap

    return chunks


def build_chunks(documents):
    """
    将原始文档转换为文档片段
    """
    all_chunks = []

    for doc in documents:
        chunks = split_text(doc["content"])

        for i, chunk in enumerate(chunks):
            all_chunks.append({
                "source": doc["source"],
                "chunk_id": i,
                "content": chunk
            })

    return all_chunks

这里采用了固定长度切片，并设置了 overlap 重叠区域。重叠的作用是避免关键信息刚好被切断。

生产环境中可以进一步优化为：

• 按标题层级切片
• 按段落切片
• 按句子切片
• Markdown结构化切片
• PDF目录结构切片
• 根据Token数量切片

九、向量模型模块

新建 utils/embedding.py：

from sentence_transformers import SentenceTransformer
import numpy as np
from config import EMBEDDING_MODEL_NAME


class EmbeddingModel:
    def __init__(self):
        self.model = SentenceTransformer(EMBEDDING_MODEL_NAME)

    def encode(self, texts):
        """
        将文本转换为向量，并进行归一化
        """
        embeddings = self.model.encode(
            texts,
            convert_to_numpy=True,
            normalize_embeddings=True
        )
        return embeddings.astype("float32")

这里对向量做了归一化，后续使用FAISS的内积检索时，归一化后的内积等价于余弦相似度。

十、构建向量索引

新建 build_index.py：

import os
import json
import faiss

from config import DATA_DIR, STORAGE_DIR, FAISS_INDEX_PATH, DOCUMENTS_PATH
from utils.loader import load_documents
from utils.splitter import build_chunks
from utils.embedding import EmbeddingModel


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

    print("正在加载文档...")
    documents = load_documents(DATA_DIR)
    print(f"加载文档数量：{len(documents)}")

    print("正在切分文档...")
    chunks = build_chunks(documents)
    print(f"生成文档片段数量：{len(chunks)}")

    if not chunks:
        print("没有可索引的文档片段")
        return

    print("正在初始化向量模型...")
    embedding_model = EmbeddingModel()

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

    print("正在生成向量...")
    embeddings = embedding_model.encode(texts)

    dimension = embeddings.shape[1]

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

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

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

    print("索引构建完成！")


if __name__ == "__main__":
    main()

运行命令：

python build_index.py

运行成功后，会在 storage/ 目录下生成：

faiss.index
documents.json

这两个文件分别保存向量索引和文档片段元数据。

十一、大模型调用模块

新建 utils/llm.py：

from openai import OpenAI
from config import OPENAI_API_KEY, OPENAI_BASE_URL, OPENAI_MODEL


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


def generate_answer(question: str, contexts: list):
    """
    基于召回上下文生成答案
    """
    context_text = "\n\n".join([
        f"【资料{i + 1}】\n{ctx['content']}"
        for i, ctx in enumerate(contexts)
    ])

    prompt = f"""
你是一个企业知识库问答助手。请严格根据给定资料回答用户问题。

要求：
1. 如果资料中有明确答案，请直接回答。
2. 如果资料中没有相关信息，请说明“根据现有资料无法确定”。
3. 不要编造资料中不存在的内容。
4. 回答尽量简洁、清晰、可执行。

资料如下：
{context_text}

用户问题：
{question}
"""

    response = client.chat.completions.create(
        model=OPENAI_MODEL,
        messages=[
            {"role": "system", "content": "你是一个严谨的企业知识库问答助手。"},
            {"role": "user", "content": prompt}
        ],
        temperature=0.2
    )

    return response.choices[0].message.content

这里的关键点是提示词约束。RAG系统最怕的问题之一是“幻觉”，也就是模型编造知识库中没有的信息。因此提示词里必须明确告诉模型：只能根据资料回答，资料不足时要如实说明。

十二、FastAPI搜索接口

新建 app.py：

import json
import faiss
from fastapi import FastAPI
from pydantic import BaseModel

from config import FAISS_INDEX_PATH, DOCUMENTS_PATH
from utils.embedding import EmbeddingModel
from utils.llm import generate_answer


app = FastAPI(title="AI搜索实战Demo")

embedding_model = EmbeddingModel()
index = faiss.read_index(FAISS_INDEX_PATH)

with open(DOCUMENTS_PATH, "r", encoding="utf-8") as f:
    documents = json.load(f)


class SearchRequest(BaseModel):
    question: str
    top_k: int = 3


class SearchResult(BaseModel):
    source: str
    chunk_id: int
    content: str
    score: float


@app.post("/search")
def search(req: SearchRequest):
    """
    只做语义检索，不调用大模型
    """
    query_embedding = embedding_model.encode([req.question])

    scores, indices = index.search(query_embedding, req.top_k)

    results = []

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

        doc = documents[idx]

        results.append({
            "source": doc["source"],
            "chunk_id": doc["chunk_id"],
            "content": doc["content"],
            "score": float(score)
        })

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


@app.post("/ask")
def ask(req: SearchRequest):
    """
    检索 + 大模型生成答案
    """
    query_embedding = embedding_model.encode([req.question])

    scores, indices = index.search(query_embedding, req.top_k)

    contexts = []

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

        doc = documents[idx]
        contexts.append({
            "source": doc["source"],
            "chunk_id": doc["chunk_id"],
            "content": doc["content"],
            "score": float(score)
        })

    answer = generate_answer(req.question, contexts)

    return {
        "question": req.question,
        "answer": answer,
        "contexts": contexts
    }

启动服务：

uvicorn app:app --reload --port 8000

访问接口文档：

http://127.0.0.1:8000/docs

十三、测试搜索效果

1. 测试语义搜索

请求：

curl -X POST "http://127.0.0.1:8000/search" \
-H "Content-Type: application/json" \
-d '{
  "question": "员工想休年假要提前多久申请？",
  "top_k": 3
}'

可能返回：

{
  "question": "员工想休年假要提前多久申请？",
  "results": [
    {
      "source": "hr_policy.md",
      "chunk_id": 0,
      "content": "员工年假根据入职年限计算。入职满一年不满十年的员工，每年享有5天年假；满十年不满二十年的员工，每年享有10天年假；满二十年以上的员工，每年享有15天年假。\n\n员工申请年假需要提前三个工作日在OA系统提交申请，并选择直属主管作为审批人。审批通过后，员工方可休假。",
      "score": 0.72
    }
  ]
}

2. 测试问答接口

请求：

curl -X POST "http://127.0.0.1:8000/ask" \
-H "Content-Type: application/json" \
-d '{
  "question": "客户退款超过5000元需要谁审批？",
  "top_k": 3
}'

可能返回：

{
  "question": "客户退款超过5000元需要谁审批？",
  "answer": "客户退款金额超过5000元时，需要部门负责人和财务负责人双重审批。",
  "contexts": [
    {
      "source": "refund_process.md",
      "chunk_id": 0,
      "content": "客户申请退款后，客服人员需要先核实订单状态。如果订单尚未发货，可以直接进入退款审批流程。\n\n如果订单已经发货，客服需要确认客户是否已经拒收或者退回商品。仓库收到退货并完成质检后，财务部门在三个工作日内完成退款。\n\n退款金额超过5000元时，需要部门负责人和财务负责人双重审批。普通退款由客服主管审批即可。",
      "score": 0.69
    }
  ]
}

十四、核心代码逻辑解析

这个AI搜索系统虽然代码不复杂，但已经包含了RAG应用的核心链路。

1. 文档进入系统

文档首先被读取为纯文本。无论原始文件是Markdown、PDF还是数据库记录，本质上都需要转成可处理的文本。

2. 文本切片

切片的目的是让召回粒度更精准。如果把整份制度文档作为一个向量，用户问“病假证明”时，可能会召回整篇制度，噪声较多；切成小段后，系统更容易定位到相关片段。

3. 向量化

Embedding模型将文本转成向量。向量不是人类可读内容，但可以表达文本语义。两个句子越相似，其向量距离通常越近。

例如：

员工想休年假要提前多久申请？
年假申请需要提前三个工作日在OA提交。

这两句话关键词并不完全一致，但语义接近，因此向量相似度会较高。

4. 向量检索

FAISS负责高效地从大量向量中找到最相似的Top K个文档片段。FAISS是本地向量检索库，适合原型和中小规模场景。如果是生产环境，也可以使用Milvus、Qdrant、Weaviate、Elasticsearch向量检索或云厂商向量数据库。

5. 大模型生成答案

向量检索只能找到相关片段，而大模型负责把这些片段组织成自然语言答案。它不是凭空回答，而是基于召回的上下文进行总结、归纳和表达。

十五、为什么AI搜索比传统搜索更适合知识库？

企业知识库通常有几个特点：

第一，用户问题表达不统一。同一个问题，员工可能有很多种问法。例如“报销流程是什么”“费用怎么走审批”“发票交给谁”可能都在问同一类问题。

第二，文档结构复杂。制度、流程、手册、通知往往分散在不同系统中，传统关键词搜索需要用户自己判断哪份文档有用。

第三，用户要的是答案，不是文档列表。传统搜索返回一堆链接，用户还要逐个打开阅读；AI搜索可以直接给出答案，并附带来源。

第四，知识更新频繁。企业流程经常变化，通过RAG模式可以只更新知识库索引，不必每次重新训练模型。

十六、生产环境优化建议

上面的Demo适合学习和原型验证。如果要真正上线，还需要做很多工程化优化。

1. 引入重排序模型

向量召回Top K之后，可以使用Reranker模型重新排序，提高最终上下文质量。

常见流程：

用户问题
  ↓
向量召回 Top 20
  ↓
Reranker重排序
  ↓
选择 Top 5
  ↓
大模型生成答案

向量检索更像粗筛，Reranker更像精排。

2. 增加关键词混合检索

纯向量检索有时对编号、专有名词、金额、日期等精确信息不敏感。生产环境可以采用混合检索：

向量检索 + BM25关键词检索 + Rerank

例如用户搜索“Q3-2024-REFUND-001”，关键词检索可能比向量检索更准确。

3. 增加引用来源

为了增强可信度，回答时应展示答案来源，例如：

答案：退款金额超过5000元时，需要部门负责人和财务负责人双重审批。

来源：
1. refund_process.md，第1段

这样用户可以追溯原文，减少误用风险。

4. 做权限控制

企业知识库往往包含敏感信息，不同员工能访问的文档范围不同。因此检索时必须加入权限过滤：

用户身份 → 可访问文档集合 → 在权限范围内检索

不能先全库检索再展示答案，否则可能造成数据泄露。

5. 建立索引更新机制

真实业务中文档会不断变化，需要支持：

• 新增文档自动入库
• 修改文档自动重建索引
• 删除文档同步删除向量
• 定时任务增量更新
• 文档版本管理

6. 增加答案质量评估

AI搜索不是上线后就结束了，还要持续评估效果。可以记录用户问题、召回片段、模型回答、用户反馈，并建立评估集。

常见指标包括：

• 召回准确率
• 答案正确率
• 答案引用准确率
• 无答案识别率
• 用户满意度
• 平均响应时间

十七、常见问题与解决方案

问题一：召回内容不相关

可能原因：

• 切片过大或过小
• Embedding模型效果不好
• 文档内容质量差
• 用户问题过于简略
• Top K设置不合理

解决方案：

• 调整chunk大小和overlap
• 更换中文效果更好的Embedding模型
• 使用混合检索
• 加入Reranker
• 对文档进行结构化清洗

问题二：模型回答编造

可能原因：

• 提示词约束不够
• 召回上下文不相关
• 模型温度过高
• 没有要求模型承认不知道

解决方案：

• 降低temperature
• 提示词中明确“只能依据资料回答”
• 无相关上下文时直接返回无法回答
• 增加引用校验
• 对低相似度结果设置阈值

问题三：响应速度慢

可能原因：

• 向量模型加载慢
• 大模型接口耗时高
• 检索数量过多
• 文档库过大

解决方案：

• 服务启动时预加载模型
• 使用更快的Embedding服务
• 缓存高频问题
• 使用流式输出
• 使用高性能向量数据库
• 对索引进行分片

十八、一个更健壮的检索阈值示例

实际业务中，不能无论召回结果质量如何都让大模型回答。可以增加相似度阈值，如果最高分低于阈值，则直接返回“没有找到相关资料”。

修改 /ask 接口：

@app.post("/ask")
def ask(req: SearchRequest):
    query_embedding = embedding_model.encode([req.question])

    scores, indices = index.search(query_embedding, req.top_k)

    contexts = []

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

        if float(score) < 0.45:
            continue

        doc = documents[idx]
        contexts.append({
            "source": doc["source"],
            "chunk_id": doc["chunk_id"],
            "content": doc["content"],
            "score": float(score)
        })

    if not contexts:
        return {
            "question": req.question,
            "answer": "根据现有资料无法确定，请补充相关文档或联系管理员。",
            "contexts": []
        }

    answer = generate_answer(req.question, contexts)

    return {
        "question": req.question,
        "answer": answer,
        "contexts": contexts
    }

这个小改动可以明显减少幻觉风险。

十九、案例总结

本文通过一个企业知识库问答案例，完整演示了AI搜索系统的基本实现流程：

文档加载
  ↓
文本切片
  ↓
Embedding向量化
  ↓
FAISS向量索引
  ↓
语义检索
  ↓
大模型生成答案
  ↓
API服务输出

从工程视角看，AI搜索并不是简单地“接一个大模型接口”，而是一套完整的数据处理、检索、生成和评估系统。大模型决定了答案表达能力，而知识库质量、切片策略、召回算法和权限体系，往往决定了系统能否真正落地。

如果只是做Demo，几十行代码就可以跑起来；但如果要服务真实企业场景，就必须关注文档质量、召回准确率、权限安全、答案可追溯、成本控制和持续评估。

AI搜索的价值不在于炫技，而在于让用户用自然语言快速获得可信答案。无论是企业知识库、客服问答、合同检索、产品手册查询，还是研发文档助手，都可以基于本文的思路扩展出完整应用。

二十、完整运行步骤汇总

最后将运行步骤整理如下：

# 1. 创建项目目录
mkdir ai-search-demo
cd ai-search-demo

# 2. 创建虚拟环境
python -m venv venv

# 3. 激活虚拟环境
source venv/bin/activate

# 4. 安装依赖
pip install fastapi uvicorn faiss-cpu sentence-transformers openai python-dotenv

# 5. 准备data目录和示例文档
mkdir data storage utils

# 6. 写入本文中的源码文件

# 7. 配置环境变量
export OPENAI_API_KEY="你的API Key"

# 8. 构建索引
python build_index.py

# 9. 启动服务
uvicorn app:app --reload --port 8000

# 10. 打开接口文档
# http://127.0.0.1:8000/docs

至此，一个完整的AI搜索实战Demo就搭建完成了。你可以继续扩展文档格式、接入企业权限系统、增加Rerank、引入混合检索，逐步把它从Demo演进为可用于生产环境的知识库搜索系统。