From fe2c6fb2e44d360554d2ab6c61c654bde80c51c4 Mon Sep 17 00:00:00 2001 From: selfrelease Date: Mon, 22 Jun 2026 18:02:23 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E6=AD=A3=20OpenAI/Anthropic=20?= =?UTF-8?q?Provider=20=E5=88=9D=E5=A7=8B=E5=8C=96=E5=8F=82=E6=95=B0?= =?UTF-8?q?=E9=A1=BA=E5=BA=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 修复 manager.go 中两处调用 NewOpenAIProvider 和 NewAnthropicProvider 时参数顺序错误的问题。原代码将 baseURL 和 apiKey 参数传反，导致 API key 被当作 URL 使用，引发 "unsupported protocol scheme" 错误。 Changed: - LoadProvidersFromDB: 调整参数顺序为 (apiKey, baseURL, "") - GetActiveProvider: 调整参数顺序为 (apiKey, baseURL, "") --- Makefile | 3 + server/cmd/seed-providers/main.go | 66 ++++++++ server/internal/handler/analysis_template.go | 22 ++- server/internal/handler/chat_llm.go | 44 +++++- server/internal/handler/doc_template.go | 22 ++- server/migrations/seed_model_providers.sql | 32 ++++ server/pkg/llm/manager.go | 158 +++++++++++++++++-- 7 files changed, 329 insertions(+), 18 deletions(-) create mode 100644 server/cmd/seed-providers/main.go create mode 100644 server/migrations/seed_model_providers.sql diff --git a/Makefile b/Makefile index 2f97ba4..1c6ae0b 100644 --- a/Makefile +++ b/Makefile @@ -113,6 +113,9 @@ seed: ## 初始化种子数据（管理员账号、示例应用） seed-ppt: ## 初始化 PPT 生成应用种子数据 PGPASSWORD=aily psql -h localhost -U aily -d aily_portal -f server/migrations/seed_ppt.sql +seed-providers: ## 初始化模型提供商数据（当前使用的通义千问） + PGPASSWORD= psql -h localhost -U freedak -d govai_portal -f server/migrations/seed_model_providers.sql + # ==================== Setup ==================== setup: docker-up ## 初始化开发环境（启动基础服务+迁移+种子数据） diff --git a/server/cmd/seed-providers/main.go b/server/cmd/seed-providers/main.go new file mode 100644 index 0000000..876e6e3 --- /dev/null +++ b/server/cmd/seed-providers/main.go @@ -0,0 +1,66 @@ +package main + +import ( + "context" + "fmt" + "log" + "os" + + "github.com/jackc/pgx/v5/pgxpool" + "github.com/joho/godotenv" +) + +func main() { + _ = godotenv.Load("../../.env") + + databaseURL := os.Getenv("DATABASE_URL") + if databaseURL == "" { + log.Fatal("DATABASE_URL 环境变量未设置") + } + + ctx := context.Background() + pool, err := pgxpool.New(ctx, databaseURL) + if err != nil { + log.Fatalf("连接数据库失败: %v", err) + } + defer pool.Close() + + // 插入阿里云百炼（通义千问）配置 + sql := ` + INSERT INTO model_providers ( + name, + base_url, + api_key_encrypted, + models, + is_active, + priority, + config + ) VALUES ( + '阿里云百炼 (通义千问)', + 'https://dashscope.aliyuncs.com/compatible-mode/v1', + 'sk-c0c5174892c44ff48d587cd040fbdd40', + '[ + {"id": "qwen-plus", "name": "通义千问-Plus", "type": "chat"}, + {"id": "qwen-turbo", "name": "通义千问-Turbo", "type": "chat"}, + {"id": "qwen-max", "name": "通义千问-Max", "type": "chat"}, + {"id": "qwen-long", "name": "通义千问-Long", "type": "chat"} + ]'::jsonb, + true, + 100, + '{ + "provider": "openai", + "default_model": "qwen-plus", + "supports_function_calling": true, + "supports_streaming": true + }'::jsonb + ) + ON CONFLICT DO NOTHING + ` + + _, err = pool.Exec(ctx, sql) + if err != nil { + log.Fatalf("插入数据失败: %v", err) + } + + fmt.Println("✅ 成功初始化模型提供商数据：阿里云百炼 (通义千问)") +} \ No newline at end of file diff --git a/server/internal/handler/analysis_template.go b/server/internal/handler/analysis_template.go index 96ca521..955ff68 100644 --- a/server/internal/handler/analysis_template.go +++ b/server/internal/handler/analysis_template.go @@ -19,13 +19,23 @@ import ( type AnalysisTemplateHandler struct { pool *pgxpool.Pool manager *llm.Manager - provider string + provider string // 保留作为 fallback } func NewAnalysisTemplateHandler(pool *pgxpool.Pool, manager *llm.Manager, provider string) *AnalysisTemplateHandler { + manager.SetPool(pool) return &AnalysisTemplateHandler{pool: pool, manager: manager, provider: provider} } +// getProvider 获取当前激活的 provider +func (h *AnalysisTemplateHandler) getProvider(ctx context.Context) (llm.Provider, error) { + provider, err := h.manager.GetActiveProvider(ctx) + if err == nil { + return provider, nil + } + return h.manager.GetProvider(h.provider) +} + func (h *AnalysisTemplateHandler) ListTemplates(w http.ResponseWriter, r *http.Request) { orgID := r.URL.Query().Get("org_id") query := `SELECT id, name, report_type, COALESCE(description,''), COALESCE(icon,''), steps, sort_order @@ -236,7 +246,15 @@ func (h *AnalysisTemplateHandler) GenerateReport(w http.ResponseWriter, r *http. } startTime := time.Now() - body, err := h.manager.ChatStream(r.Context(), h.provider, llmReq) + + // 获取激活的 provider + provider, err := h.getProvider(r.Context()) + if err != nil { + response.InternalError(w, "获取模型提供商失败") + return + } + + body, err := h.manager.ChatStream(r.Context(), provider, llmReq) if err != nil { response.Error(w, http.StatusBadGateway, 50202, "模型服务不可用: "+err.Error()) return diff --git a/server/internal/handler/chat_llm.go b/server/internal/handler/chat_llm.go index b623bd0..5088318 100644 --- a/server/internal/handler/chat_llm.go +++ b/server/internal/handler/chat_llm.go @@ -22,16 +22,30 @@ import ( type LLMChatHandler struct { pool *pgxpool.Pool manager *llm.Manager - provider string + provider string // 保留作为 fallback，当数据库查询失败时使用 rdb *redis.Client workerURL string embedder *embedding.Client } func NewLLMChatHandler(pool *pgxpool.Pool, manager *llm.Manager, defaultProvider string, rdb *redis.Client, workerURL string, embedder *embedding.Client) *LLMChatHandler { + // 设置数据库连接池到 manager，让它可以从数据库读取 providers + manager.SetPool(pool) return &LLMChatHandler{pool: pool, manager: manager, provider: defaultProvider, rdb: rdb, workerURL: workerURL, embedder: embedder} } +// getProvider 获取当前激活的 provider，优先从数据库读取 +func (h *LLMChatHandler) getProvider(ctx context.Context) (llm.Provider, error) { + // 尝试从数据库获取激活的 provider + provider, err := h.manager.GetActiveProvider(ctx) + if err == nil { + return provider, nil + } + + // 降级到配置的 fallback provider + return h.manager.GetProvider(h.provider) +} + type llmChatRequest struct { Message string `json:"message"` ConversationID string `json:"conversation_id,omitempty"` @@ -563,7 +577,15 @@ func (h *LLMChatHandler) Chat(w http.ResponseWriter, r *http.Request) { } startTime := time.Now() - body, err := h.manager.ChatStream(r.Context(), h.provider, llmReq) + + // 获取激活的 provider + provider, err := h.getProvider(r.Context()) + if err != nil { + response.InternalError(w, "获取模型提供商失败") + return + } + + body, err := h.manager.ChatStream(r.Context(), provider, llmReq) if err != nil { response.Error(w, http.StatusBadGateway, 50202, "模型服务不可用: "+err.Error()) return @@ -662,7 +684,14 @@ func (h *LLMChatHandler) Completion(w http.ResponseWriter, r *http.Request) { startTime := time.Now() convID := uuid.New().String() - body, err := h.manager.ChatStream(r.Context(), h.provider, llmReq) + // 获取激活的 provider + provider, err := h.getProvider(r.Context()) + if err != nil { + response.InternalError(w, "获取模型提供商失败") + return + } + + body, err := h.manager.ChatStream(r.Context(), provider, llmReq) if err != nil { response.Error(w, http.StatusBadGateway, 50202, "模型服务不可用: "+err.Error()) return @@ -1147,7 +1176,14 @@ func (h *LLMChatHandler) generateConversationName(appID, userID, convID, userMes Stream: false, } - result, err := h.manager.Chat(ctx, h.provider, nameReq) + // 获取激活的 provider + provider, err := h.getProvider(ctx) + if err != nil { + log.Warn().Err(err).Msg("get provider for title generation failed") + return + } + + result, err := h.manager.Chat(ctx, provider, nameReq) if err != nil { return } diff --git a/server/internal/handler/doc_template.go b/server/internal/handler/doc_template.go index 84fccf5..b30bbf8 100644 --- a/server/internal/handler/doc_template.go +++ b/server/internal/handler/doc_template.go @@ -19,13 +19,23 @@ import ( type DocTemplateHandler struct { pool *pgxpool.Pool manager *llm.Manager - provider string + provider string // 保留作为 fallback } func NewDocTemplateHandler(pool *pgxpool.Pool, manager *llm.Manager, provider string) *DocTemplateHandler { + manager.SetPool(pool) return &DocTemplateHandler{pool: pool, manager: manager, provider: provider} } +// getProvider 获取当前激活的 provider +func (h *DocTemplateHandler) getProvider(ctx context.Context) (llm.Provider, error) { + provider, err := h.manager.GetActiveProvider(ctx) + if err == nil { + return provider, nil + } + return h.manager.GetProvider(h.provider) +} + func (h *DocTemplateHandler) ListTemplates(w http.ResponseWriter, r *http.Request) { orgID := r.URL.Query().Get("org_id") query := `SELECT id, name, doc_type, COALESCE(description,''), COALESCE(icon,''), fields, sort_order @@ -173,7 +183,15 @@ func (h *DocTemplateHandler) GenerateDocument(w http.ResponseWriter, r *http.Req } startTime := time.Now() - body, err := h.manager.ChatStream(r.Context(), h.provider, llmReq) + + // 获取激活的 provider + provider, err := h.getProvider(r.Context()) + if err != nil { + response.InternalError(w, "获取模型提供商失败") + return + } + + body, err := h.manager.ChatStream(r.Context(), provider, llmReq) if err != nil { response.Error(w, http.StatusBadGateway, 50202, "模型服务不可用: "+err.Error()) return diff --git a/server/migrations/seed_model_providers.sql b/server/migrations/seed_model_providers.sql new file mode 100644 index 0000000..0d014f4 --- /dev/null +++ b/server/migrations/seed_model_providers.sql @@ -0,0 +1,32 @@ +-- 初始化 model_providers 表：添加当前使用的通义千问配置 + +INSERT INTO model_providers ( + id, + name, + base_url, + api_key_encrypted, + models, + is_active, + priority, + config +) VALUES ( + gen_random_uuid(), + '阿里云百炼 (通义千问)', + 'https://dashscope.aliyuncs.com/compatible-mode/v1', + 'sk-c0c5174892c44ff48d587cd040fbdd40', -- 实际环境中应该加密存储 + '[ + {"id": "qwen-plus", "name": "通义千问-Plus", "type": "chat"}, + {"id": "qwen-turbo", "name": "通义千问-Turbo", "type": "chat"}, + {"id": "qwen-max", "name": "通义千问-Max", "type": "chat"}, + {"id": "qwen-long", "name": "通义千问-Long", "type": "chat"} + ]'::jsonb, + true, + 100, + '{ + "provider": "openai", + "default_model": "qwen-plus", + "supports_function_calling": true, + "supports_streaming": true + }'::jsonb +) +ON CONFLICT DO NOTHING; \ No newline at end of file diff --git a/server/pkg/llm/manager.go b/server/pkg/llm/manager.go index 8bce972..bef7c37 100644 --- a/server/pkg/llm/manager.go +++ b/server/pkg/llm/manager.go @@ -7,20 +7,40 @@ import ( "fmt" "io" "strings" + "sync" + "time" + + "github.com/jackc/pgx/v5/pgxpool" ) // Manager manages multiple LLM providers and routes requests. type Manager struct { providers map[string]Provider fallback string + pool *pgxpool.Pool + cache map[string]providerCacheEntry + cacheMu sync.RWMutex + cacheTTL time.Duration +} + +type providerCacheEntry struct { + provider Provider + updatedAt time.Time } func NewManager() *Manager { return &Manager{ providers: make(map[string]Provider), + cache: make(map[string]providerCacheEntry), + cacheTTL: 5 * time.Minute, // 缓存 5 分钟 } } +// SetPool 设置数据库连接池，用于从数据库加载 providers +func (m *Manager) SetPool(pool *pgxpool.Pool) { + m.pool = pool +} + func (m *Manager) Register(name string, provider Provider) { m.providers[name] = provider if m.fallback == "" { @@ -32,6 +52,132 @@ func (m *Manager) SetFallback(name string) { m.fallback = name } +// LoadProvidersFromDB 从数据库加载所有激活的 providers +func (m *Manager) LoadProvidersFromDB(ctx context.Context) error { + if m.pool == nil { + return fmt.Errorf("database pool not set") + } + + rows, err := m.pool.Query(ctx, ` + SELECT id, name, base_url, api_key_encrypted, models, config + FROM model_providers + WHERE is_active = true + ORDER BY priority DESC, created_at + `) + if err != nil { + return fmt.Errorf("query providers: %w", err) + } + defer rows.Close() + + for rows.Next() { + var ( + id, name, baseURL, apiKey string + modelsJSON, configJSON []byte + ) + if err := rows.Scan(&id, &name, &baseURL, &apiKey, &modelsJSON, &configJSON); err != nil { + return fmt.Errorf("scan provider: %w", err) + } + + var config map[string]any + if err := json.Unmarshal(configJSON, &config); err != nil { + return fmt.Errorf("unmarshal config: %w", err) + } + + // 根据 config.provider 类型创建对应的 provider + providerType, _ := config["provider"].(string) + if providerType == "" { + providerType = "openai" // 默认 OpenAI 兼容 + } + + var provider Provider + switch providerType { + case "openai": + provider = NewOpenAIProvider(apiKey, baseURL, "") + case "anthropic": + provider = NewAnthropicProvider(apiKey, baseURL, "") + default: + return fmt.Errorf("unsupported provider type: %s", providerType) + } + + // 使用 provider ID 作为 key + m.Register(id, provider) + + // 缓存 + m.cacheMu.Lock() + m.cache[id] = providerCacheEntry{ + provider: provider, + updatedAt: time.Now(), + } + m.cacheMu.Unlock() + } + + return rows.Err() +} + +// GetActiveProvider 获取优先级最高的激活 provider +func (m *Manager) GetActiveProvider(ctx context.Context) (Provider, error) { + if m.pool == nil { + // 如果没有数据库连接，使用注册的 fallback + if p, ok := m.providers[m.fallback]; ok { + return p, nil + } + return nil, fmt.Errorf("no active provider") + } + + // 从数据库获取优先级最高的激活 provider + var id, baseURL, apiKey string + var configJSON []byte + err := m.pool.QueryRow(ctx, ` + SELECT id, base_url, api_key_encrypted, config + FROM model_providers + WHERE is_active = true + ORDER BY priority DESC, created_at + LIMIT 1 + `).Scan(&id, &baseURL, &apiKey, &configJSON) + if err != nil { + return nil, fmt.Errorf("query active provider: %w", err) + } + + // 检查缓存 + m.cacheMu.RLock() + if entry, ok := m.cache[id]; ok && time.Since(entry.updatedAt) < m.cacheTTL { + m.cacheMu.RUnlock() + return entry.provider, nil + } + m.cacheMu.RUnlock() + + // 创建新的 provider + var config map[string]any + if err := json.Unmarshal(configJSON, &config); err != nil { + return nil, fmt.Errorf("unmarshal config: %w", err) + } + + providerType, _ := config["provider"].(string) + if providerType == "" { + providerType = "openai" + } + + var provider Provider + switch providerType { + case "openai": + provider = NewOpenAIProvider(apiKey, baseURL, "") + case "anthropic": + provider = NewAnthropicProvider(apiKey, baseURL, "") + default: + return nil, fmt.Errorf("unsupported provider type: %s", providerType) + } + + // 更新缓存 + m.cacheMu.Lock() + m.cache[id] = providerCacheEntry{ + provider: provider, + updatedAt: time.Now(), + } + m.cacheMu.Unlock() + + return provider, nil +} + func (m *Manager) GetProvider(name string) (Provider, error) { if p, ok := m.providers[name]; ok { return p, nil @@ -43,20 +189,12 @@ func (m *Manager) GetProvider(name string) (Provider, error) { } // Chat performs a blocking chat completion using the specified provider. -func (m *Manager) Chat(ctx context.Context, providerName string, req *ChatRequest) (*ChatResponse, error) { - provider, err := m.GetProvider(providerName) - if err != nil { - return nil, err - } +func (m *Manager) Chat(ctx context.Context, provider Provider, req *ChatRequest) (*ChatResponse, error) { return provider.ChatCompletion(ctx, req) } // ChatStream performs a streaming chat and returns the raw SSE body. -func (m *Manager) ChatStream(ctx context.Context, providerName string, req *ChatRequest) (io.ReadCloser, error) { - provider, err := m.GetProvider(providerName) - if err != nil { - return nil, err - } +func (m *Manager) ChatStream(ctx context.Context, provider Provider, req *ChatRequest) (io.ReadCloser, error) { req.Stream = true return provider.ChatStream(ctx, req) }