feat(rag): 混合检索融合 — Milvus 向量 + Bleve 全文 + RRF + DashScope rerank

检索从向量单路升级为混合：向量(Milvus) + 全文(Bleve BM25) → RRF 融合 → 可选 rerank(DashScope gte-rerank)。 - rag/bleve.go: Bleve 全文索引(内存,随 ingest 写入;kb 过滤)；ingest 同步写 Milvus+Bleve - rag/fuse.go: RRF(Reciprocal Rank Fusion, k=60, 按文本去重)融合多路排序 - rag/rerank.go: DashScope gte-rerank 客户端(可选,env 配置,失败降级 RRF) - rag/rag.go: Search 改混合(向量+全文→RRF→可选rerank→topK)；main 读 RERANK_* env - 验证: 全模块 build✓ + e2e PASS; live——入库写双索引；查'NATS'→全文精确命中#1+向量 →RRF NATS 排首(向量=4 全文=1)；接 DashScope gte-rerank(百炼 key 有权限)→relevance score 0.19 真重排；retriever 节点端到端→DeepSeek 答 Milvus - 边界: Neo4j 图路(GraphRAG,需实体抽取)推迟；Bleve 内存索引重启重建；rerank 走 env (TODO 同 embedding 搬控制面 kind=rerank) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-11 09:53:37 +08:00
parent 755032c1d8
commit 85a5c2c1e7
8 changed files with 311 additions and 56 deletions
@@ -0,0 +1,78 @@
+package rag
+
+import (
+	"fmt"
+	"hash/fnv"
+	"log"
+
+	"github.com/blevesearch/bleve/v2"
+	"github.com/blevesearch/bleve/v2/search/query"
+)
+
+// bleveStore 是全文（BM25）检索路。内存索引：随 ingest 写入，进程重启重建。
+// 真实生产应落盘(bleve.New(path,...))；此处内存优先求简。
+type bleveStore struct {
+	idx bleve.Index
+}
+
+func openBleve() *bleveStore {
+	idx, err := bleve.NewMemOnly(bleve.NewIndexMapping())
+	if err != nil {
+		log.Printf("[rag] bleve 初始化失败，全文路降级: %v", err)
+		return &bleveStore{}
+	}
+	return &bleveStore{idx: idx}
+}
+
+func (b *bleveStore) ready() bool { return b != nil && b.idx != nil }
+
+// index 把 (kb, texts) 写入全文索引（按 kb+文本哈希做幂等 ID）。
+func (b *bleveStore) index(kb string, texts []string) error {
+	if !b.ready() {
+		return nil
+	}
+	batch := b.idx.NewBatch()
+	for _, t := range texts {
+		id := fmt.Sprintf("%s:%x", kb, fnvHash(t))
+		if err := batch.Index(id, map[string]any{"text": t, "kb": kb}); err != nil {
+			return err
+		}
+	}
+	return b.idx.Batch(batch)
+}
+
+// search 全文检索（可按 kb 过滤），返回 BM25 排序的命中。
+func (b *bleveStore) search(kb, q string, topK int) []Hit {
+	if !b.ready() || q == "" {
+		return nil
+	}
+	mq := bleve.NewMatchQuery(q)
+	mq.SetField("text")
+	var qy query.Query = mq
+	if kb != "" {
+		tq := bleve.NewTermQuery(kb)
+		tq.SetField("kb")
+		qy = bleve.NewConjunctionQuery(mq, tq)
+	}
+	req := bleve.NewSearchRequest(qy)
+	req.Size = topK
+	req.Fields = []string{"text"}
+	res, err := b.idx.Search(req)
+	if err != nil {
+		return nil
+	}
+	var hits []Hit
+	for _, h := range res.Hits {
+		text, _ := h.Fields["text"].(string)
+		if text != "" {
+			hits = append(hits, Hit{Text: text, Score: float32(h.Score)})
+		}
+	}
+	return hits
+}
+
+func fnvHash(s string) uint64 {
+	h := fnv.New64a()
+	_, _ = h.Write([]byte(s))
+	return h.Sum64()
+}
@@ -0,0 +1,26 @@
+package rag
+
+import "sort"
+
+// rrfK 是 RRF 的平滑常数（业界常用 60）。
+const rrfK = 60.0
+
+// rrf 用 Reciprocal Rank Fusion 融合多路检索的排序列表，按文本去重。
+// 每路对一个文档的贡献 = 1/(k + 该路中的名次)；累加后重排。
+func rrf(lists [][]Hit, topK int) []Hit {
+	score := map[string]float64{}
+	for _, list := range lists {
+		for rank, h := range list {
+			score[h.Text] += 1.0 / (rrfK + float64(rank+1))
+		}
+	}
+	fused := make([]Hit, 0, len(score))
+	for text, s := range score {
+		fused = append(fused, Hit{Text: text, Score: float32(s)})
+	}
+	sort.Slice(fused, func(i, j int) bool { return fused[i].Score > fused[j].Score })
+	if topK > 0 && len(fused) > topK {
+		fused = fused[:topK]
+	}
+	return fused
+}
@@ -10,11 +10,14 @@ import (
 	"sync"
 )

