pj0235-eai_agentplatform/eai_agentplatform/backend-go/internal/ai/llm.go

package ai

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strings"
	"time"

	"eai_agentplatform/backend/internal/config"
)

// ──────────────────────────────────────────────
// 基础类型
// ──────────────────────────────────────────────

// Message OpenAI 兼容消息
// Role: system / user / assistant / tool。
// 普通对话仅需 Role+Content；工具调用场景可带 ToolCalls（assistant 发起）
// 与 ToolCallID（回填 tool 结果时使用）。
type Message struct {
	Role       string     `json:"role"`
	Content    string     `json:"content"`
	ToolCalls  []ToolCall `json:"tool_calls,omitempty"`
	ToolCallID string     `json:"tool_call_id,omitempty"`
}

// ToolCall 模型发起的工具调用。
type ToolCall struct {
	ID       string      `json:"id"`
	Type     string      `json:"type,omitempty"`
	Function ToolFuncCall `json:"function"`
}

// ToolFuncCall 工具调用的函数名与参数（json.RawMessage 以保持参数原文）。
type ToolFuncCall struct {
	Name      string          `json:"name"`
	Arguments json.RawMessage `json:"arguments"`
}

// ToolSchema OpenAI 兼容 function tool 定义，随请求一并提交给 LLM。
type ToolSchema struct {
	Type     string `json:"type"` // "function"
	Function struct {
		Name        string      `json:"name"`
		Description string      `json:"description"`
		Parameters  interface{} `json:"parameters"`
	} `json:"function"`
}

// ChatResult 非流式调用结果
type ChatResult struct {
	Content      string
	Model        string
	Provider     string
	FinishReason string
	ToolCalls    []ToolCall
	Usage        struct {
		PromptTokens     int `json:"prompt_tokens"`
		CompletionTokens int `json:"completion_tokens"`
		TotalTokens      int `json:"total_tokens"`
	}
}

// ──────────────────────────────────────────────
// OpenAI 兼容 HTTP 客户端
// ──────────────────────────────────────────────

// Client 低层 HTTP 客户端（基于 RouteConfig）
type Client struct {
	route       *config.RouteConfig
	baseURL     string
	apiKey      string
	model       string
	maxTokens   int
	temperature float64
	hc          *http.Client
}

// Route 返回创建本客户端的路由（可能为 nil）。
func (c *Client) Route() *config.RouteConfig {
	return c.route
}

// defaultHTTPTimeout 默认 HTTP 整体超时；路由可用 timeout_seconds 单独覆盖
const defaultHTTPTimeout = 60 * time.Second

// NewClient 从 RouteConfig 创建客户端
func NewClient(aiRoute *config.RouteConfig) *Client {
	timeout := defaultHTTPTimeout
	if aiRoute.TimeoutSeconds > 0 {
		timeout = time.Duration(aiRoute.TimeoutSeconds) * time.Second
	}
	return &Client{
		route:       aiRoute,
		baseURL:     strings.TrimRight(aiRoute.BaseURL, "/"),
		apiKey:      aiRoute.APIKey,
		model:       aiRoute.Model,
		maxTokens:   aiRoute.MaxTokens,
		temperature: aiRoute.Temperature,
		hc:          &http.Client{Timeout: timeout},
	}
}

func (c *Client) url(path string) string {
	return c.baseURL + path
}

func (c *Client) headers() map[string]string {
	h := map[string]string{"Content-Type": "application/json"}
	if c.apiKey != "" {
		h["Authorization"] = "Bearer " + c.apiKey
	}
	return h
}

// post 发起 OpenAI 兼容请求。ctx 可被上层（如编排 150s 兜底）用 WithTimeout 取消，
// 使 LLM 调用真正中断，而不是只能等死的 http.Client.Timeout。
func (c *Client) post(ctx context.Context, path string, body any) (*http.Response, error) {
	j, err := json.Marshal(body)
	if err != nil {
		return nil, err
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url(path), bytes.NewReader(j))
	if err != nil {
		return nil, err
	}
	for k, v := range c.headers() {
		req.Header.Set(k, v)
	}
	resp, err := c.hc.Do(req)
	if err != nil {
		return nil, err
	}
	return resp, nil
}

