aiops/backend/pkg/agentcli/llm/openai.go

// 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
	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 + ")" }

// 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"`
	Stream      bool      `json:"stream,omitempty"` // 流式补全（Stream 方法置 true）
}

// 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
	}
	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
	}
	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)
}