问题：关键词搜不到，向量搜不准
我们业务方有个内部故障知识库，存了1.2万篇故障处理文档，总量约2.3GB。之前用的是MySQL LIKE查询，一线运维根本搜不到东西——他们输入「数据库连接池爆了」，MySQL LIKE '%连接池%' 能匹配，但输入「应用连不上数据库」就一条都查不出来。
更麻烦的是，故障文档里全是「CPU打满」「OOM」「线程阻塞」这类黑话，每个人描述方式不一样。3月份故障复盘时统计过，检索不到有效文档导致平均故障定位时间从40分钟拉到2.5小时。老板拍板：上RAG。
我们技术栈是PHP 8.3 + Laravel 11，内部AI服务是Python 3.10的FastAPI。本文所有代码都是这两个环境里跑通的。
方案设计：三个层级逐步演进
先说明我们的结论：只做向量检索 = 薛定谔的召回。我们实测了三种方案：
方案A：纯向量检索
把文档切块后用BGE-M3转成向量，存Milvus，查询时也转向量，余弦相似度召回Top K。这个方案实现最快，两个晚上搞定。
测试结果（测试集500个真实查询）：
| 指标 | 数值 |
|---|---|
| Recall@10 | 31.2% |
| 平均响应时间 | 820ms |
问题很明显：故障文档里大量专业词汇在语义空间里离得远。比如「TPS暴跌」和「接口超时」在语义上相关，但向量距离远；同样，「JVM内存溢出」和「堆外内存泄漏」也是。纯向量检索漏召回严重。
方案B：向量 + BM25混合检索
保住关键词能力，同时引入语义能力。ES 8.11里同时跑BM25和向量检索（ES的knn search），各取Top 50，RRF融合。
测试结果：
| 指标 | 数值 |
|---|---|
| Recall@10 | 61.8% |
| 平均响应时间 | 1.2s |
召回提上来了，但还不够。问题出在RRF做的是无脑加权融合，没有考虑相关性分数。
方案C：混合检索 + 重排
在方案B基础上加了一路BGE-Reranker-V2-M3重排，粗排取Top 30，重排后取Top 8。
测试结果：
| 指标 | 数值 |
|---|---|
| Recall@10 | 91.7% |
| 平均响应时间 | 1.8s（粗排1.2s + 重排600ms） |
| 端到端LLM回答准确率 | 84.3%（人工标注100条） |
业界常说的「RAG上限在召回、重排拉高下限」，这次实测验证了。最终我们采用方案C。
完整代码实现
整体架构
├── backend-ai/ # Python FastAPI 服务
│ ├── app.py # RAG 主服务
│ ├── embedder.py # BGE-M3 嵌入
│ ├── reranker.py # 重排器
│ └── config.yaml # 配置
├── backend-php/ # Laravel 接入层
│ └── app/Services/RagService.php
└── scripts/
└── init_kb.py # 知识库初始化脚本配置文件（config.yaml）
embedding:
model: "BAAI/bge-m3"
dimension: 1024
device: "cuda:0"
batch_size: 64
milvus:
host: "10.0.3.11"
port: 19530
collection: "fault_kb_v2"
index_type: "HNSW"
metric_type: "COSINE"
nprobe: 16
elasticsearch:
host: "10.0.3.12:9200"
index: "fault_kb_bm25_v2"
reranker:
model: "BAAI/bge-reranker-v2-m3"
device: "cuda:1"
max_length: 512
rag:
top_k_bm25: 50
top_k_vector: 50
top_k_rerank: 8
chunk_size: 512
chunk_overlap: 64模型用的BAAI/bge-m3，embedding维度1024，支持8192长度。reranker用的v2-m3，这是目前开源里效果最好的中文重排模型之一。GPU是两张4090，一张跑embedding一张跑rerank。
初始化向量库
# scripts/init_kb.py
# 完整代码，直接run
import os
import sys
sys.path.append(os.path.dirname(__file__) + "/../backend-ai")
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import DirectoryLoader, TextLoader
from pymilvus import CollectionSchema, FieldSchema, DataType, Collection, connections
from embedder import Embedder
import yaml, time
with open("backend-ai/config.yaml", "r") as f:
cfg = yaml.safe_load(f)
# 1. 加载文档
loader = DirectoryLoader("/data/fault_docs/", glob="**/*.txt", loader_cls=TextLoader)
docs = loader.load()
print(f"加载文档 {len(docs)} 篇")
# 2. 分块：chunk_size=512, overlap=64
splitter = RecursiveCharacterTextSplitter(
chunk_size=cfg["rag"]["chunk_size"],
chunk_overlap=cfg["rag"]["chunk_overlap"],
separators=["\n\n", "\n", "。", "！", "？", "；", " "]
)
chunks = splitter.split_documents(docs)
print(f"切分为 {len(chunks)} 个chunk")
# 3. 批量embedding（分批，避免显存OOM）
embedder = Embedder(cfg["embedding"], cfg["embedding"]["device"])
batch_size = cfg["embedding"]["batch_size"]
all_vectors = []
all_texts = []
for i in range(0, len(chunks), batch_size):
batch = chunks[i:i+batch_size]
texts = [c.page_content for c in batch]
vectors = embedder.encode(texts)
all_vectors.extend(vectors)
all_texts.extend(texts)
if i % 1000 == 0:
print(f"已处理 {i}/{len(chunks)}")
# 每批处理后主动回收显存
import torch
torch.cuda.empty_cache()
# 4. 建Collection + 写入Milvus
connections.connect(host=cfg["milvus"]["host"], port=cfg["milvus"]["port"])
fields = [
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=8192),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=cfg["embedding"]["dimension"])
]
schema = CollectionSchema(fields=fields, enable_dynamic_field=True)
collection = Collection(cfg["milvus"]["collection"], schema)
# 创建HNSW索引
index_params = {
"index_type": "HNSW",
"metric_type": "COSINE",
"params": {"M": 16, "efConstruction": 200}
}
collection.create_index("embedding", index_params)
collection.load()
# 写入数据
data = [{"text": t, "embedding": v} for t, v in zip(all_texts, all_vectors)]
for i in range(0, len(data), 2000):
collection.insert(data[i:i+2000])
print(f"写入完成，共 {len(data)} 条")
collection.flush()Embedder实现
# backend-ai/embedder.py
from sentence_transformers import SentenceTransformer
import numpy as np
class Embedder:
def __init__(self, cfg, device="cuda:0"):
self.model = SentenceTransformer(cfg["model"], device=device)
self.model.eval()
# BGE模型开启指令微调，短查询加指令
self.query_instruction = "为这个句子生成表示以用于检索相关文章："
def encode(self, texts, is_query=False):
if is_query and self.query_instruction:
texts = [self.query_instruction + t for t in texts]
embeddings = self.model.encode(
texts,
normalize_embeddings=True,
show_progress_bar=False
)
return embeddings
def encode_query(self, query):
return self.encode([query], is_query=True)[0]一个关键细节：BGE-M3的query需要加指令前缀「为这个句子生成表示以用于检索相关文章：」，不加的话检索效果掉3-5个百分点，实测。
混合检索实现（核心）
# backend-ai/app.py 核心部分
import asyncio
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from elasticsearch import Elasticsearch
from pymilvus import Collection, connections
from embedder import Embedder
from reranker import Reranker
import yaml
import numpy as np
app = FastAPI()
# 加载配置
with open("config.yaml", "r") as f:
cfg = yaml.safe_load(f)
# 初始化
connections.connect(host=cfg["milvus"]["host"], port=cfg["milvus"]["port"])
milvus_col = Collection(cfg["milvus"]["collection"])
milvus_col.load()
es = Elasticsearch([cfg["elasticsearch"]["host"]], timeout=30)
embedder = Embedder(cfg["embedding"])
reranker = Reranker(cfg["reranker"])
class SearchReq(BaseModel):
query: str
top_k: int = 8
def bm25_search(query: str, size: int):
# ES BM25搜索
body = {
"query": {
"multi_match": {
"query": query,
"fields": ["text^3", "title^5"],
"type": "best_fields"
}
},
"size": size
}
resp = es.search(index=cfg["elasticsearch"]["index"], body=body)
return resp["hits"]["hits"]
def vector_search(query: str, size: int):
# Milvus向量搜索
query_vec = embedder.encode_query(query)
results = milvus_col.search(
data=[query_vec],
anns_field="embedding",
param={"metric_type": "COSINE", "params": {"nprobe": cfg["milvus"]["nprobe"]}},
limit=size,
output_fields=["text"]
)
hits = []
for hit in results[0]:
hits.append({
"_score": hit.score,
"_source": {"text": hit.entity.get("text")}
})
return hits
def rrf_fusion(bm25_hits, vector_hits, k=60):
# RRF融合：对两路结果做加权融合
scores = {}
docs = {}
for idx, hit in enumerate(bm25_hits):
doc_id = hit["_id"]
score = 1.0 / (k + idx + 1)
scores[doc_id] = scores.get(doc_id, 0) + score * 0.4
docs[doc_id] = hit["_source"]["text"]
for idx, hit in enumerate(vector_hits):
doc_id = f"vec_{idx}_{hash(hit['_source']['text'][:50])}"
score = 1.0 / (k + idx + 1)
scores[doc_id] = scores.get(doc_id, 0) + score * 0.6
docs[doc_id] = hit["_source"]["text"]
# 按RRF分数排序
sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True)
return [{"doc_id": doc_id, "text": docs[doc_id], "rrf_score": score}
for doc_id, score in sorted_docs[:cfg["rag"]["top_k_rerank"]]]
@app.post("/search")
async def search(req: SearchReq):
try:
# 1. 并行执行BM25和向量检索
bm25_task = asyncio.to_thread(bm25_search, req.query, cfg["rag"]["top_k_bm25"])
vec_task = asyncio.to_thread(vector_search, req.query, cfg["rag"]["top_k_vector"])
bm25_hits, vec_hits = await asyncio.gather(bm25_task, vec_task)
# 2. RRF融合粗排
candidates = rrf_fusion(bm25_hits, vec_hits)
# 3. 重排：BGE-Reranker
final_results = reranker.rerank(req.query, candidates, req.top_k)
return {
"results": final_results,
"total": len(final_results),
"source": "hybrid + rerank"
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))重排实现
# backend-ai/reranker.py
from sentence_transformers import CrossEncoder
import numpy as np
class Reranker:
def __init__(self, cfg):
self.model = CrossEncoder(
cfg["model"],
max_length=cfg["max_length"],
device=cfg["device"]
)
def rerank(self, query: str, candidates: list, top_k: int = 8):
# 构造(query, doc)对
pairs = [(query, c["text"]) for c in candidates]
# 推理相关性分数
scores = self.model.predict(
pairs,
batch_size=16,
show_progress_bar=False
)
# 按分数降序排列
results = []
for candidate, score in zip(candidates, scores):
results.append({
**candidate,
"rerank_score": float(score)
})
results.sort(key=lambda x: x["rerank_score"], reverse=True)
return results[:top_k]Laravel接入层
<?php
// backend-php/app/Services/RagService.php
namespace App\Services;
use Illuminate\Support\Facades\Http;
class RagService
{
private string $aiBaseUrl;
public function __construct()
{
// AI服务地址
$this->aiBaseUrl = config('services.ai.base_url', 'http://10.0.2.15:8000');
}
/**
* 搜索知识库
*/
public function search(string $query, int $topK = 8): array
{
$response = Http::timeout(5)
->post($this->aiBaseUrl . '/search', [
'query' => $query,
'top_k' => $topK
]);
if ($response->failed()) {
\Log::error('RAG search failed', [
'query' => $query,
'status' => $response->status(),
'body' => $response->body()
]);
return [];
}
return $response->json()['results'] ?? [];
}
/**
* 带上下文的问答（供LLM调用）
*/
public function askWithContext(string $question, array $chatHistory = []): array
{
// 1. 先检索
$docs = $this->search($question, 8);
if (empty($docs)) {
return ['answer' => null, 'context' => [], 'hit' => false];
}
// 2. 构造上下文
$context = '';
foreach ($docs as $i => $doc) {
$context .= "[文档{$i}]\n" . $doc['text'] . "\n\n";
}
// 3. 调LLM生成答案（这里是Ollama的API）
$llmResponse = Http::timeout(30)
->post(config('services.ollama.url') . '/api/chat', [
'model' => 'qwen2.5:14b',
'messages' => [
[
'role' => 'system',
'content' => '你是故障处理专家。请根据给定的资料回答问题，'
. '不要编造不存在的操作步骤。'
. '如果资料不相关，请说明。'
],
...$chatHistory,
[
'role' => 'user',
'content' => "资料：\n{$context}\n\n问题：{$question}\n"
. "回答要求：只用资料里的信息，注明操作时的注意事项。"
]
],
'stream' => false,
'temperature' => 0.2
]);
if ($llmResponse->failed()) {
return ['answer' => null, 'context' => $docs, 'hit' => false];
}
return [
'answer' => $llmResponse->json('message.content'),
'context' => $docs,
'hit' => true
];
}
}测试集评估脚本
# scripts/evaluate.py
# 500个真实查询的评估
import requests
import time
import json
TEST_SET = "/data/test_set.jsonl"
def load_test_set():
queries = []
with open(TEST_SET, "r") as f:
for line in f:
item = json.loads(line)
queries.append(item)
return queries
def evaluate():
test_set = load_test_set()
recall_total = 0
hit_total = 0
latency_total = 0
for item in test_set:
query = item["query"]
ground_truth_ids = set(item["relevant_doc_ids"])
start = time.time()
resp = requests.post("http://10.0.2.15:8000/search",
json={"query": query, "top_k": 10},
timeout=10)
latency = (time.time() - start) * 1000
latency_total += latency
results = resp.json()["results"]
retrieved_ids = set()
for r in results:
# 用文本内容匹配ground truth
for gt_id in ground_truth_ids:
if r["doc_id"] == gt_id:
retrieved_ids.add(gt_id)
if retrieved_ids:
hit_total += 1
recall = len(retrieved_ids) / len(ground_truth_ids) if ground_truth_ids else 0
recall_total += recall
if len(test_set) % 50 == 0:
print(f"处理到 {len(test_set)} 条...")
avg_recall = recall_total / len(test_set)
hit_rate = hit_total / len(test_set)
avg_latency = latency_total / len(test_set)
print(f"评估完成：")
print(f" Recall@10: {avg_recall:.1%}")
print(f" Hit Rate@10: {hit_rate:.1%}")
print(f" 平均耗时: {avg_latency:.0f}ms")
if __name__ == "__main__":
evaluate()效果数据：全链路对比
生产环境配置：PHP 8.3 + Laravel 11，AI服务Python 3.10 + FastAPI 0.110，Milvus 2.3.5，ES 8.11.2，GPU 4090 x2，CPU 32核。
测试集500条真实查询，数据规模：
| 指标 | 方案A 纯向量 | 方案B 混合检索 | 方案C 混合+重排 |
|---|---|---|---|
| Recall@10 | 31.2% | 61.8% | 91.7% |
| Hit Rate@10 | 45.6% | 78.9% | 94.3% |
| MRR@10 | 0.22 | 0.47 | 0.73 |
| 平均响应时间 | 820ms | 1.2s | 1.8s |
| P95响应时间 | 1.4s | 2.1s | 3.2s |
线上实际使用的反馈：一线运维搜「数据库连接失败排查」，前三条命中率91.7%。平均故障定位时间从2.5小时降到35分钟，3月份上线后到现在，故障定位时长降了76%。
资源占用方面：embedding服务显存占用7.2GB，reranker服务占用6.8GB。单张4090（24GB）可以同时跑两个服务，我们用了两张是因为QA环境还跑着embedding微调任务。CPU占用平均42%（主要ES和Milvus）。
每秒请求峰值：20 QPS，P95 3.2s。翻倍到40 QPS时显存没问题，主要是ES的查询变慢。瓶颈在ES，后面给ES加了2个数据节点解决。
这组数据和结论也验证了一个点：在知识库规模不大（1万-10万文档）时，拼的不是模型大小，而是检索策略是否做全。
避坑指南（都是真踩过的）
以下6个坑，每一个都浪费了我们至少1天时间。
坑1：忘记清空Milvus Collection，插入重复数据
初始化脚本跑完发现Milvus里文档数量和预期差2倍。查了半天，发现是测试时忘了drop collection，第二次跑脚本直接往同一个collection里插入了一遍。后来在init_kb.py里加了：
try:
collection = Collection(cfg["milvus"]["collection"])
collection.drop()
print("已删除旧collection")
except FileNotFoundError:
pass加了这段，初始化脚本才能重复执行。
坑2：BGE-M3的词向量 + 句向量没分开
BGE-M3支持稠密向量、稀疏向量、多向量三种。你用 SentenceTransformer.encode() 默认只返回稠密向量。但直接拿BGE-M3做检索，稀疏向量对专业词汇的匹配能力也很重要。更坑的是，有人写教程说要把三个向量concat起来，我们照做了，Milvus维度从1024变成4096，检索效果没变化，查询时延长了62%。
结论：稠密向量足够，稀疏向量交给BM25去做。别搞花活。
坑3：重排模型输入长度512不够用
故障文档chunk_size=512字，但实际chunk平均长度在700字左右（因为有重叠和分隔符）。BGE-Reranker-V2-M3默认max_length=512，超长直接截断，后面的内容没参与重排。后来把max_length调到1024，显存多了2.1GB，但重排效果提升4个百分点。
坑4：ES的BM25不适合中文关键词
ES默认分词器standard对中文是按字切分的，导致「数据库」被切成「数」「据」「库」，搜出来的全是噪音。必须在索引里配置IK分词器，否则BM25的效果约等于没有：
# 安装IK分词插件（ES 8.11.2）
./bin/elasticsearch-plugin install \
https://get.infini.cloud/elasticsearch/analysis-ik/8.11.2同时索引mapping里要指定analyzer：
{
"mappings": {
"properties": {
"text": {
"type": "text",
"analyzer": "ik_max_word",
"search_analyzer": "ik_smart"
}
}
}
}配了IK之后，BM25的Recall@10从35.1%升到54.3%。
坑5：Ollama改了模型目录，直接404
联调阶段Ollama的模型从CPU版换到GPU版，把OLLAMA_MODELS环境变量改了，Laravel调Ollama API直接404，查了半天是Ollama重新加载了模型，但API地址从11434换成了11435（因为有另一个Ollama实例占用）。问题出在我们把 services.ollama.url 写死在.env里，没有做健康检查。
解决办法是加一个失败重试：
// 在RagService里加一个健康检查 + 自动重试
private function callOllama(array $payload): array
{
$url = config('services.ollama.url');
try {
$resp = Http::timeout(30)->post($url . '/api/chat', $payload);
if ($resp->successful()) {
return $resp->json();
}
} catch (\Exception $e) {
\Log::warning('Ollama连接失败，尝试备用节点', [
'url' => $url,
'error' => $e->getMessage()
]);
}
// 备用节点
$backupUrl = config('services.ollama.backup_url');
$resp = Http::timeout(30)->post($backupUrl . '/api/chat', $payload);
return $resp->json();
}坑6：批量嵌入时分批大小没控制，OOM
第一次跑批量嵌入，batch_size设128，4090直接OOM。BGE-M3模型本身参数不到1GB，但编码1024维向量并做归一化时，中间计算量挺大。把batch_size调到32后稳定。这个值取决于你的GPU显存，配置里写清楚，别盲目照抄别人的参数。
附加坑：LLM上下文塞满5000字，回答质量反而烂
实测下来，给LLM塞8个文档，每个512字，总上下文约4500字。一开始我们贪心，retrieve了15个文档（约8000字），结果LLM回答的时候反而忽略了文档1-2的强相关内容。后来加了prompt策略：「优先关注文档0-2，其他文档仅作为补充」，回答质量才稳定下来。
总结
RAG不是「向量库存文档 + 查一下拼prompt」就完事的。我们这套系统里，要兼顾的关键点就三个：
- 粗排要混着来：BM25管关键词精确匹配，向量管语义相似。RRF融合时我们调了权重（BM25 0.4 / 向量 0.6），效果比五五开好。
- 重排是必须的：Reranker不是锦上添花。在混合检索后加一个rerank，召回率从61.8%拉到91.7%。这个提升不是模型更大，而是Reranker专门做相关性判断，比相似度计算精准得多。
- 中文场景必须处理分词：ES的standard分词器对中文是废的。不装IK，BM25就没法用。
代码都在上面，照着跑一遍，效果数据应该差不多。有具体问题可以留言，能答的尽量答。