sundynix/sundynix-agentix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(rag): 检索三路不再静默吞错 —— 逐路诊断 + 一路挂不拖垮全部

Browse Source 
排查"向量路为什么是空的"花了半小时，因为空就是空，没有任何线索。这次把
整条检索链上的静默降级一次清掉。

真 bug（不只是可观测性）：
  - kb_search 与 Search() 都拿 rag.Ready() 当总闸，而 Ready() 只代表"向量路
    可用"（embedding + Milvus）。全文(bleve)与图谱(Neo4j)根本不依赖它们，却
    被一并毙掉 → "模型配置没下发"表现为"整个知识库什么都搜不到"，还不报错。
    改为逐路判定，任一路可用就仍有召回。

不再吞错：
  - milvus.search 原先把 error 转成 nil,nil —— 检索失败与无召回彻底无法区分；
  - bleve.search / graph.search 出错直接回 nil，连日志都没有；
  - searchPaths 丢掉 embedding 的 error。
    三处改为如实返回，错误统一打日志。

逐路诊断（RouteDiag）：每路上报 ok/empty/disabled/error + 耗时 + 原因，经
kb_search 的 diag 参数（仅试验台传，生产调用返回值不变）→ gateway → 检索
试验台。界面上现在能直接看出"这一路没配置/报错了/确实没匹配"，不必翻日志。
内存兜底索引也会在 note 里点明"重启即清零"。

