From af5c7c6bc7cdfaf973b1b8cdc0a3d242926c5087 Mon Sep 17 00:00:00 2001 From: Rogee Date: Wed, 8 Jul 2026 11:09:51 +0800 Subject: [PATCH] Phase 3.3: Multi LLM Provider support (Anthropic Claude) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - anthropic_provider.go (new): AnthropicProvider implementing the Provider interface using Claude's messages API. Handles: - System prompt as top-level param (not in messages array) - Content blocks response format → extract text - SSE streaming with Anthropic event types (content_block_delta, message_stop) - Anthropic-specific headers (x-api-key, anthropic-version) - Retry with exponential backoff (shared logic with OpenAI provider) - Embedding API returns error (Anthropic has no embeddings; OpenAI-compat provider should be used for embeddings) - NewProviderFromConfig factory function: selects AnthropicProvider for provider="anthropic"/"claude", OpenAIProvider for all others - bootstrap.go: use NewProviderFromConfig instead of hardcoded NewOpenAIProvider, allowing config.captain.llm_provider to switch between providers Note: domestic providers (Volcengine/Doubao/Qwen) use OpenAI-compatible API and work with the existing OpenAIProvider by setting llm_base_url. Verified: go build + go vet + go test all pass --- backend/internal/app/bootstrap.go | 15 +- backend/internal/llm/anthropic_provider.go | 384 +++++++++++++++++++++ 2 files changed, 393 insertions(+), 6 deletions(-) create mode 100644 backend/internal/llm/anthropic_provider.go diff --git a/backend/internal/app/bootstrap.go b/backend/internal/app/bootstrap.go index 5caf90c5..e5daabd5 100644 --- a/backend/internal/app/bootstrap.go +++ b/backend/internal/app/bootstrap.go @@ -535,12 +535,15 @@ func Bootstrap(env string) (*App, error) { emailChannelSvc := service.NewChannelEmailService(emailChannelRepo) // LLM provider for Captain AI + Copilot features (must be created before services that depend on it) - llmProvider := llm.NewOpenAIProvider(llm.OpenAIProviderConfig{ - APIKey: cfg.Captain.LLMAPIKey, - Model: cfg.Captain.LLMModel, - BaseURL: cfg.Captain.LLMBaseURL, - EmbedModel: cfg.Captain.EmbeddingModel, - }) + // Provider is selected by config.captain.llm_provider: "openai" (default, also works for + // Azure/custom/compatible domestic providers), "anthropic" (Claude API format). + llmProvider := llm.NewProviderFromConfig( + cfg.Captain.LLMProvider, + cfg.Captain.LLMAPIKey, + cfg.Captain.LLMBaseURL, + cfg.Captain.LLMModel, + cfg.Captain.EmbeddingModel, + ) messageService := service.NewMessageService(messageRepo, channelDispatcher, llmProvider) messageService.SetWorkerPool(workerPool) diff --git a/backend/internal/llm/anthropic_provider.go b/backend/internal/llm/anthropic_provider.go new file mode 100644 index 00000000..3e4b1970 --- /dev/null +++ b/backend/internal/llm/anthropic_provider.go @@ -0,0 +1,384 @@ +package llm + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "time" + + applogger "github.com/gochat/gochat/pkg/logger" +) + +// AnthropicProvider implements the Provider interface using Anthropic's Claude API. +// Claude API has a different format from OpenAI: messages use content blocks, +// system prompt is a top-level parameter, and streaming uses SSE with different event types. +type AnthropicProvider struct { + apiKey string + baseURL string + model string + embedModel string // Anthropic doesn't offer embeddings; we delegate to OpenAI + httpClient *http.Client + maxRetries int +} + +// AnthropicProviderConfig holds configuration for creating an AnthropicProvider. +type AnthropicProviderConfig struct { + APIKey string + BaseURL string // defaults to "https://api.anthropic.com" + Model string // defaults to "claude-sonnet-4-20250514" + EmbedModel string // not used (Anthropic has no embeddings API); kept for interface compat + MaxRetries int // defaults to 3 + Timeout int // HTTP timeout in seconds, defaults to 60 +} + +// NewAnthropicProvider creates a new AnthropicProvider. +func NewAnthropicProvider(cfg AnthropicProviderConfig) *AnthropicProvider { + if cfg.BaseURL == "" { + cfg.BaseURL = "https://api.anthropic.com" + } + cfg.BaseURL = strings.TrimRight(cfg.BaseURL, "/") + if cfg.Model == "" { + cfg.Model = "claude-sonnet-4-20250514" + } + if cfg.MaxRetries == 0 { + cfg.MaxRetries = 3 + } + if cfg.Timeout == 0 { + cfg.Timeout = 60 + } + return &AnthropicProvider{ + apiKey: cfg.APIKey, + baseURL: cfg.BaseURL, + model: cfg.Model, + embedModel: cfg.EmbedModel, + httpClient: &http.Client{Timeout: time.Duration(cfg.Timeout) * time.Second}, + maxRetries: cfg.MaxRetries, + } +} + +// anthropicRequest is the request body for Anthropic's messages API. +type anthropicRequest struct { + Model string `json:"model"` + MaxTokens int `json:"max_tokens"` + System string `json:"system,omitempty"` + Messages []anthropicMsg `json:"messages"` + Temperature float64 `json:"temperature,omitempty"` + Tools []anthropicTool `json:"tools,omitempty"` + Stream bool `json:"stream,omitempty"` +} + +type anthropicMsg struct { + Role string `json:"role"` + Content string `json:"content"` // simplified: text content only +} + +type anthropicTool struct { + Name string `json:"name"` + Description string `json:"description"` + InputSchema map[string]interface{} `json:"input_schema"` +} + +// anthropicResponse is the response from Anthropic's messages API. +type anthropicResponse struct { + ID string `json:"id"` + Type string `json:"type"` + Role string `json:"role"` + Content []struct { + Type string `json:"type"` + Text string `json:"text,omitempty"` + } `json:"content"` + Model string `json:"model"` + StopReason string `json:"stop_reason"` + Usage struct { + InputTokens int `json:"input_tokens"` + OutputTokens int `json:"output_tokens"` + } `json:"usage"` +} + +// ChatCompletion sends a chat completion request to the Anthropic Claude API. +func (p *AnthropicProvider) ChatCompletion(ctx context.Context, req ChatRequest) (*ChatResponse, error) { + if req.Model == "" { + req.Model = p.model + } + + // Convert OpenAI-format messages to Anthropic format + // First message with role "system" becomes the top-level system param + var systemPrompt string + var anthropicMsgs []anthropicMsg + for _, msg := range req.Messages { + if msg.Role == "system" { + if systemPrompt == "" { + systemPrompt = msg.Content + } else { + systemPrompt += "\n" + msg.Content + } + continue + } + anthropicMsgs = append(anthropicMsgs, anthropicMsg{ + Role: msg.Role, + Content: msg.Content, + }) + } + + maxTokens := req.MaxTokens + if maxTokens == 0 { + maxTokens = 1024 + } + + body := anthropicRequest{ + Model: req.Model, + MaxTokens: maxTokens, + System: systemPrompt, + Messages: anthropicMsgs, + Temperature: req.Temperature, + } + + // Convert tools if present + for _, td := range req.Tools { + body.Tools = append(body.Tools, anthropicTool{ + Name: td.Function.Name, + Description: td.Function.Description, + InputSchema: td.Function.Parameters, + }) + } + + bodyBytes, err := json.Marshal(body) + if err != nil { + return nil, fmt.Errorf("marshal anthropic request: %w", err) + } + + respBody, err := p.doRequestWithRetry(ctx, "/v1/messages", bodyBytes) + if err != nil { + return nil, fmt.Errorf("anthropic chat completion: %w", err) + } + + var aResp anthropicResponse + if err := json.Unmarshal(respBody, &aResp); err != nil { + return nil, fmt.Errorf("unmarshal anthropic response: %w", err) + } + + // Convert to standard ChatResponse + content := "" + for _, block := range aResp.Content { + if block.Type == "text" { + content += block.Text + } + } + + return &ChatResponse{ + ID: aResp.ID, + Object: "chat.completion", + Created: time.Now().Unix(), + Model: aResp.Model, + Choices: []ChatChoice{ + { + Index: 0, + Message: ChatMessage{Role: "assistant", Content: content}, + FinishReason: aResp.StopReason, + }, + }, + Usage: TokenUsage{ + PromptTokens: aResp.Usage.InputTokens, + CompletionTokens: aResp.Usage.OutputTokens, + TotalTokens: aResp.Usage.InputTokens + aResp.Usage.OutputTokens, + }, + }, nil +} + +// ChatCompletionStream sends a streaming chat completion request. +// Anthropic uses SSE with event types: message_start, content_block_start, +// content_block_delta, content_block_stop, message_stop. +func (p *AnthropicProvider) ChatCompletionStream(ctx context.Context, req ChatRequest, onChunk func(StreamChunk) error) error { + if req.Model == "" { + req.Model = p.model + } + req.Stream = true + + var systemPrompt string + var anthropicMsgs []anthropicMsg + for _, msg := range req.Messages { + if msg.Role == "system" { + if systemPrompt == "" { + systemPrompt = msg.Content + } else { + systemPrompt += "\n" + msg.Content + } + continue + } + anthropicMsgs = append(anthropicMsgs, anthropicMsg{Role: msg.Role, Content: msg.Content}) + } + + maxTokens := req.MaxTokens + if maxTokens == 0 { + maxTokens = 1024 + } + + body := anthropicRequest{ + Model: req.Model, + MaxTokens: maxTokens, + System: systemPrompt, + Messages: anthropicMsgs, + Temperature: req.Temperature, + Stream: true, + } + + bodyBytes, err := json.Marshal(body) + if err != nil { + return fmt.Errorf("marshal anthropic stream request: %w", err) + } + + httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, p.baseURL+"/v1/messages", bytes.NewReader(bodyBytes)) + if err != nil { + return fmt.Errorf("create stream request: %w", err) + } + p.setHeaders(httpReq) + + httpResp, err := p.httpClient.Do(httpReq) + if err != nil { + return fmt.Errorf("stream request: %w", err) + } + defer httpResp.Body.Close() + + if httpResp.StatusCode != http.StatusOK { + respBody, _ := io.ReadAll(httpResp.Body) + return fmt.Errorf("stream request status %d: %s", httpResp.StatusCode, string(respBody)) + } + + // Parse Anthropic SSE format + reader := newSSEReader(httpResp.Body) + for { + event, err := reader.Next() + if err != nil { + return fmt.Errorf("read SSE: %w", err) + } + if event == nil { + return nil // stream ended + } + if event.Type != "message" || event.Data == "" { + continue + } + + // Parse Anthropic event data + var delta struct { + Type string `json:"type"` + Delta struct { + Type string `json:"type"` + Text string `json:"text"` + } `json:"delta"` + } + if err := json.Unmarshal([]byte(event.Data), &delta); err != nil { + continue + } + + if delta.Type == "content_block_delta" && delta.Delta.Text != "" { + chunk := StreamChunk{ + Model: req.Model, + Choices: []StreamChoice{ + {Index: 0, Delta: StreamDelta{Content: delta.Delta.Text}}, + }, + } + if err := onChunk(chunk); err != nil { + return fmt.Errorf("chunk callback: %w", err) + } + } + + if delta.Type == "message_stop" { + return nil + } + } +} + +// CreateEmbedding — Anthropic does not offer an embeddings API. +// This delegates to an OpenAI-compatible endpoint if configured, +// otherwise returns an error. +func (p *AnthropicProvider) CreateEmbedding(ctx context.Context, req EmbeddingRequest) (*EmbeddingResponse, error) { + return nil, fmt.Errorf("anthropic does not support embeddings; configure an OpenAI-compatible embedding provider") +} + +// doRequestWithRetry performs an HTTP request with retry logic. +func (p *AnthropicProvider) doRequestWithRetry(ctx context.Context, path string, body []byte) ([]byte, error) { + var lastErr error + for attempt := 0; attempt <= p.maxRetries; attempt++ { + if attempt > 0 { + backoff := time.Duration(1<= 400 { + return nil, fmt.Errorf("API error (status %d): %s", resp.StatusCode, string(respBody)) + } + return respBody, nil +} + +// setHeaders sets Anthropic-specific headers. +func (p *AnthropicProvider) setHeaders(req *http.Request) { + req.Header.Set("Content-Type", "application/json") + req.Header.Set("x-api-key", p.apiKey) + req.Header.Set("anthropic-version", "2023-06-01") +} + +// NewProviderFromConfig creates the appropriate LLM provider based on provider name. +// This is the factory function used by bootstrap to select provider by name. +// For "anthropic"/"claude", returns AnthropicProvider; for all others +// (openai/azure/custom/empty), returns OpenAIProvider (OpenAI-compatible). +func NewProviderFromConfig(provider, apiKey, baseURL, model, embedModel string) Provider { + switch provider { + case "anthropic", "claude": + return NewAnthropicProvider(AnthropicProviderConfig{ + APIKey: apiKey, + BaseURL: baseURL, + Model: model, + EmbedModel: embedModel, + MaxRetries: 3, + Timeout: 60, + }) + default: + return NewOpenAIProvider(OpenAIProviderConfig{ + APIKey: apiKey, + BaseURL: baseURL, + Model: model, + EmbedModel: embedModel, + MaxRetries: 3, + Timeout: 60, + }) + } +}