+135
-12
@@ -10,6 +10,7 @@ import (
|
||||
"fmt"
|
||||
"log"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
@@ -61,20 +62,72 @@ type AiGatewayService struct {
|
||||
onChannelsChanged func(context.Context)
|
||||
// filterDeprecated 是「过滤弃用模型」开关(内存镜像,持久化在 settings 表)
|
||||
filterDeprecated atomic.Bool
|
||||
// streamGuard* 是 Responses 流式保险丝(instructions+tools 合计超阈值改非流式)
|
||||
streamGuardEnabled atomic.Bool
|
||||
streamGuardKB atomic.Int64
|
||||
// grokWebSearch / grokXSearch 是 xai. 模型服务端搜索工具默认注入开关
|
||||
grokWebSearch atomic.Bool
|
||||
grokXSearch atomic.Bool
|
||||
}
|
||||
|
||||
// NewAiGatewayService 组装依赖;调用 StartCleanup 后开始调用日志周期清理。
|
||||
func NewAiGatewayService(db *gorm.DB, configs *OciConfigService, client oci.Client) *AiGatewayService {
|
||||
s := &AiGatewayService{db: db, configs: configs, client: client, lastTouch: map[uint]time.Time{}}
|
||||
var row model.Setting
|
||||
if err := db.Where("key = ?", settingAiFilterDeprecated).First(&row).Error; err == nil {
|
||||
s.filterDeprecated.Store(row.Value == "1")
|
||||
}
|
||||
s.filterDeprecated.Store(loadBoolSetting(db, settingAiFilterDeprecated, false))
|
||||
s.streamGuardEnabled.Store(loadBoolSetting(db, settingAiStreamGuardEnabled, true))
|
||||
s.streamGuardKB.Store(int64(loadIntSetting(db, settingAiStreamGuardKB, defaultStreamGuardKB)))
|
||||
s.grokWebSearch.Store(loadBoolSetting(db, settingAiGrokWebSearch, true))
|
||||
s.grokXSearch.Store(loadBoolSetting(db, settingAiGrokXSearch, true))
|
||||
return s
|
||||
}
|
||||
|
||||
// settingAiFilterDeprecated 是「过滤弃用模型」开关的配置键,值 "1"/"0",缺省关闭。
|
||||
const settingAiFilterDeprecated = "ai_filter_deprecated"
|
||||
// AI 网关运行时设置的配置键;bool 值存 "1"/"0"。
|
||||
const (
|
||||
// settingAiFilterDeprecated 是「过滤弃用模型」开关,缺省关闭。
|
||||
settingAiFilterDeprecated = "ai_filter_deprecated"
|
||||
// settingAiStreamGuardEnabled / settingAiStreamGuardKB 是流式保险丝开关与
|
||||
// 阈值(KB),缺省开、60(上游对 instructions+tools >≈64.5KB 流式静默断流)。
|
||||
settingAiStreamGuardEnabled = "ai_stream_guard_enabled"
|
||||
settingAiStreamGuardKB = "ai_stream_guard_kb"
|
||||
// settingAiGrokWebSearch / settingAiGrokXSearch 是 grok 搜索工具默认注入
|
||||
// 开关,缺省开。
|
||||
settingAiGrokWebSearch = "ai_grok_web_search"
|
||||
settingAiGrokXSearch = "ai_grok_x_search"
|
||||
)
|
||||
|
||||
// defaultStreamGuardKB 是保险丝阈值缺省值,低于实测断流边界留余量。
|
||||
const defaultStreamGuardKB = 60
|
||||
|
||||
// loadBoolSetting 读 settings 表布尔键,无行或值非法时返回缺省。
|
||||
func loadBoolSetting(db *gorm.DB, key string, def bool) bool {
|
||||
var row model.Setting
|
||||
if err := db.Where("key = ?", key).First(&row).Error; err != nil {
|
||||
return def
|
||||
}
|
||||
return row.Value == "1"
|
||||
}
|
||||
|
||||
// loadIntSetting 读 settings 表整数键,无行或解析失败时返回缺省。
|
||||
func loadIntSetting(db *gorm.DB, key string, def int) int {
|
||||
var row model.Setting
|
||||
if err := db.Where("key = ?", key).First(&row).Error; err != nil {
|
||||
return def
|
||||
}
|
||||
n, err := strconv.Atoi(row.Value)
|
||||
if err != nil {
|
||||
return def
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// saveBoolSetting 持久化布尔键。
|
||||
func (s *AiGatewayService) saveBoolSetting(ctx context.Context, key string, on bool) error {
|
||||
value := "0"
|
||||
if on {
|
||||
value = "1"
|
||||
}
|
||||
return s.db.WithContext(ctx).Save(&model.Setting{Key: key, Value: value}).Error
|
||||
}
|
||||
|
||||
// FilterDeprecated 返回「过滤弃用模型」开关状态。
|
||||
func (s *AiGatewayService) FilterDeprecated() bool { return s.filterDeprecated.Load() }
|
||||
@@ -82,18 +135,54 @@ func (s *AiGatewayService) FilterDeprecated() bool { return s.filterDeprecated.L
|
||||
// SetFilterDeprecated 持久化并即时生效开关:开启后已宣布弃用
|
||||
// (deprecated_at 非空,即使未退役)的模型从列表与路由中排除。
|
||||
func (s *AiGatewayService) SetFilterDeprecated(ctx context.Context, on bool) error {
|
||||
value := "0"
|
||||
if on {
|
||||
value = "1"
|
||||
}
|
||||
err := s.db.WithContext(ctx).Save(&model.Setting{Key: settingAiFilterDeprecated, Value: value}).Error
|
||||
if err != nil {
|
||||
if err := s.saveBoolSetting(ctx, settingAiFilterDeprecated, on); err != nil {
|
||||
return fmt.Errorf("保存过滤弃用模型开关: %w", err)
|
||||
}
|
||||
s.filterDeprecated.Store(on)
|
||||
return nil
|
||||
}
|
||||
|
||||
// StreamGuard 返回流式保险丝开关与阈值(KB)。
|
||||
func (s *AiGatewayService) StreamGuard() (bool, int) {
|
||||
return s.streamGuardEnabled.Load(), int(s.streamGuardKB.Load())
|
||||
}
|
||||
|
||||
// SetStreamGuard 持久化并即时生效流式保险丝;kb 限定 1..1024。
|
||||
func (s *AiGatewayService) SetStreamGuard(ctx context.Context, on bool, kb int) error {
|
||||
if kb < 1 || kb > 1024 {
|
||||
return fmt.Errorf("流式保险丝阈值须在 1..1024 KB, 收到 %d", kb)
|
||||
}
|
||||
if err := s.saveBoolSetting(ctx, settingAiStreamGuardEnabled, on); err != nil {
|
||||
return fmt.Errorf("保存流式保险丝开关: %w", err)
|
||||
}
|
||||
err := s.db.WithContext(ctx).
|
||||
Save(&model.Setting{Key: settingAiStreamGuardKB, Value: strconv.Itoa(kb)}).Error
|
||||
if err != nil {
|
||||
return fmt.Errorf("保存流式保险丝阈值: %w", err)
|
||||
}
|
||||
s.streamGuardEnabled.Store(on)
|
||||
s.streamGuardKB.Store(int64(kb))
|
||||
return nil
|
||||
}
|
||||
|
||||
// GrokSearch 返回 grok 服务端搜索工具默认注入开关(web_search, x_search)。
|
||||
func (s *AiGatewayService) GrokSearch() (bool, bool) {
|
||||
return s.grokWebSearch.Load(), s.grokXSearch.Load()
|
||||
}
|
||||
|
||||
// SetGrokSearch 持久化并即时生效 grok 搜索工具默认注入开关。
|
||||
func (s *AiGatewayService) SetGrokSearch(ctx context.Context, web, x bool) error {
|
||||
if err := s.saveBoolSetting(ctx, settingAiGrokWebSearch, web); err != nil {
|
||||
return fmt.Errorf("保存 grok web_search 开关: %w", err)
|
||||
}
|
||||
if err := s.saveBoolSetting(ctx, settingAiGrokXSearch, x); err != nil {
|
||||
return fmt.Errorf("保存 grok x_search 开关: %w", err)
|
||||
}
|
||||
s.grokWebSearch.Store(web)
|
||||
s.grokXSearch.Store(x)
|
||||
return nil
|
||||
}
|
||||
|
||||
// SetOnChannelsChanged 注册渠道数量变化钩子(渠道创建/删除成功后调用)。
|
||||
func (s *AiGatewayService) SetOnChannelsChanged(fn func(context.Context)) {
|
||||
s.onChannelsChanged = fn
|
||||
@@ -661,6 +750,40 @@ func (s *AiGatewayService) GatewayModels(ctx context.Context, group string) (aiw
|
||||
return list, nil
|
||||
}
|
||||
|
||||
// AggregatedModel 是聚合模型目录条目(设置页黑名单添加弹窗用)。
|
||||
type AggregatedModel struct {
|
||||
Name string `json:"name"`
|
||||
Capability string `json:"capability"`
|
||||
}
|
||||
|
||||
// AggregatedModels 返回启用渠道去重后的模型目录(含能力);空能力归一为 CHAT。
|
||||
// 与模型列表口径一致:「过滤弃用」开启时弃用模型不出现在目录中。
|
||||
func (s *AiGatewayService) AggregatedModels(ctx context.Context) ([]AggregatedModel, error) {
|
||||
q := s.db.WithContext(ctx).
|
||||
Joins("JOIN ai_channels ON ai_channels.id = ai_model_caches.channel_id AND ai_channels.enabled = ?", true)
|
||||
if s.FilterDeprecated() {
|
||||
q = q.Where("ai_model_caches.deprecated_at IS NULL")
|
||||
}
|
||||
var rows []model.AiModelCache
|
||||
if err := q.Order("ai_model_caches.name ASC").Find(&rows).Error; err != nil {
|
||||
return nil, fmt.Errorf("聚合模型目录: %w", err)
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
out := []AggregatedModel{}
|
||||
for _, r := range rows {
|
||||
if seen[r.Name] {
|
||||
continue
|
||||
}
|
||||
seen[r.Name] = true
|
||||
cap := r.Capability
|
||||
if cap == "" {
|
||||
cap = "CHAT"
|
||||
}
|
||||
out = append(out, AggregatedModel{Name: r.Name, Capability: cap})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// DeprecatingModels 返回 within 窗口内即将退役或即将弃用的在池模型(按名称去重):
|
||||
// 退役(TimeOnDemandRetired)才导致不可调用,单独标注;已过弃用日但未到退役日的
|
||||
// 模型仍可正常调用,不再反复告警;已过退役日的在同步层剔除,不会出现在池中。
|
||||
|
||||
@@ -1060,3 +1060,67 @@ func TestRespPassthroughStreamSwitchesChannel(t *testing.T) {
|
||||
t.Errorf("流内容未透传: %s", payload)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAiRuntimeSettings 断言流式保险丝与 grok 注入开关的缺省值、往返与持久化。
|
||||
func TestAiRuntimeSettings(t *testing.T) {
|
||||
gw, svc := newTestGateway(t, &gatewayStubClient{fakeClient: &fakeClient{tenancy: oci.TenancyInfo{Name: "t"}}})
|
||||
ctx := context.Background()
|
||||
|
||||
if on, kb := gw.StreamGuard(); !on || kb != 60 {
|
||||
t.Fatalf("保险丝缺省应为 开/60, got %v/%d", on, kb)
|
||||
}
|
||||
if web, x := gw.GrokSearch(); !web || !x {
|
||||
t.Fatalf("grok 注入缺省应双开, got %v/%v", web, x)
|
||||
}
|
||||
for _, bad := range []int{0, -1, 1025} {
|
||||
if err := gw.SetStreamGuard(ctx, true, bad); err == nil {
|
||||
t.Errorf("阈值 %d 应报错", bad)
|
||||
}
|
||||
}
|
||||
if err := gw.SetStreamGuard(ctx, false, 80); err != nil {
|
||||
t.Fatalf("SetStreamGuard: %v", err)
|
||||
}
|
||||
if err := gw.SetGrokSearch(ctx, false, true); err != nil {
|
||||
t.Fatalf("SetGrokSearch: %v", err)
|
||||
}
|
||||
gw2 := NewAiGatewayService(gw.db, svc, &gatewayStubClient{})
|
||||
if on, kb := gw2.StreamGuard(); on || kb != 80 {
|
||||
t.Errorf("重建后保险丝应为 关/80, got %v/%d", on, kb)
|
||||
}
|
||||
if web, x := gw2.GrokSearch(); web || !x {
|
||||
t.Errorf("重建后 grok 注入应为 关/开, got %v/%v", web, x)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAggregatedModelsFilterDeprecated 断言聚合目录与模型列表口径一致:
|
||||
// 「过滤弃用」开启时弃用模型不出现,关闭时出现;空能力归一为 CHAT。
|
||||
func TestAggregatedModelsFilterDeprecated(t *testing.T) {
|
||||
gw, svc := newTestGateway(t, &gatewayStubClient{fakeClient: &fakeClient{tenancy: oci.TenancyInfo{Name: "t"}}})
|
||||
cfg := importAliveConfig(t, svc)
|
||||
ch := seedChannel(t, gw, cfg.ID, "eu-frankfurt-1", 1, 1)
|
||||
dep := time.Now().Add(-24 * time.Hour)
|
||||
old := &model.AiModelCache{ChannelID: ch.ID, ModelOcid: "ocid1..dep", Name: "meta.llama-old",
|
||||
Vendor: "meta", SyncedAt: time.Now(), DeprecatedAt: &dep}
|
||||
if err := gw.db.Create(old).Error; err != nil {
|
||||
t.Fatalf("seed deprecated cache: %v", err)
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
items, err := gw.AggregatedModels(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("AggregatedModels: %v", err)
|
||||
}
|
||||
if len(items) != 2 || items[0].Capability == "" {
|
||||
t.Fatalf("开关关:应含弃用模型且能力归一, got %+v", items)
|
||||
}
|
||||
if err := gw.SetFilterDeprecated(ctx, true); err != nil {
|
||||
t.Fatalf("SetFilterDeprecated: %v", err)
|
||||
}
|
||||
items, err = gw.AggregatedModels(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("AggregatedModels(过滤): %v", err)
|
||||
}
|
||||
if len(items) != 1 || items[0].Name == "meta.llama-old" {
|
||||
t.Fatalf("开关开:弃用模型应被过滤, got %+v", items)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -320,10 +320,63 @@ func respToolDrop(tool map[string]any) (string, bool) {
|
||||
return "", false
|
||||
}
|
||||
|
||||
// RespStreamUpgradeLimit 是流式直通的请求体安全上限。实测(2026-07,xai 面):
|
||||
// 超过 ~82KB 的流式请求上游会在推理阶段掐断流(非流式不受影响,纯体积触发,
|
||||
// 与工具构成无关),预留余量取 76KB;超限时网关改走非流式上游并合成 SSE。
|
||||
const RespStreamUpgradeLimit = 76 * 1024
|
||||
// RespGuardBytes 返回请求体中 instructions 与 tools 两字段的原始字节数之和,
|
||||
// 流式保险丝据此判定。上游对二者合计 >≈64.5KB 的流式请求会在推理阶段静默断流
|
||||
// (纯 EOF,input 正文不计入;2026-07-16 实测仍存在),阈值由设置页 AI Tab 配置。
|
||||
// 解析失败返回 0(放行,交由上游正常报错)。
|
||||
func RespGuardBytes(body []byte) int {
|
||||
var probe struct {
|
||||
Instructions json.RawMessage `json:"instructions"`
|
||||
Tools json.RawMessage `json:"tools"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &probe); err != nil {
|
||||
return 0
|
||||
}
|
||||
return len(probe.Instructions) + len(probe.Tools)
|
||||
}
|
||||
|
||||
// RespInjectGrokTools 为 xai. 前缀模型默认注入服务端搜索工具:开关开启且请求
|
||||
// tools 中不存在同名工具时追加 {"type":"web_search"} / {"type":"x_search"};
|
||||
// 已存在(含任意参数形态)不覆盖。返回改写后 body 与注入清单(观测日志用);
|
||||
// 模型不匹配、两开关全关或解析失败时原样返回。
|
||||
func RespInjectGrokTools(body []byte, model string, web, x bool) ([]byte, []string) {
|
||||
if !strings.HasPrefix(model, "xai.") || (!web && !x) {
|
||||
return body, nil
|
||||
}
|
||||
dec := json.NewDecoder(strings.NewReader(string(body)))
|
||||
dec.UseNumber()
|
||||
var m map[string]any
|
||||
if err := dec.Decode(&m); err != nil {
|
||||
return body, nil
|
||||
}
|
||||
tools, _ := m["tools"].([]any)
|
||||
missing := map[string]bool{"web_search": web, "x_search": x}
|
||||
for _, t := range tools {
|
||||
tool, ok := t.(map[string]any)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if typ, _ := tool["type"].(string); missing[typ] {
|
||||
missing[typ] = false
|
||||
}
|
||||
}
|
||||
var injected []string
|
||||
for _, typ := range []string{"web_search", "x_search"} {
|
||||
if missing[typ] {
|
||||
tools = append(tools, map[string]any{"type": typ})
|
||||
injected = append(injected, typ)
|
||||
}
|
||||
}
|
||||
if len(injected) == 0 {
|
||||
return body, nil
|
||||
}
|
||||
m["tools"] = tools
|
||||
out, err := json.Marshal(m)
|
||||
if err != nil {
|
||||
return body, nil
|
||||
}
|
||||
return out, injected
|
||||
}
|
||||
|
||||
// RespDisableStream 把请求体的 stream 改为 false(流式升级回退用),其余字段
|
||||
// 原样保留。
|
||||
|
||||
@@ -2,6 +2,7 @@ package service
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
@@ -329,6 +330,71 @@ func TestRespStreamCompletedUsage(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestRespGuardBytes 断言保险丝计量:instructions+tools 原始字节和,缺字段计 0,
|
||||
// 解析失败返回 0 放行。
|
||||
func TestRespGuardBytes(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name, body string
|
||||
want int
|
||||
}{
|
||||
{"双字段", `{"instructions":"abcd","tools":[{"type":"web_search"}],"input":"xxxxxxxx"}`,
|
||||
len(`"abcd"`) + len(`[{"type":"web_search"}]`)},
|
||||
{"仅 instructions", `{"instructions":"abcd"}`, len(`"abcd"`)},
|
||||
{"均缺失 input 不计", `{"input":"xxxxxxxxxxxxxxxx"}`, 0},
|
||||
{"解析失败放行", `not-json`, 0},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
if got := RespGuardBytes([]byte(tc.body)); got != tc.want {
|
||||
t.Fatalf("RespGuardBytes() = %d, want %d", got, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestRespInjectGrokTools 断言 grok 搜索工具默认注入:仅 xai. 模型、开关可控、
|
||||
// 已带同名工具不覆盖、注入清单正确。
|
||||
func TestRespInjectGrokTools(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name, body, model string
|
||||
web, x bool
|
||||
wantInjected []string
|
||||
wantContains []string
|
||||
}{
|
||||
{"非 xai 不注入", `{"tools":[]}`, "meta.llama-3.3", true, true, nil, nil},
|
||||
{"双开无 tools 字段注入两个", `{"model":"xai.grok-4.3"}`, "xai.grok-4.3", true, true,
|
||||
[]string{"web_search", "x_search"}, []string{`"web_search"`, `"x_search"`}},
|
||||
{"已带 web_search 只注入 x_search", `{"tools":[{"type":"web_search","filters":{"x":1}}]}`,
|
||||
"xai.grok-4.3", true, true, []string{"x_search"}, []string{`"filters"`}},
|
||||
{"开关全关不注入", `{"tools":[]}`, "xai.grok-4.3", false, false, nil, nil},
|
||||
{"仅开 x_search", `{"tools":[]}`, "xai.grok-4.3", false, true, []string{"x_search"}, nil},
|
||||
{"解析失败原样", `not-json`, "xai.grok-4.3", true, true, nil, nil},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
out, injected := RespInjectGrokTools([]byte(tc.body), tc.model, tc.web, tc.x)
|
||||
if fmt.Sprint(injected) != fmt.Sprint(tc.wantInjected) {
|
||||
t.Fatalf("injected = %v, want %v", injected, tc.wantInjected)
|
||||
}
|
||||
if len(injected) == 0 && string(out) != tc.body {
|
||||
t.Fatalf("未注入时应原样返回: %s", out)
|
||||
}
|
||||
for _, sub := range append(tc.wantContains, toolTypes(injected)...) {
|
||||
if !strings.Contains(string(out), sub) {
|
||||
t.Fatalf("输出缺少 %s: %s", sub, out)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// toolTypes 把注入清单转为输出应包含的片段断言。
|
||||
func toolTypes(injected []string) []string {
|
||||
var out []string
|
||||
for _, typ := range injected {
|
||||
out = append(out, `{"type":"`+typ+`"}`)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// TestRespDisableStream 断言流式升级回退把 stream 置 false 且其余字段保留。
|
||||
func TestRespDisableStream(t *testing.T) {
|
||||
out, err := RespDisableStream([]byte(`{"model":"m","stream":true,"max_output_tokens":64}`))
|
||||
|
||||
Reference in New Issue
Block a user