From da7b29d2e3948d5de9a7e8e89a2500f718f11928 Mon Sep 17 00:00:00 2001 From: Wang Defa <1+wangdefa@noreply.gitea.bcde.io> Date: Thu, 16 Jul 2026 12:31:36 +0800 Subject: [PATCH] =?UTF-8?q?=E8=AE=BE=E7=BD=AE=E9=A1=B5AI=20Tab:=E7=BD=91?= =?UTF-8?q?=E5=85=B3=E8=AE=BE=E7=BD=AE=E5=85=A5=E5=BA=93=E3=80=81=E4=BF=9D?= =?UTF-8?q?=E9=99=A9=E4=B8=9D=E6=94=B9=E7=BB=B4=E5=BA=A6=E4=B8=8Egrok?= =?UTF-8?q?=E5=B7=A5=E5=85=B7=E6=B3=A8=E5=85=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/AI网关.md | 34 +++++-- docs/docs.go | 69 ++++++++++++- docs/swagger.json | 69 ++++++++++++- docs/swagger.yaml | 49 ++++++++- internal/api/aiadmin.go | 62 ++++++++++- internal/api/aigateway.go | 16 ++- internal/api/routes_ai.go | 1 + internal/service/aigateway.go | 147 ++++++++++++++++++++++++--- internal/service/aigateway_test.go | 64 ++++++++++++ internal/service/airesponses.go | 61 ++++++++++- internal/service/airesponses_test.go | 66 ++++++++++++ 11 files changed, 600 insertions(+), 38 deletions(-) diff --git a/docs/AI网关.md b/docs/AI网关.md index cc5c7e4..018b16a 100644 --- a/docs/AI网关.md +++ b/docs/AI网关.md @@ -167,21 +167,37 @@ Codex 工具兼容现状： ### 超大流式请求 +上游流式断流有两个独立触发维度，状态不同： + > [!WARNING] -> 2026-07 实测：OCI xAI 兼容面对**完整请求体超过约 82 KB** 的流式请求， -> 可能在推理阶段提前断流；该现象由请求体积触发，与工具构成或字符集无关， -> 同一请求改为非流式可正常完成。 +> **instructions + tools 合计超过约 64.5 KB** 的流式请求，上游会在推理阶段 +> 静默断连（纯 EOF，无 error / 终态事件）；同请求非流式总是成功，`input` +> 正文完全不计入。2026-07-13 定位（字节级二分），**2026-07-16 复测仍存在** +> （70.4 KB 断 / 59.7 KB 过，Chicago，API Key 与签名行为一致）。 -网关按协议采取不同保护： +> [!NOTE] +> **完整请求体超过约 82 KB**（含 input）的纯体积断流（2026-07-15 定位） +> **已被上游修复**：2026-07-16 复核 83 KB、真实 codex 形态 104.5 KB、200 KB、 +> 400 KB 流式均正常完成（Chicago 与 Phoenix 两区、签名与 API Key 两路径对照）。 + +网关当前行为： -- **Responses**：兼容改写后的请求体超过 **76 KB** 时，预防性改为非流式上游 - 请求，再合成最小 SSE 序列（`response.created` → - `response.output_item.done` → `response.completed`）；结果语义保留，但没有增量输出 - **Chat Completions / Messages**：若上游在客户端收到任何内容前断流，自动用 - 非流式重做，并合成对应 chunk / event 序列 + 非流式重做，并合成对应 chunk / event 序列（可兜住 64.5 KB 断流） +- **Responses**：直通协议中途无法透明重试（客户端已收到事件）。流式保险丝按 + `instructions + tools` 字节和判定：超过阈值时预防性改非流式上游 + 合成最小 + SSE 事件序列（`response.created` → `response.output_item.done` → + `response.completed`），语义保留但无增量输出。默认开、60 KB，可在 + **设置 → AI → 流式保险丝** 调整或关闭 - **已开始输出的流**：不能透明重试；调用日志会记录提前终止，客户端可能只拿到 部分事件 +### grok 服务端搜索工具默认注入 + +对 `xai.` 前缀模型的 Responses 请求，网关按开关默认注入 `web_search` / +`x_search` 工具；请求 tools 已包含同名工具时保持原样，不覆盖参数。默认双开， +可在 **设置 → AI → grok 服务端搜索工具** 关闭。 + ### ZDR 与文件输入 Responses 的 `input_file` 内容块（`file_url` / `file_data`）实测会被上游拒绝： @@ -239,7 +255,7 @@ Codex 工具兼容改写。请求会重新编码，不承诺字节级原样转 | `reasoning` | ➡️ | 整个对象保留；`effort` 不校验档位 | | `tool_choice` | ◐ | 普通形态保留；namespace 对象会重限定，全部工具被剥离时删除 | | `store` | 🔄 | 无论客户端传什么，上游请求都强制改写为 `false` | -| `stream` | ◐ | 支持 SSE；请求体超过 76 KB 时改为非流式上游并合成 SSE | +| `stream` | ◐ | 支持 SSE；网关按 instructions+tools 字节和触发预防性非流式回退（保险丝，默认开 60 KB，见[已知限制](#limitations)） | | 其余标准与未知顶层字段 | ➡️ | `context_management`、`include`、`metadata`、`prompt`、`prompt_cache_key`、`service_tier`、`truncation`、`user` 等均保留，由 OCI 决定是否接受 | diff --git a/docs/docs.go b/docs/docs.go index ea57b19..de90216 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -786,6 +786,27 @@ const docTemplate = `{ } } }, + "/api/v1/ai-model-catalog": { + "get": { + "security": [ + { + "BearerAuth": [] + } + ], + "tags": [ + "AI 管理" + ], + "summary": "聚合模型目录", + "responses": { + "200": { + "description": "OK", + "schema": { + "$ref": "#/definitions/internal_api.itemsResponse-oci-portal_internal_service_AggregatedModel" + } + } + } + } + }, "/api/v1/ai-models": { "get": { "security": [ @@ -839,7 +860,7 @@ const docTemplate = `{ "summary": "更新 AI 网关全局设置", "parameters": [ { - "description": "开启后已宣布弃用(即使未退役)的模型从列表与路由中排除", + "description": "全量提交;保险丝阈值限 1..1024 KB", "name": "body", "in": "body", "required": true, @@ -854,6 +875,15 @@ const docTemplate = `{ "schema": { "$ref": "#/definitions/internal_api.aiSettingsResponse" } + }, + "400": { + "description": "Bad Request", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } } } } @@ -6050,7 +6080,22 @@ const docTemplate = `{ "type": "object", "properties": { "filterDeprecated": { + "description": "FilterDeprecated 开启后已宣布弃用(即使未退役)的模型从列表与路由中排除", "type": "boolean" + }, + "grokWebSearch": { + "description": "GrokWebSearch / GrokXSearch 是 xai. 模型服务端搜索工具默认注入开关;\n请求 tools 已包含同名工具时不覆盖", + "type": "boolean" + }, + "grokXSearch": { + "type": "boolean" + }, + "streamGuardEnabled": { + "description": "StreamGuardEnabled / StreamGuardKB 是 Responses 流式保险丝:\ninstructions+tools 合计超阈值(KB)的流式请求改非流式上游并合成 SSE", + "type": "boolean" + }, + "streamGuardKB": { + "type": "integer" } } }, @@ -6621,6 +6666,17 @@ const docTemplate = `{ } } }, + "internal_api.itemsResponse-oci-portal_internal_service_AggregatedModel": { + "type": "object", + "properties": { + "items": { + "type": "array", + "items": { + "$ref": "#/definitions/oci-portal_internal_service.AggregatedModel" + } + } + } + }, "internal_api.itemsResponse-oci-portal_internal_service_NotifyChannelView": { "type": "object", "properties": { @@ -9358,6 +9414,17 @@ const docTemplate = `{ } } }, + "oci-portal_internal_service.AggregatedModel": { + "type": "object", + "properties": { + "capability": { + "type": "string" + }, + "name": { + "type": "string" + } + } + }, "oci-portal_internal_service.AuditEventsView": { "type": "object", "properties": { diff --git a/docs/swagger.json b/docs/swagger.json index 4d563b2..340e725 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -779,6 +779,27 @@ } } }, + "/api/v1/ai-model-catalog": { + "get": { + "security": [ + { + "BearerAuth": [] + } + ], + "tags": [ + "AI 管理" + ], + "summary": "聚合模型目录", + "responses": { + "200": { + "description": "OK", + "schema": { + "$ref": "#/definitions/internal_api.itemsResponse-oci-portal_internal_service_AggregatedModel" + } + } + } + } + }, "/api/v1/ai-models": { "get": { "security": [ @@ -832,7 +853,7 @@ "summary": "更新 AI 网关全局设置", "parameters": [ { - "description": "开启后已宣布弃用(即使未退役)的模型从列表与路由中排除", + "description": "全量提交;保险丝阈值限 1..1024 KB", "name": "body", "in": "body", "required": true, @@ -847,6 +868,15 @@ "schema": { "$ref": "#/definitions/internal_api.aiSettingsResponse" } + }, + "400": { + "description": "Bad Request", + "schema": { + "type": "object", + "additionalProperties": { + "type": "string" + } + } } } } @@ -6043,7 +6073,22 @@ "type": "object", "properties": { "filterDeprecated": { + "description": "FilterDeprecated 开启后已宣布弃用(即使未退役)的模型从列表与路由中排除", "type": "boolean" + }, + "grokWebSearch": { + "description": "GrokWebSearch / GrokXSearch 是 xai. 模型服务端搜索工具默认注入开关;\n请求 tools 已包含同名工具时不覆盖", + "type": "boolean" + }, + "grokXSearch": { + "type": "boolean" + }, + "streamGuardEnabled": { + "description": "StreamGuardEnabled / StreamGuardKB 是 Responses 流式保险丝:\ninstructions+tools 合计超阈值(KB)的流式请求改非流式上游并合成 SSE", + "type": "boolean" + }, + "streamGuardKB": { + "type": "integer" } } }, @@ -6614,6 +6659,17 @@ } } }, + "internal_api.itemsResponse-oci-portal_internal_service_AggregatedModel": { + "type": "object", + "properties": { + "items": { + "type": "array", + "items": { + "$ref": "#/definitions/oci-portal_internal_service.AggregatedModel" + } + } + } + }, "internal_api.itemsResponse-oci-portal_internal_service_NotifyChannelView": { "type": "object", "properties": { @@ -9351,6 +9407,17 @@ } } }, + "oci-portal_internal_service.AggregatedModel": { + "type": "object", + "properties": { + "capability": { + "type": "string" + }, + "name": { + "type": "string" + } + } + }, "oci-portal_internal_service.AuditEventsView": { "type": "object", "properties": { diff --git a/docs/swagger.yaml b/docs/swagger.yaml index e227ee4..ac49a10 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -49,7 +49,22 @@ definitions: internal_api.aiSettingsResponse: properties: filterDeprecated: + description: FilterDeprecated 开启后已宣布弃用(即使未退役)的模型从列表与路由中排除 type: boolean + grokWebSearch: + description: |- + GrokWebSearch / GrokXSearch 是 xai. 模型服务端搜索工具默认注入开关; + 请求 tools 已包含同名工具时不覆盖 + type: boolean + grokXSearch: + type: boolean + streamGuardEnabled: + description: |- + StreamGuardEnabled / StreamGuardKB 是 Responses 流式保险丝: + instructions+tools 合计超阈值(KB)的流式请求改非流式上游并合成 SSE + type: boolean + streamGuardKB: + type: integer type: object internal_api.attachBootVolumeRequest: properties: @@ -427,6 +442,13 @@ definitions: $ref: '#/definitions/oci-portal_internal_model.UserIdentity' type: array type: object + internal_api.itemsResponse-oci-portal_internal_service_AggregatedModel: + properties: + items: + items: + $ref: '#/definitions/oci-portal_internal_service.AggregatedModel' + type: array + type: object internal_api.itemsResponse-oci-portal_internal_service_NotifyChannelView: properties: items: @@ -2232,6 +2254,13 @@ definitions: description: 卷本身的名称（attachment 名是自动生成的） type: string type: object + oci-portal_internal_service.AggregatedModel: + properties: + capability: + type: string + name: + type: string + type: object oci-portal_internal_service.AuditEventsView: properties: cursor: @@ -3135,6 +3164,18 @@ paths: summary: AI 调用日志列表 tags: - AI 管理 + /api/v1/ai-model-catalog: + get: + responses: + "200": + description: OK + schema: + $ref: '#/definitions/internal_api.itemsResponse-oci-portal_internal_service_AggregatedModel' + security: + - BearerAuth: [] + summary: 聚合模型目录 + tags: + - AI 管理 /api/v1/ai-models: get: responses: @@ -3161,7 +3202,7 @@ paths: - AI 管理 put: parameters: - - description: 开启后已宣布弃用(即使未退役)的模型从列表与路由中排除 + - description: 全量提交;保险丝阈值限 1..1024 KB in: body name: body required: true @@ -3172,6 +3213,12 @@ paths: description: OK schema: $ref: '#/definitions/internal_api.aiSettingsResponse' + "400": + description: Bad Request + schema: + additionalProperties: + type: string + type: object security: - BearerAuth: [] summary: 更新 AI 网关全局设置 diff --git a/internal/api/aiadmin.go b/internal/api/aiadmin.go index 098798a..92968b3 100644 --- a/internal/api/aiadmin.go +++ b/internal/api/aiadmin.go @@ -422,9 +422,47 @@ func aiPathID(c *gin.Context) (uint, bool) { return uint(id), true } +// modelCatalog 返回启用渠道聚合去重后的模型目录(含能力),黑名单添加弹窗用。 +// +// @Summary 聚合模型目录 +// @Tags AI 管理 +// @Success 200 {object} itemsResponse[service.AggregatedModel] +// @Security BearerAuth +// @Router /api/v1/ai-model-catalog [get] +func (h *aiAdminHandler) modelCatalog(c *gin.Context) { + items, err := h.gw.AggregatedModels(c.Request.Context()) + if err != nil { + respondError(c, err) + return + } + c.JSON(http.StatusOK, gin.H{"items": items}) +} + // aiSettingsResponse 是 AI 网关全局设置(文档与响应共用)。 type aiSettingsResponse struct { + // FilterDeprecated 开启后已宣布弃用(即使未退役)的模型从列表与路由中排除 FilterDeprecated bool `json:"filterDeprecated"` + // StreamGuardEnabled / StreamGuardKB 是 Responses 流式保险丝: + // instructions+tools 合计超阈值(KB)的流式请求改非流式上游并合成 SSE + StreamGuardEnabled bool `json:"streamGuardEnabled"` + StreamGuardKB int `json:"streamGuardKB"` + // GrokWebSearch / GrokXSearch 是 xai. 模型服务端搜索工具默认注入开关; + // 请求 tools 已包含同名工具时不覆盖 + GrokWebSearch bool `json:"grokWebSearch"` + GrokXSearch bool `json:"grokXSearch"` +} + +// currentAiSettings 汇总网关运行时设置为响应体。 +func (h *aiAdminHandler) currentAiSettings() aiSettingsResponse { + guardOn, guardKB := h.gw.StreamGuard() + web, x := h.gw.GrokSearch() + return aiSettingsResponse{ + FilterDeprecated: h.gw.FilterDeprecated(), + StreamGuardEnabled: guardOn, + StreamGuardKB: guardKB, + GrokWebSearch: web, + GrokXSearch: x, + } } // aiSettings 返回 AI 网关全局设置。 @@ -435,15 +473,16 @@ type aiSettingsResponse struct { // @Security BearerAuth // @Router /api/v1/ai-settings [get] func (h *aiAdminHandler) aiSettings(c *gin.Context) { - c.JSON(http.StatusOK, aiSettingsResponse{FilterDeprecated: h.gw.FilterDeprecated()}) + c.JSON(http.StatusOK, h.currentAiSettings()) } -// updateAiSettings 更新 AI 网关全局设置(当前仅「过滤弃用模型」开关)。 +// updateAiSettings 更新 AI 网关全局设置(过滤弃用/流式保险丝/grok 搜索工具默认注入)。 // // @Summary 更新 AI 网关全局设置 // @Tags AI 管理 -// @Param body body aiSettingsResponse true "开启后已宣布弃用(即使未退役)的模型从列表与路由中排除" +// @Param body body aiSettingsResponse true "全量提交;保险丝阈值限 1..1024 KB" // @Success 200 {object} aiSettingsResponse +// @Failure 400 {object} map[string]string // @Security BearerAuth // @Router /api/v1/ai-settings [put] func (h *aiAdminHandler) updateAiSettings(c *gin.Context) { @@ -452,9 +491,22 @@ func (h *aiAdminHandler) updateAiSettings(c *gin.Context) { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } - if err := h.gw.SetFilterDeprecated(c.Request.Context(), req.FilterDeprecated); err != nil { + if req.StreamGuardKB < 1 || req.StreamGuardKB > 1024 { + c.JSON(http.StatusBadRequest, gin.H{"error": "streamGuardKB 须在 1..1024"}) + return + } + ctx := c.Request.Context() + if err := h.gw.SetFilterDeprecated(ctx, req.FilterDeprecated); err != nil { respondError(c, err) return } - c.JSON(http.StatusOK, aiSettingsResponse{FilterDeprecated: h.gw.FilterDeprecated()}) + if err := h.gw.SetStreamGuard(ctx, req.StreamGuardEnabled, req.StreamGuardKB); err != nil { + respondError(c, err) + return + } + if err := h.gw.SetGrokSearch(ctx, req.GrokWebSearch, req.GrokXSearch); err != nil { + respondError(c, err) + return + } + c.JSON(http.StatusOK, h.currentAiSettings()) } diff --git a/internal/api/aigateway.go b/internal/api/aigateway.go index a89782f..24ed34b 100644 --- a/internal/api/aigateway.go +++ b/internal/api/aigateway.go @@ -587,8 +587,13 @@ func (h *aiGatewayHandler) responsesPassthrough(c *gin.Context, raw []byte, req return } logRespCompat(req.Model, compat) + web, x := h.gw.GrokSearch() + if injectedBody, injected := service.RespInjectGrokTools(body, req.Model, web, x); len(injected) > 0 { + body = injectedBody + log.Printf("responses 直通(model=%s): 默认注入 %s", req.Model, strings.Join(injected, ", ")) + } if req.Stream { - if len(body) > service.RespStreamUpgradeLimit { + if on, kb := h.gw.StreamGuard(); on && service.RespGuardBytes(body) > kb*1024 { h.responsesStreamUpgrade(c, body, req, compat) return } @@ -630,9 +635,9 @@ func logRespCompat(model string, compat service.RespCompat) { log.Printf("responses 直通(model=%s): %s", model, strings.Join(parts, "; ")) } -// responsesStreamUpgrade 流式升级回退:上游对超大流式请求会在推理途中掐断 -// (实测 >~82KB,见 RespStreamUpgradeLimit),超限时改调非流式上游拿完整响应, -// 本地合成最小 SSE 事件序列回给客户端;丢失增量输出,换会话不中断。 +// responsesStreamUpgrade 流式升级回退:instructions+tools 合计超过保险丝阈值 +// (设置页 AI Tab 配置,默认开 60KB;上游对 >≈64.5KB 会静默断流)时改调非流式 +// 上游拿完整响应,本地合成最小 SSE 事件序列回给客户端;丢失增量输出,换会话不中断。 func (h *aiGatewayHandler) responsesStreamUpgrade(c *gin.Context, body []byte, req aiwire.RespRequest, compat service.RespCompat) { start := time.Now() nsBody, err := service.RespDisableStream(body) @@ -640,7 +645,8 @@ func (h *aiGatewayHandler) responsesStreamUpgrade(c *gin.Context, body []byte, r aiError(c, http.StatusBadRequest, "invalid_request_error", err.Error()) return } - log.Printf("responses 直通(model=%s): 请求体 %dKB 超流式安全上限,改走非流式合成 SSE", req.Model, len(body)/1024) + log.Printf("responses 直通(model=%s): instructions+tools %dKB 超保险丝阈值,改走非流式合成 SSE", + req.Model, service.RespGuardBytes(body)/1024) payload, meta, err := h.gw.RespPassthrough(c.Request.Context(), nsBody, req.Model, keyGroup(c)) entry := h.logEntry(c, "responses", req.Model, true, meta, start) if err != nil { diff --git a/internal/api/routes_ai.go b/internal/api/routes_ai.go index 614c43e..eef3a4d 100644 --- a/internal/api/routes_ai.go +++ b/internal/api/routes_ai.go @@ -39,6 +39,7 @@ func registerAiAdmin(secured *gin.RouterGroup, aiGateway *service.AiGatewayServi secured.GET("/ai-channels/:id/models", aiadmin.listChannelModels) secured.POST("/ai-channels/:id/test-model", aiadmin.testChannelModel) secured.GET("/ai-models", aiadmin.gatewayModels) + secured.GET("/ai-model-catalog", aiadmin.modelCatalog) secured.GET("/ai-settings", aiadmin.aiSettings) secured.PUT("/ai-settings", aiadmin.updateAiSettings) secured.GET("/ai-blacklist", aiadmin.listBlacklist) diff --git a/internal/service/aigateway.go b/internal/service/aigateway.go index 8303507..4558cda 100644 --- a/internal/service/aigateway.go +++ b/internal/service/aigateway.go @@ -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)才导致不可调用,单独标注;已过弃用日但未到退役日的 // 模型仍可正常调用,不再反复告警;已过退役日的在同步层剔除,不会出现在池中。 diff --git a/internal/service/aigateway_test.go b/internal/service/aigateway_test.go index 48dc12e..88e70ad 100644 --- a/internal/service/aigateway_test.go +++ b/internal/service/aigateway_test.go @@ -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) + } +} diff --git a/internal/service/airesponses.go b/internal/service/airesponses.go index a07fd78..0efcd64 100644 --- a/internal/service/airesponses.go +++ b/internal/service/airesponses.go @@ -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(流式升级回退用),其余字段 // 原样保留。 diff --git a/internal/service/airesponses_test.go b/internal/service/airesponses_test.go index 187d91a..f32f65a 100644 --- a/internal/service/airesponses_test.go +++ b/internal/service/airesponses_test.go @@ -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}`))