欢迎光临

汤不热吧技术社区引入 AI 知识库:基于 Milvus 向量数据库与 LangChain 的 RAG 语义搜索系统搭建全记录

随着社区文章数量突破 3000 篇,传统的基于 SQL LIKE 的全文搜索已经难以满足用户精准获取技术内容的需求。为此,汤不热吧技术社区近期完成了 AI 语义搜索系统的建设——基于 Milvus 向量数据库与 LangChain 框架,构建了一套完整的 RAG(Retrieval-Augmented Generation)知识库系统。本文详细记录整个技术方案的选型、架构设计、部署过程与性能调优。

AI知识库架构示意

一、为什么需要 RAG 语义搜索

传统的数据库全文索引(如 MySQL FULLTEXT、Elasticsearch BM25)主要依赖关键词匹配。对于技术问答场景,这种方式存在三个核心缺陷:

  • 同义词盲区:搜索「容器化部署」无法匹配「Docker 编排」
  • 语义鸿沟:「怎么让 Python 程序跑得更快」与「Python 性能优化技巧」本质是同一问题,但关键词完全不重叠
  • 长尾查询效果差:低频关键词组合(如「Kubernetes 上部署 Flink 算子」)很难命中已有文章

我们统计了过去 90 天的搜索日志:约有 34% 的搜索会话中用户发起了至少两次查询,说明首次搜索未能找到所需内容。RAG 系统通过向量嵌入将文本映射到高维语义空间,能理解查询的意图而非字面,从根本上解决了上述问题。

二、技术选型与架构设计

2.1 向量数据库选型对比

方案 部署复杂度 亿级规模性能 成本 社区活跃度
Milvus 中(Docker Compose) 优秀(毫秒级) 开源免费 高(34k+ Stars)
Qdrant 低(单二进制) 良好 开源免费 中(20k+ Stars)
Weaviate 中(Docker) 良好 开源免费 中(12k+ Stars)
Pinecone 极低(SaaS) 优秀 高(按量付费)

综合考虑社区规模(万级文档)、预算(零成本优先)和运维能力(有 Kubernetes 集群),我们选择 Milvus 2.4 作为向量数据库,部署在已有的 Kubernetes 集群上。

2.2 整体架构

系统分为三个层级:


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
┌─────────────────────────────────────────────────────┐
│                  用户查询入口                        │
│   (WordPress 搜索框 / 搜索 Widget / REST API)       │
└──────────┬──────────────────────────────────────────┘
           │
           ▼
┌─────────────────────────────────────────────────────┐
│               Query Pipeline (LangChain)             │
│                                                      │
│   1. 查询改写 ──→ 2. 向量化(embedding) ──→ 3. 检索   │
│                                              │      │
│   4. 重排序(Cross-encoder) ──→ 5. 生成回答          │
└──────────────────┬──────────────────────────────────┘
                   │
        ┌──────────┼──────────┐
        ▼          ▼          ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Milvus   │ │ PostgreSQL│ │ Redis    │
│ 向量库   │ │ 元数据   │ │ 缓存     │
└──────────┘ └──────────┘ └──────────┘

其中,Milvus 负责向量存储与 ANN 检索,PostgreSQL 存储文章元数据(标题、分类、标签等),Redis 缓存热门查询结果。

三、Milvus 在 Kubernetes 上的部署

3.1 Helm 安装

我们使用 Milvus Operator 进行 Kubernetes 部署,这是官方推荐的生产方案:


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 添加 Helm 仓库
helm repo add milvus https://milvus-io.github.io/milvus-helm/
helm repo update

# 创建命名空间
kubectl create namespace milvus

# 安装 Milvus 集群模式(3 个数据节点 + 3 个查询节点)
helm install tbr-milvus milvus/milvus \
  --namespace milvus \
  --set cluster.enabled=true \
  --set persistence.enabled=true \
  --set persistence.size=100Gi \
  --set messageChannel.name=pulsar \
  --set standalone.replicas=0 \
  --set dataNode.replicas=3 \
  --set queryNode.replicas=3 \
  --set indexNode.replicas=2 \
  --set resource.limits.memory=8Gi \
  --set resource.limits.cpu=4

安装完成后检查状态:


1
2
3
4
5
6
7
8
9
10
11
12
13
14
kubectl get pods -n milvus

