AskaEth
66dda084cd
M41 的「LLM / embedding / 视觉 上游错误体」一族（M38 #5 #6 #7 #8 #9 #14）： 非 2xx 时把响应体截断后拼进 error，而 error 会被 chat run / loop step / retrieve_fallback 打日志，等于把她的提问、记忆正文、要念的台词写进日志。 - model：openai/anthropic 两处 upstreamError 收口到 upstream.go 的 providerUpstreamError；anthropic 流内 error 事件的 evt.Error.Message （原本无上限）整段丢弃，只保留令牌形状的 type - embedding：readErrorBody 删除，改走 helper（input 就是她的记忆正文） - vision：错误路径不再沿用取图的 50 MiB 上限，改为 4 KiB 有界读且不拼原文 - voice：classifyDashscopeStatus 只解 code/request_id，正文只留在内存里做 分类判据；错误体读取同样改为 4 KiB 有界 - 测试：五个上游各一条「日志/error 里搜不到哨兵正文」用例，且能变异打红
255 lines
8.1 KiB
Go
255 lines
8.1 KiB
Go
package model
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"log/slog"
|
||
"net/http"
|
||
"strings"
|
||
"time"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||
)
|
||
|
||
// OpenAI 兼容 /chat/completions 协议的线格式约束。
|
||
const (
|
||
// sseDataPrefix 标记携带 JSON 负载的 SSE 事件行。
|
||
sseDataPrefix = "data: "
|
||
// sseDone 是终止用的 SSE 哨兵。
|
||
sseDone = "[DONE]"
|
||
// sseBufInit 是 bufio.Scanner 的初始读缓冲大小。
|
||
sseBufInit = 64 * 1024
|
||
// maxSSELineBytes 限制单条 SSE 事件行的长度。
|
||
maxSSELineBytes = 1024 * 1024
|
||
// defaultHTTPTimeout 在 Options.Timeout 为 0 时约束单次提供商调用。
|
||
defaultHTTPTimeout = 180 * time.Second
|
||
)
|
||
|
||
// OpenAIChatAdapter 调用 OpenAI 兼容的 /chat/completions 端点。它由
|
||
// config.LLMConfig 构建，同时实现非流式与 token 级 SSE 流式的
|
||
// Adapter。
|
||
type OpenAIChatAdapter struct {
|
||
name string
|
||
baseURL string
|
||
apiKey string
|
||
model string
|
||
client *http.Client
|
||
recorder Recorder
|
||
// usageSink 是可选的用量落库出口（切片 U1）。为 nil 时只记结构化日志，
|
||
// 行为逐字等于引入落库之前 —— 这也是所有既有测试与未接线调用方的形态。
|
||
usageSink UsageSink
|
||
// reasoningKind 是本端点已配置的推理开关形状
|
||
// （models.json 组的 reasoning.kind / LLM_REASONING_KIND，裁决 03.4.2/08.4.1）。
|
||
// 为空表示 "auto"：按模型名/base URL 推断厂商，
|
||
// 形状含糊时什么都不发。
|
||
reasoningKind string
|
||
// log 上报 reasoning_effort 校验与被忽略的思考指令。它
|
||
// 永不为 nil：调用方未提供时，会装入一个带有本适配器来源
|
||
// 标记的丢弃式日志器，因此每个决定都可追溯。
|
||
log *slog.Logger
|
||
}
|
||
|
||
// OpenAIOption 用于配置 OpenAIChatAdapter。
|
||
type OpenAIOption func(*OpenAIChatAdapter)
|
||
|
||
// WithHTTPClient 覆盖默认 HTTP 客户端（用于在测试与生产中
|
||
// 注入 mock 或调优过的 transport）。
|
||
func WithHTTPClient(c *http.Client) OpenAIOption {
|
||
return func(a *OpenAIChatAdapter) { a.client = c }
|
||
}
|
||
|
||
// WithRecorder 装入按调用计账的 Recorder，用于 CallRecord 统计。
|
||
func WithRecorder(r Recorder) OpenAIOption {
|
||
return func(a *OpenAIChatAdapter) { a.recorder = r }
|
||
}
|
||
|
||
// WithUsageSink 装入用量落库出口（切片 U1）。nil 会被忽略，从而保留
|
||
// 「只记日志」的默认行为；实现方必须非阻塞（见 UsageSink）。
|
||
func WithUsageSink(s UsageSink) OpenAIOption {
|
||
return func(a *OpenAIChatAdapter) { a.usageSink = s }
|
||
}
|
||
|
||
// WithModel 覆盖适配器的默认模型名。
|
||
func WithModel(m string) OpenAIOption {
|
||
return func(a *OpenAIChatAdapter) { a.model = m }
|
||
}
|
||
|
||
// WithOpenAILogger 挂上用于上报 reasoning_effort 校验与
|
||
// 被忽略思考指令的日志器。nil 日志器会被忽略，
|
||
// 从而保留丢弃式的默认值。
|
||
func WithOpenAILogger(l *slog.Logger) OpenAIOption {
|
||
return func(a *OpenAIChatAdapter) {
|
||
if l != nil {
|
||
a.log = l
|
||
}
|
||
}
|
||
}
|
||
|
||
// NewOpenAIChat 由供应商配置构建适配器。baseURL 会先去掉
|
||
// 末尾斜杠，再拼接 /chat/completions 路径。
|
||
// 端点已配置的推理形状（cfg.ReasoningKind）随
|
||
// 适配器一起走，因此由同一份配置构建的每条链——组链、
|
||
// 它的按角色链以及本地级——都以
|
||
// 相同方式解出自己的推理开关。
|
||
func NewOpenAIChat(cfg config.LLMConfig, opts ...OpenAIOption) *OpenAIChatAdapter {
|
||
a := &OpenAIChatAdapter{
|
||
name: "openai",
|
||
baseURL: strings.TrimRight(cfg.BaseURL, "/"),
|
||
apiKey: cfg.APIKey,
|
||
model: cfg.Model,
|
||
reasoningKind: strings.TrimSpace(cfg.ReasoningKind),
|
||
client: &http.Client{
|
||
Transport: http.DefaultTransport,
|
||
Timeout: defaultHTTPTimeout,
|
||
},
|
||
log: discardLogger("runtime/model/openai"),
|
||
}
|
||
for _, o := range opts {
|
||
o(a)
|
||
}
|
||
return a
|
||
}
|
||
|
||
// Name 返回适配器标识。
|
||
func (a *OpenAIChatAdapter) Name() string { return a.name }
|
||
|
||
// Chat 执行一次非流式补全。
|
||
func (a *OpenAIChatAdapter) Chat(ctx context.Context, req Request, opts Options) (Response, error) {
|
||
start := time.Now()
|
||
if a.baseURL == "" {
|
||
return Response{}, ErrUpstream
|
||
}
|
||
ctx, cancel := withCallTimeout(ctx, opts)
|
||
defer cancel()
|
||
model, err := a.resolveModel(req)
|
||
if err != nil {
|
||
return Response{}, err
|
||
}
|
||
req.Model = model
|
||
body, err := a.buildRequest(req, false)
|
||
if err != nil {
|
||
return Response{}, err
|
||
}
|
||
httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, a.endpoint(), bytes.NewReader(body))
|
||
if err != nil {
|
||
return Response{}, fmt.Errorf("model: %w: build request: %v", ErrUpstream, err)
|
||
}
|
||
resp, err := a.doRequest(httpReq, false)
|
||
if err != nil {
|
||
return Response{}, err
|
||
}
|
||
defer resp.Body.Close()
|
||
if resp.StatusCode != http.StatusOK {
|
||
return Response{}, a.upstreamError(resp)
|
||
}
|
||
var wire openAIChatResponse
|
||
if err := json.NewDecoder(resp.Body).Decode(&wire); err != nil {
|
||
return Response{}, fmt.Errorf("model: %w: decode response: %v", ErrUpstream, err)
|
||
}
|
||
out := responseFromWire(&wire)
|
||
a.record(ctx, req, start, out)
|
||
return out, nil
|
||
}
|
||
|
||
// ChatStream 通过 onEvent 流出结构化事件，并在提供商于
|
||
// 末块上报用量时返回带用量的聚合响应。
|
||
func (a *OpenAIChatAdapter) ChatStream(ctx context.Context, req Request, onEvent func(StreamEvent)) (Response, error) {
|
||
start := time.Now()
|
||
if a.baseURL == "" {
|
||
return Response{}, ErrUpstream
|
||
}
|
||
model, err := a.resolveModel(req)
|
||
if err != nil {
|
||
return Response{}, err
|
||
}
|
||
req.Model = model
|
||
body, err := a.buildRequest(req, true)
|
||
if err != nil {
|
||
return Response{}, err
|
||
}
|
||
httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, a.endpoint(), bytes.NewReader(body))
|
||
if err != nil {
|
||
return Response{}, fmt.Errorf("model: %w: build request: %v", ErrUpstream, err)
|
||
}
|
||
resp, err := a.doRequest(httpReq, true)
|
||
if err != nil {
|
||
return Response{}, err
|
||
}
|
||
defer resp.Body.Close()
|
||
if resp.StatusCode != http.StatusOK {
|
||
return Response{}, a.upstreamError(resp)
|
||
}
|
||
out, err := scanSSE(resp.Body, onEvent)
|
||
if err != nil {
|
||
return Response{}, err
|
||
}
|
||
out.Model = model
|
||
a.record(ctx, req, start, out)
|
||
return out, nil
|
||
}
|
||
|
||
// withCallTimeout 在 opts.Timeout 为正时用该超时包裹 ctx。
|
||
func withCallTimeout(parent context.Context, opts Options) (context.Context, context.CancelFunc) {
|
||
if opts.Timeout > 0 {
|
||
return context.WithTimeout(parent, opts.Timeout)
|
||
}
|
||
return parent, func() {}
|
||
}
|
||
|
||
// resolveModel 返回实际生效的模型名。
|
||
func (a *OpenAIChatAdapter) resolveModel(req Request) (string, error) {
|
||
if req.Model != "" {
|
||
return req.Model, nil
|
||
}
|
||
if a.model != "" {
|
||
return a.model, nil
|
||
}
|
||
return "", ErrNoModel
|
||
}
|
||
|
||
// endpoint 返回完整的 /chat/completions URL。
|
||
func (a *OpenAIChatAdapter) endpoint() string {
|
||
return a.baseURL + "/chat/completions"
|
||
}
|
||
|
||
func (a *OpenAIChatAdapter) doRequest(httpReq *http.Request, stream bool) (*http.Response, error) {
|
||
if a.apiKey != "" {
|
||
httpReq.Header.Set("Authorization", "Bearer "+a.apiKey)
|
||
}
|
||
httpReq.Header.Set("Content-Type", "application/json")
|
||
if stream {
|
||
httpReq.Header.Set("Accept", "text/event-stream")
|
||
} else {
|
||
httpReq.Header.Set("Accept", "application/json")
|
||
}
|
||
resp, err := a.client.Do(httpReq)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("model: %w: %v", ErrUpstream, err)
|
||
}
|
||
return resp, nil
|
||
}
|
||
|
||
// upstreamError 把一次非 2xx 响应交给包内唯一的上游错误收口（见 upstream.go）：
|
||
// 只留 status / code / type 与体积摘要，响应体正文与 error.message 不出镜。
|
||
func (a *OpenAIChatAdapter) upstreamError(resp *http.Response) error {
|
||
return providerUpstreamError(a.name, a.baseURL, resp)
|
||
}
|
||
|
||
// record 上报一次成功调用的用量：先记一条结构化用量日志（含缓存命中率，
|
||
// 所有调用点都覆盖 —— 见 usage.go），再喂给可选的 CallRecord 计账出口。
|
||
// ctx 用来取本次调用所属的会话（loop 播种，见 session_ctx.go）。
|
||
func (a *OpenAIChatAdapter) record(ctx context.Context, req Request, start time.Time, resp Response) {
|
||
logUsage(ctx, a.log, a.usageSink, a.name, req, resp, time.Since(start))
|
||
if a.recorder == nil {
|
||
return
|
||
}
|
||
a.recorder(CallRecord{
|
||
Model: req.Model,
|
||
Purpose: req.Purpose,
|
||
Tokens: resp.TotalTokens,
|
||
TookMs: time.Since(start).Milliseconds(),
|
||
})
|
||
}
|