// ──────────────────────────────────────────────
// 非流式调用
// EmptyCompletionError 上游回了 200，但正文是空的。
//
// 为什么要单独成类型：**「空正文」有好几种成因，处置完全不同，而 finish_reason
// 是唯一能把它们分开的东西**。
//
// 最要命的一种是推理模型把 max_tokens 全花在思考上 —— 此时响应里**根本没有
// content 字段**（不是空串，是不存在），finish_reason=length。实测某对话路由
// （max_tokens=4096）对 2500 字的密集中文输入，reasoning_content 写了 7187 字、
// 正文 0 字；把预算提到 8192 才有正文（2470 字，reasoning 9590 字）。
// 这种情况报「空正文」会把人引去查模型名和密钥，而真正该做的是调大预算；
// 报「思考吃光了预算」才能一眼看到病根。
//
// 只由 GenerateFull 返回（Generate 只回字符串，拿不到 finish_reason）。
type EmptyCompletionError struct {
	FinishReason     string
	CompletionTokens int
}

// Error 保留「LLM 返回空正文」前缀：这是这条例外此前的文案，外部可能有按
// 子串匹配的地方，补上诊断信息不该顺手改掉已经存在的说法。
func (e *EmptyCompletionError) Error() string {
	if e.FinishReason == "" {
		return "LLM 返回空正文（响应里没有 finish_reason）"
	}
	if strings.EqualFold(e.FinishReason, "length") {
		return fmt.Sprintf(
			"LLM 返回空正文（finish_reason=length，completion_tokens=%d）：预算被思考用光了，"+
				"推理模型写正文之前就把 max_tokens 花完了，调大该路由的 max_tokens 才可能写出正文",
			e.CompletionTokens)
	}
	return fmt.Sprintf("LLM 返回空正文（finish_reason=%q，completion_tokens=%d）", e.FinishReason, e.CompletionTokens)
}

// TransientUpstreamError 上游这一次没成，但**同样的请求重发有机会成**。
//
// 与 EmptyCompletionError 的区别就是「要不要重试」这一件事，而这件事必须由类型
// 决定、不能由文案猜：
//
//   - EmptyCompletionError 是模型确实写不出正文（预算被思考吃光、模型拒答），
//     重发一百次都是同一个结果，重试只是再花一次钱；
//   - 本类型是**协议或链路层面的抖动** —— 网关回了 200 却给一个没有 choices 的
//     报文、限流、上游 5xx、连接被掐断。这些重发就有机会成，不重发则整步白跑。
//
// 实测背景：某技能的第 3 步要连打 11 次模型（逐字稿按 1200 字分块），第 4 次
// 撞上的就是这个 —— 网关 HTTP 200，报文里却没有 choices，整步就此失败。
type TransientUpstreamError struct {
	// StatusCode 0 表示压根没拿到状态码（连接就没建起来）。
	// 目前只用来说明情况，重试决策在构造处就定了。
	StatusCode int
	// Message 是**原样的**错误文案。故意不做二次拼装：这些文案是既有说法，
	// 记录日志、前端提示都可能在看，加个类型标记没理由顺手改动它们。
	Message string
	// Err 保留底层错误，让 errors.Is/As 还能穿透下去
	// （例如上层要判 context.DeadlineExceeded）。此前靠 %w 提供的这条链不能断。
	Err error
}

func (e *TransientUpstreamError) Error() string { return e.Message }

func (e *TransientUpstreamError) Unwrap() error { return e.Err }

// IsTransient 判断 err 值不值得重发。
//
// 走 errors.As，所以**经过 %w 包装后依然认得出** —— 回退链会给错误套上
// 「所有路由均失败: [路由 id] …」两层包装，调用方拿到的是最外层那个。
func IsTransient(err error) bool {
	var transient *TransientUpstreamError
	return errors.As(err, &transient)
}

