From 8fb2f6282b8644baaae861749d65ed99420248b3 Mon Sep 17 00:00:00 2001 From: AsterZephyr <2046084122@qq.com> Date: Thu, 13 Nov 2025 21:25:59 +0800 Subject: [PATCH] feat(rag): add Path Retriever, Simple Fusion and LLMLingua HTTP compression MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This commit enhances the RAG pipeline with three major features: 1. **Path Retriever (双路径稀疏检索)** - Implements path-based sparse retrieval for hierarchical document structures - Supports configurable path fields (know_path, file_path, etc.) - Enables dual sparse retrieval (BM25 + Path) with automatic fusion - Adds PathRetriever with BM25-weighted path field queries - Updates retrieval provider to classify path as sparse retrieval type 2. **Simple Fusion Strategy** - Adds SimpleFusionStrategy matching EasyRAG's HybridRetriever.fusion behavior - Merges results by document ID, keeping highest score per document - Supports configurable topK limit after fusion - Provides simple alternative to RRF for result merging 3. **LLMLingua HTTP Compression Integration** - Adds HTTPCompressor for external compression services (e.g., LLMLingua) - Extends pipeline.post.compress config with endpoint and headers support - Supports method: http or llmlingua for external service calls - Adds validation for HTTP compression endpoint requirements - Integrates HTTPCompressor into RAG client initialization Changes: - Add retriever/path.go: PathRetriever implementation - Add retriever/README_PATH_RETRIEVER.md: Path Retriever documentation - Add fusion/simple.go: SimpleFusionStrategy implementation - Update rag_client.go: Integrate Path Retriever and HTTPCompressor - Update retrieval/provider.go: Classify path as sparse retrieval - Update post/compress.go: Add HTTPCompressor with batch compression - Update config/pipeline.go: Add endpoint and headers to compress config - Update config/validation.go: Validate HTTP compression endpoint - Update server.go: Load HTTP compression configuration - Update README.md: Document new compression and retrieval features This enables dual sparse retrieval workflows and flexible external compression service integration, improving retrieval accuracy and context compression options. --- .../mcp-server/servers/rag/README.md | 11 +- .../mcp-server/servers/rag/config/pipeline.go | 8 +- .../servers/rag/config/validation.go | 9 + .../mcp-server/servers/rag/fusion/simple.go | 159 +++++++++++++ .../mcp-server/servers/rag/post/compress.go | 221 +++++++++++++++++- .../servers/rag/post/compress_test.go | 64 +++++ .../mcp-server/servers/rag/rag_client.go | 29 ++- .../servers/rag/retrieval/provider.go | 3 +- .../rag/retriever/README_PATH_RETRIEVER.md | 208 +++++++++++++++++ .../mcp-server/servers/rag/retriever/path.go | 172 ++++++++++++++ .../mcp-server/servers/rag/server.go | 11 + 11 files changed, 887 insertions(+), 8 deletions(-) create mode 100644 plugins/golang-filter/mcp-server/servers/rag/fusion/simple.go create mode 100644 plugins/golang-filter/mcp-server/servers/rag/retriever/README_PATH_RETRIEVER.md create mode 100644 plugins/golang-filter/mcp-server/servers/rag/retriever/path.go diff --git a/plugins/golang-filter/mcp-server/servers/rag/README.md b/plugins/golang-filter/mcp-server/servers/rag/README.md index 7cc1374b3..7047256d8 100644 --- a/plugins/golang-filter/mcp-server/servers/rag/README.md +++ b/plugins/golang-filter/mcp-server/servers/rag/README.md @@ -198,6 +198,16 @@ data: #### Embedding - **OpenAI 兼容** +### Pipeline Post 阶段 + +`pipeline.post` 用于控制检索后的 rerank 与上下文压缩流程： + +- `rerank`：支持 `provider=http` 通过 `endpoint` 调用外部重排服务，或使用 `llm/keyword/model` 内建策略。 +- `compress`：开启后可选择 `method`： + - `truncate`（默认）：按 `target_ratio` 截断文档。 + - `selective/summary/extraction`：依赖 `llm`，分别执行相关句抽取、摘要或句子提取。 + - `http`（或 `llmlingua`）：通过 `endpoint` 调用外部压缩服务（例如 LLMLingua 微服务）。可在 `headers` 中设置自定义请求头（如 `Authorization`）。服务返回的文档顺序会作为新的上下文顺序使用。 + #### Vector Database - **Milvus** @@ -324,4 +334,3 @@ Open your browser and navigate to http://localhost:8000 - diff --git a/plugins/golang-filter/mcp-server/servers/rag/config/pipeline.go b/plugins/golang-filter/mcp-server/servers/rag/config/pipeline.go index 6d0be8b3e..638990236 100644 --- a/plugins/golang-filter/mcp-server/servers/rag/config/pipeline.go +++ b/plugins/golang-filter/mcp-server/servers/rag/config/pipeline.go @@ -156,9 +156,11 @@ type PostConfig struct { APIKey string `json:"api_key,omitempty" yaml:"api_key,omitempty"` // For model-based reranker } `json:"rerank" yaml:"rerank"` Compress struct { - Enable bool `json:"enable,omitempty" yaml:"enable,omitempty"` - Method string `json:"method,omitempty" yaml:"method,omitempty"` - TargetRatio float64 `json:"target_ratio,omitempty" yaml:"target_ratio,omitempty"` + Enable bool `json:"enable,omitempty" yaml:"enable,omitempty"` + Method string `json:"method,omitempty" yaml:"method,omitempty"` + TargetRatio float64 `json:"target_ratio,omitempty" yaml:"target_ratio,omitempty"` + Endpoint string `json:"endpoint,omitempty" yaml:"endpoint,omitempty"` + Headers map[string]string `json:"headers,omitempty" yaml:"headers,omitempty"` } `json:"compress" yaml:"compress"` } diff --git a/plugins/golang-filter/mcp-server/servers/rag/config/validation.go b/plugins/golang-filter/mcp-server/servers/rag/config/validation.go index 8f4299c86..4aaf88343 100644 --- a/plugins/golang-filter/mcp-server/servers/rag/config/validation.go +++ b/plugins/golang-filter/mcp-server/servers/rag/config/validation.go @@ -247,6 +247,15 @@ func (c *Config) validatePipeline() ValidationErrors { Message: fmt.Sprintf("compress.target_ratio must be in [0, 1], got %.2f", c.Pipeline.Post.Compress.TargetRatio), }) } + method := strings.ToLower(c.Pipeline.Post.Compress.Method) + if method == "http" || method == "llmlingua" || method == "llm-lingua" { + if c.Pipeline.Post.Compress.Endpoint == "" { + errs = append(errs, ValidationError{ + Field: "pipeline.post.compress.endpoint", + Message: "endpoint is required when compress.method is http/llmlingua", + }) + } + } } } diff --git a/plugins/golang-filter/mcp-server/servers/rag/fusion/simple.go b/plugins/golang-filter/mcp-server/servers/rag/fusion/simple.go new file mode 100644 index 000000000..4e81ccd28 --- /dev/null +++ b/plugins/golang-filter/mcp-server/servers/rag/fusion/simple.go @@ -0,0 +1,159 @@ +package fusion + +import ( + "context" + "sort" + "strconv" + + "github.com/alibaba/higress/plugins/golang-filter/mcp-server/servers/rag/schema" +) + +// SimpleFusionStrategy implements a simple fusion method similar to EasyRAG's HybridRetriever.fusion. +// It merges results by document ID, keeping the highest score for each document, +// and optionally applies a topK limit. +type SimpleFusionStrategy struct { + TopK int // If > 0, limits the number of results after fusion +} + +// NewSimpleFusionStrategy creates a new simple fusion strategy. +func NewSimpleFusionStrategy(topK int) *SimpleFusionStrategy { + return &SimpleFusionStrategy{TopK: topK} +} + +// Fuse merges retriever results by keeping the highest score for each document ID. +// This is similar to EasyRAG's HybridRetriever.fusion() method. +func (s *SimpleFusionStrategy) Fuse(ctx context.Context, inputs []RetrieverResult, params map[string]any) ([]schema.SearchResult, error) { + if len(inputs) == 0 { + return []schema.SearchResult{}, nil + } + + // Extract topK from params if provided + topK := s.TopK + if v := simpleLookupInt(params, "topk"); v > 0 { + topK = v + } + if v := simpleLookupInt(params, "top_k"); v > 0 { + topK = v + } + + // Merge results by document ID, keeping the highest score + scores := make(map[string]schema.SearchResult) + for _, in := range inputs { + if len(in.Results) == 0 { + continue + } + for _, item := range in.Results { + id := item.Document.ID + if id == "" { + // Skip documents without ID + continue + } + + // Ensure metadata carries retriever information + if item.Document.Metadata == nil { + item.Document.Metadata = make(map[string]interface{}) + } + item.Document.Metadata["retriever_type"] = in.Retriever + if in.Provider != "" { + item.Document.Metadata["retriever_provider"] = in.Provider + } + + existing, ok := scores[id] + if !ok { + // First occurrence of this document + scores[id] = item + } else { + // Keep the document with the highest score + if item.Score > existing.Score { + scores[id] = item + } + } + } + } + + // Convert map to slice + out := make([]schema.SearchResult, 0, len(scores)) + for _, result := range scores { + out = append(out, result) + } + + // Sort by score descending + sort.Slice(out, func(i, j int) bool { + return out[i].Score > out[j].Score + }) + + // Apply topK limit if specified + if topK > 0 && len(out) > topK { + out = out[:topK] + } + + return out, nil +} + +// Name implements Strategy. +func (s *SimpleFusionStrategy) Name() string { return "simple" } + +// Fusion is a convenience function similar to EasyRAG's HybridRetriever.fusion(). +// It merges multiple result lists by keeping the highest score for each document. +func Fusion(lists [][]schema.SearchResult) []schema.SearchResult { + if len(lists) == 0 { + return []schema.SearchResult{} + } + + scores := make(map[string]schema.SearchResult) + for _, list := range lists { + for _, item := range list { + id := item.Document.ID + if id == "" { + continue + } + + existing, ok := scores[id] + if !ok { + scores[id] = item + } else { + if item.Score > existing.Score { + scores[id] = item + } + } + } + } + + out := make([]schema.SearchResult, 0, len(scores)) + for _, result := range scores { + out = append(out, result) + } + + sort.Slice(out, func(i, j int) bool { + return out[i].Score > out[j].Score + }) + + return out +} + +// simpleLookupInt is a helper function to extract int from params. +func simpleLookupInt(params map[string]any, key string) int { + if params == nil { + return 0 + } + switch v := params[key].(type) { + case int: + return v + case int32: + return int(v) + case int64: + return int(v) + case float64: + return int(v) + case float32: + return int(v) + case string: + if v == "" { + return 0 + } + if n, err := strconv.Atoi(v); err == nil { + return n + } + } + return 0 +} diff --git a/plugins/golang-filter/mcp-server/servers/rag/post/compress.go b/plugins/golang-filter/mcp-server/servers/rag/post/compress.go index fdd4f770d..a19e611f4 100644 --- a/plugins/golang-filter/mcp-server/servers/rag/post/compress.go +++ b/plugins/golang-filter/mcp-server/servers/rag/post/compress.go @@ -1,10 +1,14 @@ package post import ( + "bytes" "context" + "encoding/json" "fmt" + "net/http" "strings" + "github.com/alibaba/higress/plugins/golang-filter/mcp-server/servers/rag/common/httpx" "github.com/alibaba/higress/plugins/golang-filter/mcp-server/servers/rag/common/logger" "github.com/alibaba/higress/plugins/golang-filter/mcp-server/servers/rag/llm" "github.com/alibaba/higress/plugins/golang-filter/mcp-server/servers/rag/schema" @@ -356,6 +360,142 @@ func (e *ExtractionCompressor) BatchCompress(ctx context.Context, results []sche return compressed, nil } +// ================================================================================ +// 5. HTTP Compressor (External microservice, e.g., LLMLingua) +// ================================================================================ + +// HTTPCompressor delegates compression to an external HTTP service. +type HTTPCompressor struct { + Endpoint string + Client *httpx.Client + Headers map[string]string + TargetRatio float64 +} + +func (h *HTTPCompressor) Compress(ctx context.Context, text string, query string) (string, float64, error) { + if h.Endpoint == "" || text == "" { + return text, 0, nil + } + results := []schema.SearchResult{{ + Document: schema.Document{ + ID: "compress-single", + Content: text, + }, + }} + compressed, err := h.BatchCompress(ctx, results, query) + if err != nil || len(compressed) == 0 { + return text, 0, err + } + out := compressed[0].Document.Content + if out == "" { + return text, 0, err + } + return out, calculateCompressionRatio(text, out), nil +} + +func (h *HTTPCompressor) BatchCompress(ctx context.Context, results []schema.SearchResult, query string) ([]schema.SearchResult, error) { + if h.Endpoint == "" || len(results) == 0 { + return results, nil + } + + logger.Infof("HTTPCompressor: compressing %d documents via %s", len(results), h.Endpoint) + + req := httpCompressRequest{ + Query: query, + TargetRatio: h.TargetRatio, + Documents: make([]httpCompressDocument, 0, len(results)), + } + index := make(map[string]int, len(results)) + for i, result := range results { + docID := result.Document.ID + if docID == "" { + docID = fmt.Sprintf("compress-%d", i) + } + index[docID] = i + req.Documents = append(req.Documents, httpCompressDocument{ + ID: docID, + Text: result.Document.Content, + Metadata: result.Document.Metadata, + }) + } + + resp, err := h.doRequest(ctx, &req) + if err != nil { + logger.Warnf("HTTPCompressor: request failed: %v", err) + return results, err + } + if resp == nil || len(resp.Documents) == 0 { + logger.Warnf("HTTPCompressor: empty response, using original results") + return results, nil + } + + out := make([]schema.SearchResult, 0, len(resp.Documents)) + for _, doc := range resp.Documents { + idx, ok := index[doc.ID] + if !ok { + continue + } + item := results[idx] + if doc.Text != "" { + item.Document.Content = doc.Text + } + if doc.Metadata != nil { + if item.Document.Metadata == nil { + item.Document.Metadata = make(map[string]any, len(doc.Metadata)) + } + for k, v := range doc.Metadata { + item.Document.Metadata[k] = v + } + } + if doc.Score != 0 { + item.Score = doc.Score + } + out = append(out, item) + } + + if len(out) == 0 { + logger.Warnf("HTTPCompressor: no matching documents in response, using original results") + return results, nil + } + return out, nil +} + +func (h *HTTPCompressor) doRequest(ctx context.Context, payload *httpCompressRequest) (*httpCompressResponse, error) { + body, err := json.Marshal(payload) + if err != nil { + return nil, fmt.Errorf("http compressor marshal request: %w", err) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, h.Endpoint, bytes.NewReader(body)) + if err != nil { + return nil, fmt.Errorf("http compressor new request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + for k, v := range h.Headers { + req.Header.Set(k, v) + } + client := h.ensureClient() + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("http compressor request failed: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, fmt.Errorf("http compressor status %d", resp.StatusCode) + } + var result httpCompressResponse + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("http compressor decode response: %w", err) + } + return &result, nil +} + +func (h *HTTPCompressor) ensureClient() *httpx.Client { + if h.Client == nil { + h.Client = httpx.NewFromConfig(nil) + } + return h.Client +} + // ================================================================================ // Helper functions // ================================================================================ @@ -372,13 +512,92 @@ func calculateCompressionRatio(original, compressed string) float64 { return reduction } +type httpCompressRequest struct { + Query string `json:"query"` + TargetRatio float64 `json:"target_ratio,omitempty"` + Documents []httpCompressDocument `json:"documents"` +} + +type httpCompressDocument struct { + ID string `json:"id"` + Text string `json:"text"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +type httpCompressResponse struct { + Documents []httpCompressedDocument `json:"documents"` +} + +type httpCompressedDocument struct { + ID string `json:"id"` + Text string `json:"text"` + Metadata map[string]any `json:"metadata,omitempty"` + Score float64 `json:"score,omitempty"` +} + // ================================================================================ // Compressor Factory // ================================================================================ +// CompressorOption configures the HTTP/remote compressor factory. +type CompressorOption func(*compressorOptions) + +type compressorOptions struct { + endpoint string + headers map[string]string + client *httpx.Client +} + +// WithHTTPEndpoint sets the remote compressor endpoint. +func WithHTTPEndpoint(endpoint string) CompressorOption { + return func(opts *compressorOptions) { + opts.endpoint = endpoint + } +} + +// WithHTTPHeaders sets static headers (e.g., Authorization) for the HTTP compressor. +func WithHTTPHeaders(headers map[string]string) CompressorOption { + return func(opts *compressorOptions) { + if len(headers) == 0 { + return + } + if opts.headers == nil { + opts.headers = make(map[string]string, len(headers)) + } + for k, v := range headers { + opts.headers[k] = v + } + } +} + +// WithHTTPClient injects a custom httpx.Client. +func WithHTTPClient(client *httpx.Client) CompressorOption { + return func(opts *compressorOptions) { + opts.client = client + } +} + // NewCompressor creates a Compressor based on method and configuration -func NewCompressor(method string, targetRatio float64, llmProvider llm.Provider) Compressor { +func NewCompressor(method string, targetRatio float64, llmProvider llm.Provider, opts ...CompressorOption) Compressor { + options := compressorOptions{} + for _, opt := range opts { + if opt != nil { + opt(&options) + } + } + switch strings.ToLower(method) { + case "http", "llmlingua", "llm-lingua": + if options.endpoint == "" { + logger.Warnf("HTTP compression requires endpoint, falling back to truncate") + return &TruncateCompressor{TargetRatio: targetRatio} + } + return &HTTPCompressor{ + Endpoint: options.endpoint, + Client: options.client, + Headers: options.headers, + TargetRatio: targetRatio, + } case "selective": if llmProvider == nil { logger.Warnf("Selective compression requires LLM provider, falling back to truncate") diff --git a/plugins/golang-filter/mcp-server/servers/rag/post/compress_test.go b/plugins/golang-filter/mcp-server/servers/rag/post/compress_test.go index e19ad78ed..f3daecd16 100644 --- a/plugins/golang-filter/mcp-server/servers/rag/post/compress_test.go +++ b/plugins/golang-filter/mcp-server/servers/rag/post/compress_test.go @@ -2,6 +2,9 @@ package post import ( "context" + "encoding/json" + "net/http" + "net/http/httptest" "strings" "testing" @@ -217,6 +220,60 @@ func TestSelectiveCompressor_BatchCompress(t *testing.T) { } } +func TestHTTPCompressor_BatchCompress(t *testing.T) { + t.Helper() + + var seenHeader bool + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("X-Test") == "1" { + seenHeader = true + } + var req httpCompressRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Fatalf("failed to decode request: %v", err) + } + if req.TargetRatio != 0.3 { + t.Fatalf("expected target ratio 0.3, got %f", req.TargetRatio) + } + resp := httpCompressResponse{ + Documents: []httpCompressedDocument{ + {ID: req.Documents[0].ID, Text: "compressed-1"}, + {ID: req.Documents[1].ID, Text: "compressed-2"}, + }, + } + _ = json.NewEncoder(w).Encode(resp) + })) + defer server.Close() + + compressor := &HTTPCompressor{ + Endpoint: server.URL, + Headers: map[string]string{"X-Test": "1"}, + TargetRatio: 0.3, + } + + input := []schema.SearchResult{ + {Document: schema.Document{ID: "a", Content: "aaaa bbbb cccc"}}, + {Document: schema.Document{ID: "b", Content: "dddd eeee ffff"}}, + } + + output, err := compressor.BatchCompress(context.Background(), input, "query") + if err != nil { + t.Fatalf("BatchCompress failed: %v", err) + } + if !seenHeader { + t.Fatal("expected custom header to be forwarded") + } + if len(output) != 2 { + t.Fatalf("expected 2 documents, got %d", len(output)) + } + if output[0].Document.Content != "compressed-1" { + t.Fatalf("unexpected first doc content: %s", output[0].Document.Content) + } + if output[1].Document.Content != "compressed-2" { + t.Fatalf("unexpected second doc content: %s", output[1].Document.Content) + } +} + func TestBatchCompress_AllEmpty(t *testing.T) { mockProvider := &MockCompressorLLMProvider{ response: "", @@ -291,6 +348,13 @@ func TestNewCompressor_FallbackWithoutLLM(t *testing.T) { } } +func TestNewCompressor_HTTP(t *testing.T) { + compressor := NewCompressor("http", 0.5, nil, WithHTTPEndpoint("http://example.com")) + if _, ok := compressor.(*HTTPCompressor); !ok { + t.Error("Expected HTTPCompressor") + } +} + // ================================================================================ // Utility Tests // ================================================================================ diff --git a/plugins/golang-filter/mcp-server/servers/rag/rag_client.go b/plugins/golang-filter/mcp-server/servers/rag/rag_client.go index 6791ffac1..18093a5a8 100644 --- a/plugins/golang-filter/mcp-server/servers/rag/rag_client.go +++ b/plugins/golang-filter/mcp-server/servers/rag/rag_client.go @@ -134,7 +134,7 @@ func NewRAGClient(config *config.Config) (*RAGClient, error) { retrievers = append(retrievers, vectorRet) register(vectorRet, "vector", ragclient.config.VectorDB.Provider, "vector") - // Optional: add BM25 / Web retrievers from config + // Optional: add BM25 / Web / Path retrievers from config for _, rc := range ragclient.config.Pipeline.Retrievers { switch rc.Type { case "bm25": @@ -150,6 +150,21 @@ func NewRAGClient(config *config.Config) (*RAGClient, error) { } retrievers = append(retrievers, bm) register(bm, rc.Type, rc.Provider, rc.Params["name"]) + case "path": + // Path retriever for hierarchical document path-based retrieval + pathRet := &retriever.PathRetriever{ + Endpoint: rc.Params["endpoint"], + Index: rc.Params["index"], + Client: httpx.NewFromConfig(ragclient.config.Pipeline.HTTP), + PathField: rc.Params["path_field"], // e.g., "know_path", "file_path" + } + if tk := rc.Params["top_k"]; tk != "" { + if n, err := strconv.Atoi(tk); err == nil { + pathRet.MaxTopK = n + } + } + retrievers = append(retrievers, pathRet) + register(pathRet, rc.Type, rc.Provider, rc.Params["name"]) case "web": web := &retriever.WebSearchRetriever{ Provider: rc.Provider, @@ -346,7 +361,17 @@ func NewRAGClient(config *config.Config) (*RAGClient, error) { if targetRatio == 0 { targetRatio = 0.7 // Default ratio } - ragclient.compressor = post.NewCompressor(method, targetRatio, ragclient.llmProvider) + var compressorOpts []post.CompressorOption + if compressCfg.Endpoint != "" { + compressorOpts = append(compressorOpts, + post.WithHTTPEndpoint(compressCfg.Endpoint), + post.WithHTTPClient(httpx.NewFromConfig(ragclient.config.Pipeline.HTTP)), + ) + } + if len(compressCfg.Headers) > 0 { + compressorOpts = append(compressorOpts, post.WithHTTPHeaders(compressCfg.Headers)) + } + ragclient.compressor = post.NewCompressor(method, targetRatio, ragclient.llmProvider, compressorOpts...) } // Initialize Pre-Retrieve Provider if enabled diff --git a/plugins/golang-filter/mcp-server/servers/rag/retrieval/provider.go b/plugins/golang-filter/mcp-server/servers/rag/retrieval/provider.go index ca1e9b46a..3f4051af5 100644 --- a/plugins/golang-filter/mcp-server/servers/rag/retrieval/provider.go +++ b/plugins/golang-filter/mcp-server/servers/rag/retrieval/provider.go @@ -606,7 +606,8 @@ func variantKeyForRetriever(r retriever.Retriever) string { switch strings.ToLower(r.Type()) { case "vector": return "dense" - case "bm25": + case "bm25", "path": + // Both BM25 and Path retrievers are sparse retrieval methods return "sparse" case "web": return "web" diff --git a/plugins/golang-filter/mcp-server/servers/rag/retriever/README_PATH_RETRIEVER.md b/plugins/golang-filter/mcp-server/servers/rag/retriever/README_PATH_RETRIEVER.md new file mode 100644 index 000000000..5d6a00ddc --- /dev/null +++ b/plugins/golang-filter/mcp-server/servers/rag/retriever/README_PATH_RETRIEVER.md @@ -0,0 +1,208 @@ +# Path Retriever - 双路径稀疏检索 + +## 概述 + +Path Retriever 实现了基于文档路径的稀疏检索功能，与 BM25 Retriever 配合使用可以实现双路径稀疏检索（BM25 + Path Retriever），通过融合机制提升检索效果。 + +## 架构设计 + +### 1. Path Retriever 实现 + +Path Retriever 继承自统一的 `Retriever` 接口，与 BM25 Retriever 保持一致的架构： + +- **类型标识**: `"path"` +- **检索目标**: 文档的路径字段（如 `know_path`, `file_path` 等） +- **检索策略**: 使用 BM25 算法，但针对路径字段进行优化加权 + +### 2. 双路径融合机制 + +系统通过现有的 Fusion 机制自动处理双路径检索结果： + +- **RRF (Reciprocal Rank Fusion)**: 默认融合策略，对 BM25 和 Path 检索结果进行融合（类似 EasyRAG 的 `reciprocal_rank_fusion`） +- **Simple Fusion**: 简单融合策略，保留每个文档的最高分数（类似 EasyRAG 的 `HybridRetriever.fusion`） +- **Weighted Strategy**: 支持为不同检索器设置权重 +- **自动去重**: 基于文档 ID 自动去重和合并 + +### 3. 配置方式 + +在 `PipelineConfig.Retrievers` 中配置 Path Retriever： + +```yaml +pipeline: + retrievers: + # BM25 主检索器（内容检索） + - type: bm25 + provider: elasticsearch + params: + endpoint: "http://es:9200" + index: "rag_bm25" + top_k: "10" + name: "bm25_main" + + # Path 路径检索器（路径检索） + - type: path + provider: elasticsearch + params: + endpoint: "http://es:9200" + index: "rag_bm25" + path_field: "know_path" # 路径字段名，可选值: know_path, file_path, path, document_path + top_k: "10" + name: "path_retriever" +``` + +### 4. Retrieval Profile 配置 + +在 Retrieval Profile 中指定使用双路径检索： + +```yaml +retrieval_profiles: + - name: "dual_sparse_profile" + retrievers: + - "bm25_main" + - "path_retriever" + top_k: 10 + per_retriever_top_k: 10 + variant_budgets: + sparse: 20 # 为稀疏检索（包括BM25和Path）分配预算 +``` + +### 5. 工作原理 + +1. **并行检索**: BM25 和 Path Retriever 并行执行检索 +2. **结果融合**: 使用配置的 Fusion 策略（默认 RRF）合并结果 +3. **去重排序**: 基于文档 ID 去重，按融合分数排序 +4. **返回结果**: 返回 TopK 个融合后的结果 + +### 6. 路径字段支持 + +Path Retriever 支持以下路径字段： + +- `know_path`: 知识路径（默认） +- `file_path`: 文件路径 +- `path`: 通用路径 +- `document_path`: 文档路径 +- `metadata.*`: 元数据中的任意路径字段 + +### 7. 查询优化 + +Path Retriever 的查询策略： + +1. **路径字段优先**: 对路径字段设置更高的 boost（2.0） +2. **元数据路径**: 支持 `metadata.path_field` 格式（boost 1.5） +3. **内容回退**: 如果路径不匹配，回退到内容检索（boost 0.5） + +### 8. 与 EasyRAG 的对比 + +参考 EasyRAG 的实现方式： + +```python +# EasyRAG 中的实现 +self.sparse_retriever = BM25Retriever.from_defaults( + nodes=self.nodes, + embed_type=f_embed_type_2, # 内容检索 + ... +) +self.path_retriever = BM25Retriever.from_defaults( + nodes=self.nodes, + embed_type=5, # 路径检索 (know_path) + ... +) +node_with_scores = HybridRetriever.fusion([ + node_with_scores, + node_with_scores_path, +]) +``` + +我们的实现： + +- **架构一致性**: 复用现有的 Retriever 接口和 Fusion 机制 +- **配置驱动**: 通过配置文件灵活控制，无需代码修改 +- **可扩展性**: 支持多个 Path Retriever 实例，针对不同路径字段 + +## 使用示例 + +### 完整配置示例 + +```yaml +pipeline: + enable_hybrid: true + rrf_k: 60 + retrievers: + - type: bm25 + provider: elasticsearch + params: + endpoint: "http://elasticsearch:9200" + index: "knowledge_base" + top_k: "10" + name: "bm25_content" + + - type: path + provider: elasticsearch + params: + endpoint: "http://elasticsearch:9200" + index: "knowledge_base" + path_field: "know_path" + top_k: "10" + name: "path_knowledge" + + retrieval_profiles: + - name: "dual_sparse" + retrievers: + - "bm25_content" + - "path_knowledge" + top_k: 10 + per_retriever_top_k: 10 + variant_budgets: + sparse: 20 +``` + +### 预期效果 + +- **内容匹配**: BM25 Retriever 负责内容语义匹配 +- **路径匹配**: Path Retriever 负责文档结构/路径匹配 +- **融合提升**: 两者融合后能够同时捕获内容和结构信息，提升检索准确性 + +## 扩展性 + +### 添加新的路径字段 + +只需在配置中指定不同的 `path_field` 参数： + +```yaml +- type: path + params: + path_field: "custom_path_field" +``` + +### 多路径检索器 + +可以配置多个 Path Retriever，针对不同路径字段： + +```yaml +retrievers: + - type: path + params: + path_field: "know_path" + name: "path_knowledge" + - type: path + params: + path_field: "file_path" + name: "path_file" +``` + +然后在 Profile 中同时使用： + +```yaml +retrievers: + - "bm25_content" + - "path_knowledge" + - "path_file" +``` + +## 注意事项 + +1. **索引要求**: Elasticsearch 索引需要包含路径字段，并建立相应的索引 +2. **字段映射**: 确保路径字段在 Elasticsearch 中正确映射 +3. **性能考虑**: 双路径检索会增加检索时间，建议合理设置 `top_k` 和 `per_retriever_top_k` +4. **融合策略**: 根据实际效果调整 RRF 参数或使用 Weighted Strategy + diff --git a/plugins/golang-filter/mcp-server/servers/rag/retriever/path.go b/plugins/golang-filter/mcp-server/servers/rag/retriever/path.go new file mode 100644 index 000000000..1e525513c --- /dev/null +++ b/plugins/golang-filter/mcp-server/servers/rag/retriever/path.go @@ -0,0 +1,172 @@ +package retriever + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "net/url" + "path" + "strings" + + "github.com/alibaba/higress/plugins/golang-filter/mcp-server/servers/rag/common/httpx" + "github.com/alibaba/higress/plugins/golang-filter/mcp-server/servers/rag/schema" +) + +// PathRetriever queries an Elasticsearch-like backend using BM25 on path/knowledge path fields. +// It's similar to BM25Retriever but focuses on path-based retrieval for hierarchical document structures. +// Endpoint example: http://es:9200 +// Index example: rag_bm25 +type PathRetriever struct { + Endpoint string + Index string + Client *httpx.Client + MaxTopK int + // PathField specifies which metadata field to use for path retrieval + // Common values: "file_path", "know_path", "path", "document_path" + PathField string +} + +func (r *PathRetriever) Type() string { return "path" } + +type pathSearchRequest struct { + Size int `json:"size"` + Query map[string]interface{} `json:"query"` +} + +// Reuse esHit and esSearchResponse types from bm25.go for consistency +type pathHit struct { + ID string `json:"_id"` + Score float64 `json:"_score"` + Source map[string]interface{} `json:"_source"` +} +type pathHits struct { + Hits []pathHit `json:"hits"` +} +type pathSearchResponse struct { + Hits pathHits `json:"hits"` +} + +func (r *PathRetriever) Search(ctx context.Context, query string, topK int) ([]schema.SearchResult, error) { + if r.Endpoint == "" || r.Index == "" { + return []schema.SearchResult{}, nil + } + if topK <= 0 { + topK = 10 + } + if r.MaxTopK > 0 && r.MaxTopK < topK { + topK = r.MaxTopK + } + + // Determine path field to search + pathField := r.PathField + if pathField == "" { + // Default to common path field names + pathField = "know_path" + } + + // Build query targeting path fields with higher weight + q := pathSearchRequest{ + Size: topK, + Query: map[string]interface{}{ + "bool": map[string]interface{}{ + "should": []map[string]interface{}{ + { + "match": map[string]interface{}{ + pathField: map[string]interface{}{ + "query": query, + "boost": 2.0, // Higher weight for path field + }, + }, + }, + { + "match": map[string]interface{}{ + metadataField(pathField): map[string]interface{}{ + "query": query, + "boost": 1.5, + }, + }, + }, + // Fallback to content if path doesn't match + { + "match": map[string]interface{}{ + "content": map[string]interface{}{ + "query": query, + "boost": 0.5, + }, + }, + }, + }, + "minimum_should_match": 1, + }, + }, + } + + bs, err := json.Marshal(q) + if err != nil { + return nil, fmt.Errorf("path retriever encode query: %w", err) + } + // Build URL: {endpoint}/{index}/_search + u, err := url.Parse(r.Endpoint) + if err != nil { + return nil, err + } + u.Path = path.Join(u.Path, r.Index, "_search") + req, err := http.NewRequestWithContext(ctx, http.MethodPost, u.String(), bytes.NewReader(bs)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/json") + if r.Client == nil { + return nil, fmt.Errorf("path retriever http client not configured") + } + resp, err := r.Client.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, fmt.Errorf("path retriever http status %d", resp.StatusCode) + } + + var psr pathSearchResponse + if err := json.NewDecoder(resp.Body).Decode(&psr); err != nil { + return nil, err + } + + out := make([]schema.SearchResult, 0, len(psr.Hits.Hits)) + for _, h := range psr.Hits.Hits { + content := "" + if v, ok := h.Source["content"].(string); ok { + content = v + } + // fallback: if no content, try title or any other field + if content == "" { + if v, ok := h.Source["title"].(string); ok { + content = v + } + } + doc := schema.Document{ID: h.ID, Content: content, Metadata: h.Source} + out = append(out, schema.SearchResult{Document: doc, Score: h.Score}) + } + return out, nil +} + +func metadataField(field string) string { + field = strings.TrimSpace(field) + if field == "" { + return "metadata.know_path" + } + if strings.HasPrefix(field, "metadata.") { + return field + } + return "metadata." + strings.TrimPrefix(field, "metadata.") +} + +// ClientHTTP unwraps httpx.Client to stdlib http.Client via Do +func (r *PathRetriever) ClientHTTP() *http.Client { + return &http.Client{Transport: roundTripperFunc(func(req *http.Request) (*http.Response, error) { + return r.Client.Do(req) + })} +} diff --git a/plugins/golang-filter/mcp-server/servers/rag/server.go b/plugins/golang-filter/mcp-server/servers/rag/server.go index 5558e71f8..a5dab57cb 100644 --- a/plugins/golang-filter/mcp-server/servers/rag/server.go +++ b/plugins/golang-filter/mcp-server/servers/rag/server.go @@ -407,6 +407,17 @@ func (c *RAGConfig) ParseConfig(cfg map[string]any) error { if f, ok := cmp["target_ratio"].(float64); ok { pc.Post.Compress.TargetRatio = f } + if s, ok := cmp["endpoint"].(string); ok { + pc.Post.Compress.Endpoint = s + } + if hdrs, ok := cmp["headers"].(map[string]any); ok { + pc.Post.Compress.Headers = make(map[string]string, len(hdrs)) + for hk, hv := range hdrs { + if vs, ok := hv.(string); ok { + pc.Post.Compress.Headers[hk] = vs + } + } + } } }