# 预期输出:
# NAME                            READY   STATUS    RESTARTS
# tbr-milvus-datac-node-0         1/1     Running   0
# tbr-milvus-datac-node-1         1/1     Running   0
# tbr-milvus-datac-node-2         1/1     Running   0
# tbr-milvus-index-node-0         1/1     Running   0
# tbr-milvus-index-node-1         1/1     Running   0
# tbr-milvus-mixcoord-xxx          1/1     Running   0
# tbr-milvus-proxy-xxx             1/1     Running   0
# tbr-milvus-querynode-0          1/1     Running   0
# tbr-milvus-querynode-1          1/1     Running   0
# tbr-milvus-querynode-2          1/1     Running   0

3.2 性能调优关键参数

在索引构建中,我们采用了 IVF_FLAT 索引加上 Product Quantization(PQ)压缩:


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from pymilvus import CollectionSchema, FieldSchema, DataType, Collection

# 创建 Collection 并配置索引
index_params = {
    "metric_type": "IP",       # 内积距离,搭配归一化后等价于余弦相似度
    "index_type": "IVF_FLAT",
    "params": {
        "nlist": 1024,          # 聚类中心数,越大召回率越高但建索引越慢
        "nprobe": 16,           # 查询时搜索的聚类数,平衡速度与精度
        "pq_m": 32,             # PQ 压缩维度,减小内存占用
    }
}

collection.create_index(
    field_name="embedding",
    index_params=index_params
)

# 加载到内存(全量加载以获得最佳性能)
collection.load()

经过调优后,100 万条数据的 ANN 搜索延迟控制在 15ms 以内(P99),内存占用约 2.8GB。

四、Embedding 模型选择与文本分块策略

4.1 嵌入模型

我们测试了三款中文嵌入模型:

模型 维度 检索精度 (Recall@10) 延迟(单条) 是否免费
BAAI/bge-large-zh-v1.5 1024 92.3% 85ms
text-embedding-3-small (OpenAI) 512 90.1% 320ms (API) 付费
shibing624/text2vec-base-chinese 768 85.6% 45ms

最终选用 BGE-large-zh-v1.5,部署在本地 GPU 节点(1x NVIDIA T4)上:


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 使用 FlagEmbedding 库加载模型
from FlagEmbedding import FlagModel

model = FlagModel(
    'BAAI/bge-large-zh-v1.5',
    query_instruction_for_retrieval="为这个句子生成以检索的嵌入:",
    use_fp16=True,
    device='cuda:0'
)

def embed_texts(texts: list[str]) -> list[list[float]]:
    embeddings = model.encode(texts)
    # L2 归一化,用于 IP 距离
    embeddings = embeddings / np.linalg.norm(embeddings, axis=1, keepdims=True)
    return embeddings.tolist()

4.2 文本分块策略

文章分块是 RAG 系统的关键环节。分块过大会包含噪声,过小则丢失上下文。我们采用递归滑动窗口分块:


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,          # 每个块 512 tokens
    chunk_overlap=128,       # 重叠 128 tokens 保持上下文连贯
    separators=["\n## ", "\n### ", "\n\n", "\n", "。", "!", "?", ",", " ", ""],
    length_function=len,
)

# 预处理所有文章
for article_id, title, content in fetch_all_articles():
    # 合并标题和正文
    full_text = f"## {title}\n\n{content}"
    chunks = text_splitter.split_text(full_text)
   
    for i, chunk in enumerate(chunks):
        embedding = embed_texts([chunk])[0]
        insert_to_milvus(article_id, title, chunk, i, embedding)

最终我们为 3482 篇已有文章生成了约 28,000 个文本块,平均每篇文章被切分为 8 个块。

五、LangChain 检索管线搭建

5.1 混合检索策略

单一向量检索在某些场景下精度不够。我们实现了向量检索 + BM25 关键词检索的混合方案:


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_milvus import Milvus

# 向量检索器
vector_store = Milvus(
    embedding_function=model.encode,
    collection_name="tbr_articles",
    connection_args={"host": "tbr-milvus.milvus", "port": "19530"},
)

# BM25 检索器
bm25_retriever = BM25Retriever.from_texts(
    texts=[chunk["text"] for chunk in all_chunks],
    metadatas=[chunk["meta"] for chunk in all_chunks],
    k=5,
)

# 集成检索器(权重:向量 0.6, BM25 0.4)
ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_store.as_retriever(search_kwargs={"k": 10}), bm25_retriever],
    weights=[0.6, 0.4],
)

5.2 重排序(Re-ranking)

混合检索出的候选结果通过 Cross-encoder 精排,大幅提升最终排序质量:


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from sentence_transformers import CrossEncoder

reranker = CrossEncoder('BAAI/bge-reranker-v2-m3', device='cuda:0')