// ──────────────────────────────────────────────

// Generate 非流式对话，返回完整正文
func (c *Client) Generate(ctx context.Context, messages []Message) (string, error) {
	resp, err := c.post(ctx, "/chat/completions", map[string]any{
		"model":       c.model,
		"messages":    messages,
		"stream":      false,
		"temperature": c.temperature,
		"max_tokens":  c.maxTokens,
	})
	if err != nil {
		return "", fmt.Errorf("LLM 服务不可达: %w", err)
	}
	defer resp.Body.Close()

	data, err := io.ReadAll(resp.Body)
	if err != nil {
		return "", fmt.Errorf("读取 LLM 响应失败: %w", err)
	}
	if resp.StatusCode != http.StatusOK {
		return "", fmt.Errorf("LLM 返回 %d: %s", resp.StatusCode, truncate(string(data), 200))
	}

	var out struct {
		Choices []struct {
			Message struct {
				Content string `json:"content"`
			} `json:"message"`
		} `json:"choices"`
	}
	if err := json.Unmarshal(data, &out); err != nil {
		return "", fmt.Errorf("LLM 响应解析失败: %w", err)
	}
	if len(out.Choices) == 0 || strings.TrimSpace(out.Choices[0].Message.Content) == "" {
		return "", fmt.Errorf("LLM 返回空正文")
	}
	return out.Choices[0].Message.Content, nil
}

// GenerateFull 非流式调用，返回完整 ChatResult（含 usage）
func (c *Client) GenerateFull(ctx context.Context, messages []Message) (*ChatResult, error) {
	resp, err := c.post(ctx, "/chat/completions", map[string]any{
		"model":       c.model,
		"messages":    messages,
		"stream":      false,
		"temperature": c.temperature,
		"max_tokens":  c.maxTokens,
	})
	if err != nil {
		// 连接建不起来 / 被掐断 / 超时：重发有意义。
		return nil, &TransientUpstreamError{
			Message: fmt.Sprintf("LLM 服务不可达: %v", err),
			Err:     err,
		}
	}
	defer resp.Body.Close()

	data, err := io.ReadAll(resp.Body)
	if err != nil {
		// 正文读到一半断了，同样属于这一趟的抖动。
		return nil, &TransientUpstreamError{
			Message: fmt.Sprintf("读取 LLM 响应失败: %v", err),
			Err:     err,
		}
	}
	if resp.StatusCode != http.StatusOK {
		msg := fmt.Sprintf("LLM 返回 %d: %s", resp.StatusCode, truncate(string(data), 200))
		// 限流和上游 5xx 是**这一趟**的问题，重发可能就过了；
		// 4xx（模型名写错、密钥无效、参数非法）重发一百次也是同一个结果，不标可重试。
		if resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode >= 500 {
			return nil, &TransientUpstreamError{StatusCode: resp.StatusCode, Message: msg}
		}
		return nil, errors.New(msg)
	}

	var raw struct {
		Model   string `json:"model"`
		Choices []struct {
			Message struct {
				Content string `json:"content"`
			} `json:"message"`
			FinishReason string `json:"finish_reason"`
		} `json:"choices"`
		Usage *struct {
			PromptTokens     int `json:"prompt_tokens"`
			CompletionTokens int `json:"completion_tokens"`
			TotalTokens      int `json:"total_tokens"`
		} `json:"usage"`
	}
	if err := json.Unmarshal(data, &raw); err != nil {
		// 报文根本不是 JSON：与「没有 choices」同一种病 —— 网关塞了别的东西进来。
		return nil, &TransientUpstreamError{
			StatusCode: resp.StatusCode,
			Message:    fmt.Sprintf("LLM 响应解析失败: %v", err),
			Err:        err,
		}
	}
	if len(raw.Choices) == 0 {
		// 没有 choices 是**协议层面**的异常，跟「模型没写出正文」不是一回事：
		// 网关回 200 却是别的报文形状（限流、上游报错包在 200 里等）都会落到这里。
		// 所以把原始报文带上 —— 只说一句「空正文」，拿到的人根本无从下手。
		// 与上面状态码分支一样截断，且保留「LLM 返回空正文」这个前缀。
		//
		// 标成可重试：报文形状不对是**这一趟**的事，不是模型对这个输入的回答。
		return nil, &TransientUpstreamError{
			StatusCode: resp.StatusCode,
			Message:    fmt.Sprintf("LLM 返回空正文（响应里没有 choices）：%s", truncate(string(data), 300)),
		}
	}
	if strings.TrimSpace(raw.Choices[0].Message.Content) == "" {
		completionTokens := 0
		if raw.Usage != nil {
			completionTokens = raw.Usage.CompletionTokens
		}
		return nil, &EmptyCompletionError{
			FinishReason:     raw.Choices[0].FinishReason,
			CompletionTokens: completionTokens,
		}
	}

	result := &ChatResult{
		Content:      raw.Choices[0].Message.Content,
		Model:        raw.Model,
		FinishReason: raw.Choices[0].FinishReason,
	}
	if raw.Usage != nil {
		result.Usage.PromptTokens = raw.Usage.PromptTokens
		result.Usage.CompletionTokens = raw.Usage.CompletionTokens
		result.Usage.TotalTokens = raw.Usage.TotalTokens
	}
	return result, nil
}

