// openai.go - OpenAI 兼容 LLM 客户端实现 // Logan.Li // 复用 aiops 既有 LLM 调用模式(model.LLMConfig + resolveEndpoint + /chat/completions + Bearer)。 // 支持所有 OpenAI 兼容后端:openai / deepseek / qwen / moonshot / ollama / custom。 package llm import ( "bufio" "bytes" "context" "encoding/json" "fmt" "io" "net/http" "strings" "time" ) // Config LLM 客户端配置(endpoint 已由上层解析为完整 base url) // Logan.Li type Config struct { Protocol string // "openai" | "anthropic",由 client.go 工厂决定具体实现 Endpoint string // 完整 base url,如 https://api.deepseek.com/v1 APIKey string // Bearer token Model string // 模型名,如 deepseek-chat / gpt-4o-mini Temperature float64 // 采样温度 MaxTokens int // 单次最大输出 token ReasoningEnabled bool // 是否启用推理/Plan 模式(OpenAI o-series / DeepSeek R1 等) // Logan.Li Timeout time.Duration // HTTP 超时(0 表示使用默认 180s) } // OpenAIClient OpenAI 兼容客户端 // Logan.Li type OpenAIClient struct { cfg Config client *http.Client } // NewOpenAIClient 创建 OpenAI 兼容客户端 // Logan.Li // 顶层工厂在 client.go(按 Protocol 路由到 OpenAI / Anthropic 实现)。 func NewOpenAIClient(cfg Config) Client { timeout := cfg.Timeout if timeout <= 0 { timeout = 180 * time.Second } return &OpenAIClient{ cfg: cfg, client: &http.Client{Timeout: timeout}, } } // Name 返回客户端名称 // Logan.Li func (c *OpenAIClient) Name() string { return "openai-compatible(" + c.cfg.Model + ")" } // isOpenAIReasoningModel 判断是否为使用新推理参数的 OpenAI 模型。 // Logan.Li func isOpenAIReasoningModel(model string) bool { m := strings.ToLower(model) if index := strings.LastIndex(m, "/"); index >= 0 { m = m[index+1:] } return strings.HasPrefix(m, "o1") || strings.HasPrefix(m, "o3") || strings.HasPrefix(m, "o4") || strings.HasPrefix(m, "gpt-5") } // isGenericReasoningModel 判断是否为通用推理开关模型(DeepSeek R1 等) // Logan.Li func isGenericReasoningModel(model string) bool { m := strings.ToLower(model) return strings.Contains(m, "reasoning") || strings.Contains(m, "r1") || strings.Contains(m, "deepseek-reasoner") } // applyReasoningParams 根据模型类型应用 token 与 reasoning 参数。 // Logan.Li func applyReasoningParams(payload *openaiRequest, enabled bool) { if isOpenAIReasoningModel(payload.Model) { // 新版 OpenAI 推理模型使用 max_completion_tokens 且不接收 temperature。 // Logan.Li payload.MaxCompletionTokens = payload.MaxTokens payload.MaxTokens = 0 payload.Temperature = 0 if enabled { payload.ReasoningEffort = "medium" } return } if enabled && isGenericReasoningModel(payload.Model) { payload.Reasoning = true } } // openaiRequest OpenAI chat/completions 请求体 // Logan.Li type openaiRequest struct { Model string `json:"model"` Messages []Message `json:"messages"` Tools []Tool `json:"tools,omitempty"` ToolChoice string `json:"tool_choice,omitempty"` // "auto" 让模型自主决定 Temperature float64 `json:"temperature,omitempty"` MaxTokens int `json:"max_tokens,omitempty"` MaxCompletionTokens int `json:"max_completion_tokens,omitempty"` // 新版 OpenAI 推理模型的输出上限 // Logan.Li Stream bool `json:"stream,omitempty"` // 流式补全(Stream 方法置 true) ReasoningEffort string `json:"reasoning_effort,omitempty"` // OpenAI o-series 推理强度 // Logan.Li Reasoning bool `json:"reasoning,omitempty"` // DeepSeek 等部分平台推理开关 // Logan.Li } // openaiResponse OpenAI chat/completions 响应体 // Logan.Li type openaiResponse struct { Choices []struct { Message struct { Role string `json:"role"` Content string `json:"content"` ToolCalls []ToolCall `json:"tool_calls"` } `json:"message"` } `json:"choices"` Usage Usage `json:"usage"` Error *struct { Message string `json:"message"` Type string `json:"type"` } `json:"error,omitempty"` } // Complete 调用 OpenAI 兼容接口完成一轮补全 // Logan.Li // 返回 assistant 消息(含思考文本 content 与工具调用计划 tool_calls)+ token 统计。 func (c *OpenAIClient) Complete(ctx context.Context, req *CompletionRequest) (*CompletionResponse, error) { payload := openaiRequest{ Model: c.cfg.Model, Messages: req.Messages, Tools: req.Tools, ToolChoice: "auto", Temperature: c.cfg.Temperature, MaxTokens: c.cfg.MaxTokens, } if payload.MaxTokens <= 0 { payload.MaxTokens = 4096 } applyReasoningParams(&payload, req.ReasoningEnabled || c.cfg.ReasoningEnabled) body, err := json.Marshal(payload) if err != nil { return nil, fmt.Errorf("序列化请求失败: %w", err) } url := c.cfg.Endpoint + "/chat/completions" httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) if err != nil { return nil, fmt.Errorf("构造请求失败: %w", err) } httpReq.Header.Set("Content-Type", "application/json") httpReq.Header.Set("Authorization", "Bearer "+c.cfg.APIKey) resp, err := c.client.Do(httpReq) if err != nil { return nil, fmt.Errorf("LLM 连接失败: %w", err) } defer resp.Body.Close() raw, _ := io.ReadAll(resp.Body) if resp.StatusCode != http.StatusOK { return nil, fmt.Errorf("LLM 返回错误 (HTTP %d): %s", resp.StatusCode, extractErrorMessage(raw)) } var parsed openaiResponse if err := json.Unmarshal(raw, &parsed); err != nil { return nil, fmt.Errorf("解析 LLM 响应失败: %w", err) } if parsed.Error != nil && parsed.Error.Message != "" { return nil, fmt.Errorf("LLM 错误: %s", parsed.Error.Message) } if len(parsed.Choices) == 0 { return nil, fmt.Errorf("LLM 未返回任何结果") } choice := parsed.Choices[0].Message return &CompletionResponse{ Model: c.cfg.Model, Message: Message{ Role: choice.Role, Content: choice.Content, ToolCalls: choice.ToolCalls, }, Usage: parsed.Usage, }, nil } // Stream 流式补全(OpenAI 兼容 SSE),逐 token 回调 onDelta,返回聚合响应 // Logan.Li // 解析 data: {"choices":[{"delta":{"content":"..."}}]} 增量;[DONE] 结束;末块 usage 若有则取。 func (c *OpenAIClient) Stream(ctx context.Context, req *CompletionRequest, onDelta func(string)) (*CompletionResponse, error) { payload := openaiRequest{ Model: c.cfg.Model, Messages: req.Messages, Tools: req.Tools, ToolChoice: "auto", Temperature: c.cfg.Temperature, MaxTokens: c.cfg.MaxTokens, Stream: true, } if payload.MaxTokens <= 0 { payload.MaxTokens = 4096 } applyReasoningParams(&payload, req.ReasoningEnabled || c.cfg.ReasoningEnabled) body, err := json.Marshal(payload) if err != nil { return nil, fmt.Errorf("序列化请求失败: %w", err) } url := c.cfg.Endpoint + "/chat/completions" httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) if err != nil { return nil, fmt.Errorf("构造请求失败: %w", err) } httpReq.Header.Set("Content-Type", "application/json") httpReq.Header.Set("Authorization", "Bearer "+c.cfg.APIKey) httpReq.Header.Set("Accept", "text/event-stream") streamClient := &http.Client{Timeout: 0} // 流式由 ctx 控制超时 resp, err := streamClient.Do(httpReq) if err != nil { return nil, fmt.Errorf("LLM 连接失败: %w", err) } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { raw, _ := io.ReadAll(resp.Body) return nil, fmt.Errorf("LLM 返回错误 (HTTP %d): %s", resp.StatusCode, extractErrorMessage(raw)) } var contentBuilder strings.Builder var usage Usage reader := bufio.NewReader(resp.Body) for { line, readErr := reader.ReadString('\n') line = strings.TrimSpace(line) if line != "" && strings.HasPrefix(line, "data:") { data := strings.TrimSpace(strings.TrimPrefix(line, "data:")) if data == "[DONE]" { break } var chunk struct { Choices []struct { Delta struct { Content string `json:"content"` } `json:"delta"` } `json:"choices"` Usage *Usage `json:"usage"` } if json.Unmarshal([]byte(data), &chunk) == nil { if len(chunk.Choices) > 0 && chunk.Choices[0].Delta.Content != "" { delta := chunk.Choices[0].Delta.Content contentBuilder.WriteString(delta) if onDelta != nil { onDelta(delta) } } if chunk.Usage != nil { usage = *chunk.Usage } } } if readErr != nil { break } } return &CompletionResponse{ Model: c.cfg.Model, Message: Message{Role: "assistant", Content: contentBuilder.String()}, Usage: usage, }, nil } // extractErrorMessage 从上游错误响应提取 message 字段 // Logan.Li func extractErrorMessage(raw []byte) string { var m struct { Error struct { Message string `json:"message"` } `json:"error"` } if json.Unmarshal(raw, &m) == nil && m.Error.Message != "" { return m.Error.Message } return string(raw) }