Cyrene_v2/internal/runtime/model/openai.go

package model

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"net/http"
	"strings"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)

// 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, trace.URLCause(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, trace.URLCause(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, trace.URLCause(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(),
	})
}