// ──────────────────────────────────────────────
// 流式调用
// ──────────────────────────────────────────────

// GenerateStream SSE 流式调用，逐 chunk 回调
func (c *Client) GenerateStream(ctx context.Context, messages []Message, onChunk func(string)) error {
	resp, err := c.post(ctx, "/chat/completions", map[string]any{
		"model":          c.model,
		"messages":       messages,
		"stream":         true,
		"temperature":    c.temperature,
		"max_tokens":     c.maxTokens,
		"stream_options": map[string]any{"include_usage": true},
	})
	if err != nil {
		return fmt.Errorf("LLM 服务不可达: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		data, _ := io.ReadAll(resp.Body)
		return fmt.Errorf("LLM 返回 %d: %s", resp.StatusCode, truncate(string(data), 200))
	}

	// 流式空闲超时保护：若超过 idleTimeout 没有收到任何可消费 chunk，
	// 判定为卡死并返回错误，避免前端无限等待（此前本地模型卡顿时会干等 120s）。
	const idleTimeout = 45 * time.Second

	scanner := bufio.NewScanner(resp.Body)
	scanner.Buffer(make([]byte, 64*1024), 1024*1024)
	scannerCh := make(chan string, 1)
	// 驱动 scanner 的 goroutine，配合 select 实现空闲超时。
	// 函数返回时 defer resp.Body.Close() 会让 scanner.Scan() 立即返回，
	// goroutine 随之退出并通过 close 通知主循环（缓冲 channel 避免阻塞发送）。
	go func() {
		for scanner.Scan() {
			scannerCh <- scanner.Text()
		}
		close(scannerCh)
	}()

	idleTimer := time.NewTimer(idleTimeout)
	defer idleTimer.Stop()
	for {
		select {
		case <-idleTimer.C:
			return fmt.Errorf("LLM 流式响应空闲超时（%s 内无数据）", idleTimeout)
		case line, ok := <-scannerCh:
			if !ok {
				return nil
			}
			if !strings.HasPrefix(line, "data:") {
				continue
			}
			payload := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
			if payload == "[DONE]" {
				return nil
			}
			var chunk struct {
				Choices []struct {
					Delta struct {
						Content string `json:"content"`
					} `json:"delta"`
				} `json:"choices"`
			}
			if err := json.Unmarshal([]byte(payload), &chunk); err != nil {
				continue
			}
			if len(chunk.Choices) > 0 && chunk.Choices[0].Delta.Content != "" {
				// 收到有效内容，重置空闲计时
				if !idleTimer.Stop() {
					<-idleTimer.C
				}
				idleTimer.Reset(idleTimeout)
				onChunk(chunk.Choices[0].Delta.Content)
			}
		}
	}
}

// ──────────────────────────────────────────────
// Embedding
// ──────────────────────────────────────────────

// Embed 批量向量化
func (c *Client) Embed(ctx context.Context, inputs []string) ([][]float64, error) {
	resp, err := c.post(ctx, "/embeddings", map[string]any{
		"model": c.model, "input": inputs,
	})
	if err != nil {
		return nil, fmt.Errorf("embedding 服务不可达: %w", err)
	}
	defer resp.Body.Close()

	data, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, fmt.Errorf("读取 embedding 响应失败: %w", err)
	}
	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("embedding 返回 %d: %s", resp.StatusCode, truncate(string(data), 200))
	}

	var out struct {
		Data []struct {
			Embedding []float64 `json:"embedding"`
		} `json:"data"`
	}
	if err := json.Unmarshal(data, &out); err != nil {
		return nil, fmt.Errorf("embedding 响应解析失败: %w", err)
	}
	res := make([][]float64, len(out.Data))
	for i, d := range out.Data {
		res[i] = d.Embedding
	}
	return res, nil
}