def rerank(query: str, candidates: list) -> list[dict]:
    """对候选结果进行重排序"""
    pairs = [[query, cand[0]] for cand in candidates]
   
    # Cross-encoder 打分,范围 0-1
    scores = reranker.predict(pairs, batch_size=16)
   
    # 按新分数降序排列
    results = []
    for idx in np.argsort(scores)[::-1]:
        results.append({
            "score": float(scores[idx]),
            "text": candidates[idx][0],
            "metadata": candidates[idx][2],
        })
    return results

引入重排序后,NDCG@10 从 0.78 提升至 0.91。

六、缓存与性能优化

6.1 Redis 多级缓存

热门查询的重复性很高。我们使用 Redis 实现了三级缓存:


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import hashlib
import json
import redis.asyncio as redis

r = redis.Redis(host='tbr-redis', port=6379, decode_responses=True)

async def cached_search(query: str, top_k: int = 5):
    query_key = hashlib.md5(query.encode()).hexdigest()
   
    # Level 1: 完全相同的查询(TTL: 1 小时)
    cached = await r.get(f"search:exact:{query_key}")
    if cached:
        return json.loads(cached)
   
    # Level 2: 语义相似查询(找过去相似的查询结果复用)
    similar = await find_similar_queries(query)
    if similar and similar["score"] > 0.95:
        await r.setex(f"search:exact:{query_key}", 3600, json.dumps(similar["results"]))
        return similar["results"]
   
    # 执行完整检索管线
    results = await full_search_pipeline(query, top_k)
   
    # Level 3: 缓存结果(TTL: 30 分钟)
    await r.setex(f"search:exact:{query_key}", 1800, json.dumps(results))
   
    return results

缓存命中率达到 41%,平均检索延迟从 680ms 降至 42ms。

6.2 异步写入

新文章发布后需要异步更新向量索引:


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# WordPress 文章发布钩子触发
# 通过 RabbitMQ 异步处理
def on_article_publish(article_id, title, content):
    # 放入消息队列,避免阻塞文章发布
    rabbitmq_client.publish(
        exchange="article_events",
        routing_key="article.published",
        body=json.dumps({
            "article_id": article_id,
            "title": title,
            "content": content,
            "action": "upsert",
        })
    )

# 消费者处理
def handle_article_event(ch, method, properties, body):
    event = json.loads(body)
    if event["action"] == "upsert":
        chunks = text_splitter.split_text(f"## {event['title']}\n\n{event['content']}")
        for i, chunk in enumerate(chunks):
            embedding = embed_texts([chunk])[0]
            # 先删除该文章的旧块,再插入新块
            collection.delete(f"article_id == {event['article_id']}")
            collection.insert([{
                "article_id": event["article_id"],
                "chunk_idx": i,
                "text": chunk,
                "embedding": embedding,
            }])

七、生产环境效果数据

系统上线后两周的运营数据:

  • 搜索成功率:搜索后点击文章的比例从 56% 提升至 83%
  • 单次搜索解决率:仅需一次搜索即找到目标文章的用户从 66% 提升至 79%
  • 平均搜索耗时:P50 延迟 38ms,P99 延迟 145ms
  • 日均处理请求:约 3200 次搜索请求
  • Top-5 命中率:87.3%(用户所需内容出现在前 5 条结果中)

值得一提的是,上线后社区用户的反馈非常积极。多位活跃用户在论坛反馈搜索体验有了质的飞跃、终于能搜到以前怎么找都找不到的老文章了。

八、后续规划

当前系统还有几个正在开发中的改进方向:

  1. 多模态检索:支持搜索代码截图中的代码片段(使用 ColPali 视觉模型)
  2. 个性化重排序:根据用户历史点击行为,动态调整排序权重
  3. 增量学习的 Embedding:微调嵌入模型以适应社区特有的技术术语和缩写
  4. 搜索引导:搜索框输入时自动补全技术关键词,辅助用户构造更好的查询

以上是汤不热吧 AI 知识库从选型到上线的完整技术方案。我们会持续开源社区的核心技术组件,欢迎在 GitHub 上关注我们的技术分享,也欢迎大家在社区搜索体验后留下反馈意见!


本文作者:汤不热吧技术团队 | 最后更新:2026 年 7 月

【本站文章皆为原创,未经允许不得转载】:汤不热吧 » 汤不热吧技术社区引入 AI 知识库:基于 Milvus 向量数据库与 LangChain 的 RAG 语义搜索系统搭建全记录
分享到: 更多 (0)