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