// ──────────────────────────────────────────────
// Fallback 链调用
// ──────────────────────────────────────────────

// buildRouteChain 构建「主路由 + 回退链」。
//
// primary 为 nil 时必须显式报错而不是继续：链里存 nil 会在循环里 NewClient(nil)，
// 而取 primary.RouteID 更是直接空指针。历史上 batch_extract / contract_review
// 就是直接传了 nil，导致这两个接口必 500。调用方应先用 config.GetRoute 解析路由。
func buildRouteChain(primary *config.RouteConfig) ([]*config.RouteConfig, error) {
	if primary == nil {
		return nil, fmt.Errorf("主路由为 nil：调用方未解析 AI 路由，请先用 config.GetRoute 取路由")
	}
	aiRouteChain := []*config.RouteConfig{primary}
	if fallbacks, err := config.GetFallbackRoutes(primary.RouteID); err == nil && len(fallbacks) > 0 {
		aiRouteChain = append(aiRouteChain, fallbacks...)
	}
	return aiRouteChain, nil
}

// GenerateWithFallback 依次尝试主路由 + 回退链，首个成功返回。
// ctx 贯穿到各路由的 HTTP 请求，使上层超时（如编排 150s 兜底）真正能取消 LLM 调用。
func GenerateWithFallback(ctx context.Context, primary *config.RouteConfig, messages []Message) (string, error) {
	aiRouteChain, err := buildRouteChain(primary)
	if err != nil {
		return "", err
	}

	var lastErr error
	for _, aiRoute := range aiRouteChain {
		client := NewClient(aiRoute)
		start := time.Now()
		content, err := client.Generate(ctx, messages)
		if err == nil {
			config.ReportRouteHealth(config.RouteHealth{
				AIRouteID:     aiRoute.RouteID,
				Category:      aiRoute.Category,
				Healthy:       true,
				Checked:       true,
				LatencyMs:     time.Since(start).Milliseconds(),
				LastCheckedAt: time.Now(),
			})
			return content, nil
		}
		config.ReportRouteHealth(config.RouteHealth{
			AIRouteID:     aiRoute.RouteID,
			Category:      aiRoute.Category,
			Healthy:       false,
			Checked:       true,
			LatencyMs:     time.Since(start).Milliseconds(),
			LastCheckedAt: time.Now(),
			LastError:     err.Error(),
		})
		lastErr = fmt.Errorf("[%s] %w", aiRoute.RouteID, err)
	}
	return "", fmt.Errorf("所有路由均失败: %w", lastErr)
}