测试：3 组，覆盖"无 embedding 时全文仍可召回"、三种空的区分、内存索引提示。
把总闸加回去验证过第一条确实会红——测试能抓到这个回归，不是摆设。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Blizzard
2026-07-20 15:10:03 +08:00
parent 00ad6c8ac0
commit b26fe21408
10 changed files with 341 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sundynix-mcp-go/internal/rag/rag.go
+120 -16
View File
@@ -10,6 +10,7 @@ import (
"os"
"strconv"
"sync"
"time"
"github.com/sundynix/sundynix-shared/contract"
)
@@ -233,17 +234,17 @@ func itoa(n int) string {
return string(b)
}
// Search 混合检索：Milvus(向量) + Bleve(全文) → RRF 融合 → 可选 rerank → topK。降级时返回空。
// Search 混合检索：Milvus(向量) + Bleve(全文) + Neo4j(图谱) → RRF 融合 → 可选 rerank → topK。
// 注意这里**不再**用 Ready() 当总闸：Ready() 只代表"向量路可用"，而全文/图谱两路不依赖
// embedding 与 Milvus。以前一刀切返回空，导致 embedding 配置缺失时整个知识库像是"什么都搜不到"，
// 且无任何错误信息。现在各路独立判定，任一路可用就仍有召回（见 searchPaths 的 RouteDiag）。
func (e *Engine) Search(ctx context.Context, kb, query string, topK int) ([]Hit, error) {
if !e.Ready() {
return nil, nil
}
if topK <= 0 {
topK = 5
}
fanout := topK * 3
vecHits, ftHits, graphHits := e.searchPaths(ctx, kb, query, fanout)
vecHits, ftHits, graphHits, _ := e.searchPaths(ctx, kb, query, fanout)
// RRF 融合（三路，按文本去重）
cand := rrf([][]Hit{vecHits, ftHits, graphHits}, fanout)
log.Printf("[rag] hybrid: 向量=%d 全文=%d 图谱=%d → 融合=%d", len(vecHits), len(ftHits), len(graphHits), len(cand))
@@ -286,26 +287,129 @@ func (e *Engine) DeleteDoc(ctx context.Context, kb, fileID string) error {
return nil
}
// searchPaths 跑三路召回，返回各路命中（供混合融合与离线评测按单路对比）。
func (e *Engine) searchPaths(ctx context.Context, kb, query string, fanout int) (vec, ft, graph []Hit) {
if vecs, err := e.embed().Embed(ctx, []string{query}); err == nil && len(vecs) > 0 {
vec, _ = e.mv.search(ctx, kb, vecs[0], fanout)
// RouteDiag 是一路召回的诊断。存在的理由：三路里任何一路挂掉都**不会报错**，
// 只表现为召回变差——"这一路没配置"、"这一路报错了"、"这一路确实没匹配"
// 在结果上完全一样（都是空数组），运维无从分辨。检索试验台据此告诉人是哪一环坏了。
type RouteDiag struct {
Name string `json:"name"` // vector | fulltext | graph
Status string `json:"status"` // ok | empty | disabled | error
Hits int `json:"hits"` //
MS int64 `json:"ms"` // 该路耗时
Error string `json:"error,omitempty"` // status=error 时的原文
Note string `json:"note,omitempty"` // 给人看的解释
}
func diagOf(name string, hits []Hit, err error, disabled bool, note string, started time.Time) RouteDiag {
d := RouteDiag{Name: name, Hits: len(hits), MS: time.Since(started).Milliseconds(), Note: note}
switch {
case disabled:
d.Status = "disabled"
case err != nil:
d.Status, d.Error = "error", err.Error()
case len(hits) == 0:
d.Status = "empty"
default:
d.Status = "ok"
}
return d
}
// searchPaths 跑三路召回，返回各路命中 + 各路诊断（供混合融合、离线评测与检索试验台）。
// 任何一路失败都不阻断其它路，但失败会被如实记录并打日志——绝不静默当成"没召回"。
func (e *Engine) searchPaths(ctx context.Context, kb, query string, fanout int) (vec, ft, graph []Hit, diags []RouteDiag) {
// ── 向量路：embedding 与 Milvus 任一环节失败都要区分出来 ──
t := time.Now()
var vErr error
var vNote string
vDisabled := !e.embed().ready() || e.mv == nil
if vDisabled {
vNote = "embedding 未配置或 Milvus 未连接"
} else {
vecs, err := e.embed().Embed(ctx, []string{query})
switch {
case err != nil:
vErr, vNote = fmt.Errorf("embedding: %w", err), "查询向量化失败，向量路本次无贡献"
case len(vecs) == 0:
vErr, vNote = errors.New("embedding 返回空向量"), "向量化返回空结果"
default:
vec, vErr = e.mv.search(ctx, kb, vecs[0], fanout)
if vErr != nil {
vNote = "Milvus 检索失败"
} else if len(vec) == 0 {
vNote = "该知识库在 Milvus 中没有向量（未入库或集合被重建过）"
}
}
}
diags = append(diags, diagOf("vector", vec, vErr, vDisabled, vNote, t))
// ── 全文路 ──
t = time.Now()
ftDisabled := !e.bleve.ready()
var ftErr error
ftNote := ""
if ftDisabled {
ftNote = "全文索引未就绪"
} else {
ft, ftErr = e.bleve.search(kb, query, fanout)
if !e.bleve.persistent {
ftNote = "索引为内存兜底（重启已清零，历史文档需重新入库）"
}
}
diags = append(diags, diagOf("fulltext", ft, ftErr, ftDisabled, ftNote, t))
// ── 图谱路 ──
t = time.Now()
gDisabled := !e.graph.ready()
var gErr error
gNote := ""
if gDisabled {
gNote = "Neo4j 未连接或未配置"
} else {
graph, gErr = e.graph.search(ctx, kb, query, fanout)
}
diags = append(diags, diagOf("graph", graph, gErr, gDisabled, gNote, t))
for _, d := range diags {
if d.Status == "error" {
log.Printf("[rag] ⚠️ %s 路检索失败 kb=%s: %s", d.Name, kb, d.Error)
}
}
ft = e.bleve.search(kb, query, fanout)
graph = e.graph.search(ctx, kb, query, fanout)
return
}
// SearchByModeDiag 与 SearchByMode 同源，额外返回各路诊断（检索试验台用）。
// 试验台要回答的是"为什么这一路是空的"，光有命中数回答不了。
func (e *Engine) SearchByModeDiag(ctx context.Context, kb, query string, topK int, mode string) ([]Hit, []RouteDiag) {
if topK <= 0 {
topK = 5
}
fanout := topK * 3
vec, ft, graph, diags := e.searchPaths(ctx, kb, query, fanout)
var hits []Hit
switch mode {
case "vector":
hits = vec
case "fulltext":
hits = ft
case "graph":
hits = graph
default:
hits = rrf([][]Hit{vec, ft, graph}, fanout)
}
if len(hits) > topK {
hits = hits[:topK]
}
return hits, diags
}
// SearchByMode 按指定模式返回 topK（评测用，纯检索不 rerank，便于公平对比）。
// mode: vector|fulltext|graph|hybrid(RRF)。
func (e *Engine) SearchByMode(ctx context.Context, kb, query string, topK int, mode string) []Hit {
if !e.Ready() || topK <= 0 {
if topK <= 0 {
topK = 5
}
if topK <= 0 {
topK = 5
}
fanout := topK * 3
vec, ft, graph := e.searchPaths(ctx, kb, query, fanout)
vec, ft, graph, _ := e.searchPaths(ctx, kb, query, fanout)
var hits []Hit
switch mode {
case "vector":