-// Engine 聚合 embedding 与 Milvus，对外提供入库/检索。embedding 可热更新（控制面下发）。
+// Engine 聚合 embedding + Milvus(向量) + Bleve(全文) + RRF 融合 + 可选 rerank。
+// embedding 可热更新（控制面下发）。
 type Engine struct {
-	mu  sync.RWMutex
-	emb *embedClient
-	mv  *milvusStore
+	mu     sync.RWMutex
+	emb    *embedClient
+	mv     *milvusStore
+	bleve  *bleveStore
+	rerank *rerankClient
 }

 // SetEmbedding 热更新 embedding 配置（控制面变更时调用）。空配置=关闭向量检索。
@@ -36,8 +39,12 @@ func (e *Engine) embed() *embedClient {
 }

 // Open 建立 RAG 引擎。embedding 未配 / Milvus 连不上 → 降级（检索返回空，不阻断工具服务）。
-func Open(ctx context.Context, milvusAddr, embBase, embKey, embModel string) *Engine {
-	e := &Engine{}
+// rerank* 为空则不启用重排（融合结果直接返回）。
+func Open(ctx context.Context, milvusAddr, embBase, embKey, embModel, rerankBase, rerankKey, rerankModel string) *Engine {
+	e := &Engine{bleve: openBleve(), rerank: newRerankClient(rerankBase, rerankKey, rerankModel)}
+	if e.rerank.ready() {
+		log.Printf("[rag] rerank: %s model=%s", rerankBase, rerankModel)
+	}
 	if embBase != "" && embModel != "" {
 		e.SetEmbedding(embBase, embKey, embModel) // env 初值（控制面会覆盖）
 	} else {
@@ -74,10 +81,11 @@ func (e *Engine) Ingest(ctx context.Context, kb, text string) (int, error) {
 	if err := e.mv.insert(ctx, kb, chunks, vecs); err != nil {
 		return 0, err
 	}
+	_ = e.bleve.index(kb, chunks) // 同步写全文索引（失败不阻断向量入库）
 	return len(chunks), nil
 }

-// Search 向量化查询 → Milvus topK 检索。降级时返回空。
+// Search 混合检索：Milvus(向量) + Bleve(全文) → RRF 融合 → 可选 rerank → topK。降级时返回空。
 func (e *Engine) Search(ctx context.Context, kb, query string, topK int) ([]Hit, error) {
 	if !e.Ready() {
 		return nil, nil
@@ -85,11 +93,32 @@ func (e *Engine) Search(ctx context.Context, kb, query string, topK int) ([]Hit,
 	if topK <= 0 {
 		topK = 5
 	}
+	fanout := topK * 3
+
+	// 向量路
 	vecs, err := e.embed().Embed(ctx, []string{query})
 	if err != nil || len(vecs) == 0 {
 		return nil, err
 	}
-	return e.mv.search(ctx, kb, vecs[0], topK)
+	vecHits, _ := e.mv.search(ctx, kb, vecs[0], fanout)
+	// 全文路
+	ftHits := e.bleve.search(kb, query, fanout)
+	// RRF 融合（按文本去重）
+	cand := rrf([][]Hit{vecHits, ftHits}, fanout)
+	log.Printf("[rag] hybrid: 向量=%d 全文=%d → 融合=%d", len(vecHits), len(ftHits), len(cand))
+
+	// 可选 rerank：对融合候选重排取 topK
+	if e.rerank.ready() && len(cand) > 1 {
+		if rr, rerr := e.rerank.rerank(ctx, query, cand, topK); rerr == nil {
+			return rr, nil
+		} else {
+			log.Printf("[rag] rerank 降级（用 RRF 结果）: %v", rerr)
+		}
+	}
+	if len(cand) > topK {
+		cand = cand[:topK]
+	}
+	return cand, nil
 }

 func (e *Engine) Close() {
@@ -0,0 +1,85 @@
+package rag
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"sort"
+	"time"
+)
+
+// rerankClient 调用 DashScope 文本重排（gte-rerank）。可选阶段：未配则跳过。
+// DashScope 原生格式（非 OpenAI 兼容）：
+//
+//	POST {baseURL}  {model, input:{query, documents}, parameters:{top_n, return_documents:false}}
+//	resp {output:{results:[{index, relevance_score}]}}
+type rerankClient struct {
+	baseURL string
+	apiKey  string
+	model   string
+	hc      *http.Client
+}
+
+func newRerankClient(baseURL, apiKey, model string) *rerankClient {
+	if baseURL == "" || model == "" {
+		return nil
+	}
+	return &rerankClient{baseURL: baseURL, apiKey: apiKey, model: model, hc: &http.Client{Timeout: 20 * time.Second}}
+}
+
+func (r *rerankClient) ready() bool { return r != nil && r.baseURL != "" }
+
+// rerank 用重排模型对候选重新打分排序，返回前 topN。出错时返回原序（降级）。
+func (r *rerankClient) rerank(ctx context.Context, query string, hits []Hit, topN int) ([]Hit, error) {
+	docs := make([]string, len(hits))
+	for i, h := range hits {
+		docs[i] = h.Text
+	}
+	body, _ := json.Marshal(map[string]any{
+		"model": r.model,
+		"input": map[string]any{"query": query, "documents": docs},
+		"parameters": map[string]any{"top_n": topN, "return_documents": false},
+	})
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, r.baseURL, bytes.NewReader(body))
+	if err != nil {
+		return hits, err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	if r.apiKey != "" {
+		req.Header.Set("Authorization", "Bearer "+r.apiKey)
+	}
+	resp, err := r.hc.Do(req)
+	if err != nil {
+		return hits, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode >= 400 {
+		buf := new(bytes.Buffer)
+		_, _ = buf.ReadFrom(resp.Body)
+		return hits, fmt.Errorf("rerank http %d: %s", resp.StatusCode, buf.String())
+	}
+	var out struct {
+		Output struct {
+			Results []struct {
+				Index          int     `json:"index"`
+				RelevanceScore float32 `json:"relevance_score"`
+			} `json:"results"`
+		} `json:"output"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+		return hits, err
+	}
+	res := make([]Hit, 0, len(out.Output.Results))
+	for _, rr := range out.Output.Results {
+		if rr.Index >= 0 && rr.Index < len(hits) {
+			res = append(res, Hit{Text: hits[rr.Index].Text, Score: rr.RelevanceScore})
+		}
+	}
+	if len(res) == 0 {
+		return hits, nil
+	}
+	sort.Slice(res, func(i, j int) bool { return res[i].Score > res[j].Score })
+	return res, nil
+}