// GenerateFullWithFallback 非流式 + 回退链：返回结果与「实际命中」的路由（用于审计）。
// ctx 贯穿到各路由的 HTTP 请求，使上层超时真正能取消 LLM 调用。
func GenerateFullWithFallback(ctx context.Context, primary *config.RouteConfig, messages []Message) (*ChatResult, *config.RouteConfig, error) {
	aiRouteChain, err := buildRouteChain(primary)
	if err != nil {
		return nil, nil, err
	}

	var lastErr error
	for _, aiRoute := range aiRouteChain {
		client := NewClient(aiRoute)
		start := time.Now()
		res, err := client.GenerateFull(ctx, messages)
		if err == nil {
			config.ReportRouteHealth(config.RouteHealth{
				AIRouteID:     aiRoute.RouteID,
				Category:      aiRoute.Category,
				Healthy:       true,
				Checked:       true,
				LatencyMs:     time.Since(start).Milliseconds(),
				LastCheckedAt: time.Now(),
			})
			res.Provider = aiRoute.Provider
			return res, aiRoute, nil
		}
		config.ReportRouteHealth(config.RouteHealth{
			AIRouteID:     aiRoute.RouteID,
			Category:      aiRoute.Category,
			Healthy:       false,
			Checked:       true,
			LatencyMs:     time.Since(start).Milliseconds(),
			LastCheckedAt: time.Now(),
			LastError:     err.Error(),
		})
		lastErr = fmt.Errorf("[%s] %w", aiRoute.RouteID, err)
	}
	return nil, nil, fmt.Errorf("所有路由均失败: %w", lastErr)
}

// GenerateStreamWithFallback 流式 + 回退链：返回实际命中的路由。
// ctx 贯穿到各路由的 SSE 请求，使上层超时（如前端断开）真正能取消 LLM 调用。
func GenerateStreamWithFallback(ctx context.Context, primary *config.RouteConfig, messages []Message, onChunk func(string)) (*config.RouteConfig, error) {
	aiRouteChain, err := buildRouteChain(primary)
	if err != nil {
		return nil, err
	}

	var lastErr error
	for _, aiRoute := range aiRouteChain {
		// 判据只有 config 那一处，本文件原先自己抄了一份（缺「回环地址免密钥」）。
		if config.RequiresRouteAPIKey(aiRoute) && strings.TrimSpace(aiRoute.APIKey) == "" {
			config.ReportRouteHealth(config.RouteHealth{
				AIRouteID:     aiRoute.RouteID,
				Category:      aiRoute.Category,
				Healthy:       false,
				Checked:       true,
				LastCheckedAt: time.Now(),
				LastError:     "未配置 API Key",
			})
			lastErr = fmt.Errorf("[%s] 未配置 API Key", aiRoute.RouteID)
			continue
		}
		client := NewClient(aiRoute)
		start := time.Now()
		if err := client.GenerateStream(ctx, messages, onChunk); err == nil {
			config.ReportRouteHealth(config.RouteHealth{
				AIRouteID:     aiRoute.RouteID,
				Category:      aiRoute.Category,
				Healthy:       true,
				Checked:       true,
				LatencyMs:     time.Since(start).Milliseconds(),
				LastCheckedAt: time.Now(),
			})
			return aiRoute, nil
		} else {
			config.ReportRouteHealth(config.RouteHealth{
				AIRouteID:     aiRoute.RouteID,
				Category:      aiRoute.Category,
				Healthy:       false,
				Checked:       true,
				LatencyMs:     time.Since(start).Milliseconds(),
				LastCheckedAt: time.Now(),
				LastError:     err.Error(),
			})
			lastErr = fmt.Errorf("[%s] %w", aiRoute.RouteID, err)
		}
	}
	return nil, fmt.Errorf("所有路由均失败: %w", lastErr)
}

func truncate(s string, n int) string {
	r := []rune(s)
	if len(r) <= n {
		return s
	}
	return string(r[:n]) + "..."
}