更新consul部分

ragflow/client.go

@@ -5,6 +5,7 @@ import (
 	"net/url"
 	"strings"
 	"sync"
+	"sync/atomic"
 
 	commonHttp "gitee.com/red-future---jilin-g/common/http"
 	"github.com/gogf/gf/v2/errors/gerror"
@@ -23,27 +24,57 @@ func initClient() {
 		ctx := context.Background()
 
 		// 读取配置
-		baseURL, apiKey := loadConfig(ctx)
+		endpoints, apiKey := loadConfig(ctx)
 
 		// 如果配置不完整，跳过初始化
-		if baseURL == "" || apiKey == "" {
+		if len(endpoints) == 0 || apiKey == "" {
-			g.Log().Warning(ctx, "⚠️  RAGFlow 配置未找到，请在项目 config.yml 中添加 ragflow.base_url 和 ragflow.api_key")
+			g.Log().Warning(ctx, "⚠️  RAGFlow 配置未找到，请在 config.yml 中添加 ragflow.base_url 或在 Consul 中配置 ragflow.endpoints")
 			return
 		}
 
 		globalClient = &Client{
-			BaseURL: strings.TrimSuffix(baseURL, "/"),
+			Endpoints: endpoints,
-			APIKey:  apiKey,
+			APIKey:    apiKey,
 		}
 
-		g.Log().Infof(ctx, "✅ RAGFlow 客户端初始化成功: baseURL=%s", baseURL)
+		if len(endpoints) == 1 {
+			g.Log().Infof(ctx, "✅ RAGFlow 客户端初始化成功: endpoint=%s", endpoints[0])
+		} else {
+			g.Log().Infof(ctx, "✅ RAGFlow 客户端初始化成功: endpoints=%v (负载均衡)", endpoints)
+		}
 	})
 }
 
-// loadConfig 从配置文件加载 RAGFlow 配置
+// loadConfig 从配置加载 RAGFlow 配置（支持实例级配置）
-func loadConfig(ctx context.Context) (baseURL, apiKey string) {
+// 优先级：
-	baseURL = g.Cfg().MustGet(ctx, "ragflow.base_url", "").String()
+// 1. Consul实例级配置 ragflow.endpoints (数组)
+// 2. Consul全局配置 ragflow.endpoints (数组)
+// 3. config.yml的 ragflow.base_url (单个URL，向后兼容)
+func loadConfig(ctx context.Context) (endpoints []string, apiKey string) {
+	// 尝试从Consul读取endpoints（支持实例级配置）
+	// 注意：这里不能直接导入customerService/service包，会造成循环依赖
+	// 所以只能从config.yml读取，Consul配置需要在customerservice层面调用时传入
+
+	// 读取API Key
 	apiKey = g.Cfg().MustGet(ctx, "ragflow.api_key", "").String()
+
+	// 尝试读取endpoints数组（从config.yml或Consul同步的配置）
+	endpointsConfig := g.Cfg().MustGet(ctx, "ragflow.endpoints")
+	if !endpointsConfig.IsEmpty() {
+		endpoints = endpointsConfig.Strings()
+		// 去除尾部斜杠
+		for i := range endpoints {
+			endpoints[i] = strings.TrimSuffix(endpoints[i], "/")
+		}
+		return
+	}
+
+	// Fallback到单个base_url（向后兼容）
+	baseURL := g.Cfg().MustGet(ctx, "ragflow.base_url", "").String()
+	if baseURL != "" {
+		endpoints = []string{strings.TrimSuffix(baseURL, "/")}
+	}
+
 	return
 }
 
@@ -53,10 +84,24 @@ func GetGlobalClient() *Client {
 	return globalClient
 }
 
-// Client RAGFlow API 客户端
+// Client RAGFlow API 客户端（支持负载均衡）
 type Client struct {
-	BaseURL string
+	Endpoints    []string      // RAGFlow实例列表
-	APIKey  string
+	APIKey       string        // API密钥
+	currentIndex atomic.Uint64 // 当前轮询索引（原子操作）
+}
+
+// getNextEndpoint 获取下一个endpoint（轮询算法）
+func (c *Client) getNextEndpoint() string {
+	if len(c.Endpoints) == 0 {
+		return ""
+	}
+	if len(c.Endpoints) == 1 {
+		return c.Endpoints[0]
+	}
+	// 原子递增并取模，实现轮询
+	idx := c.currentIndex.Add(1) % uint64(len(c.Endpoints))
+	return c.Endpoints[idx]
 }
 
 // CommonResponse 通用响应结构
@@ -71,9 +116,13 @@ func (r *CommonResponse) IsSuccess() bool {
 	return r.Code == 0
 }
 
-// request 发送 HTTP 请求（使用统一的common/http包）
+// request 发送 HTTP 请求（使用统一的common/http包，支持负载均衡）
 func (c *Client) request(ctx context.Context, method, path string, body interface{}, result interface{}) (err error) {
-	fullURL := c.BaseURL + path
+	endpoint := c.getNextEndpoint()
+	if endpoint == "" {
+		return gerror.New("RAGFlow endpoints not configured")
+	}
+	fullURL := endpoint + path
 
 	headers := map[string]string{
 		"Authorization": "Bearer " + c.APIKey,

ragflow/document.go

@@ -162,8 +162,12 @@ func (c *Client) UploadDocumentFromText(ctx context.Context, datasetId, content,
 		filename = "document.txt"
 	}
 
-	// 构造URL
+	// 构造URL（使用负载均衡）
-	url := c.BaseURL + "/api/v1/datasets/" + datasetId + "/documents"
+	endpoint := c.getNextEndpoint()
+	if endpoint == "" {
+		return "", gerror.New("RAGFlow endpoints not configured")
+	}
+	url := endpoint + "/api/v1/datasets/" + datasetId + "/documents"
 
 	// 创建multipart writer
 	body := &bytes.Buffer{}

redis/types.go

@@ -38,14 +38,16 @@ type BatchStreamMessage struct {
 
 // ResponseStreamMessage RAGFlow 响应消息结构（MQ 消息）
 type ResponseStreamMessage struct {
-	UserId    string `json:"userId"`    // 用户ID
+	UserId      string `json:"userId"`                // 用户ID
-	Platform  string `json:"platform"`  // 平台标识
+	Platform    string `json:"platform"`              // 平台标识
-	TenantId  string `json:"tenantId"`  // 租户ID
+	TenantId    string `json:"tenantId"`              // 租户ID
-	Question  string `json:"question"`  // 用户问题
+	AccountId   string `json:"accountId,omitempty"`   // 账号ID
-	Content   string `json:"content"`   // RAGFlow 回复内容
+	AccountName string `json:"accountName,omitempty"` // 客服账号名称
-	SessionId string `json:"sessionId"` // RAGFlow Session ID
+	Question    string `json:"question"`              // 用户问题
-	Timestamp int64  `json:"timestamp"` // 时间戳（秒）
+	Content     string `json:"content"`               // RAGFlow 回复内容
-	MessageId string `json:"messageId"` // 原始消息ID
+	SessionId   string `json:"sessionId"`             // RAGFlow Session ID
+	Timestamp   int64  `json:"timestamp"`             // 时间戳（秒）
+	MessageId   string `json:"messageId"`             // 原始消息ID
 }
 
 // FollowUpMessage 追问消息结构（RabbitMQ 延时队列）