diff --git a/docs/docs.go b/docs/docs.go index 65feb5d..5e77fa3 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -860,7 +860,7 @@ const docTemplate = `{ "summary": "更新 AI 网关全局设置", "parameters": [ { - "description": "全量提交;保险丝阈值限 1..1024 KB", + "description": "全量提交;保险丝阈值限 1..1024 KB,上游无响应预算限 30..900 秒", "name": "body", "in": "body", "required": true, @@ -6102,6 +6102,10 @@ const docTemplate = `{ }, "streamGuardKB": { "type": "integer" + }, + "upstreamWaitSeconds": { + "description": "UpstreamWaitSeconds 是 responses 直通的上游无响应预算(秒):非流式为单次\n尝试总超时,流式为等待响应头预算;multi-agent/搜索类模型需远超 60s", + "type": "integer" } } }, diff --git a/docs/swagger.json b/docs/swagger.json index 3245420..142bc88 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -853,7 +853,7 @@ "summary": "更新 AI 网关全局设置", "parameters": [ { - "description": "全量提交;保险丝阈值限 1..1024 KB", + "description": "全量提交;保险丝阈值限 1..1024 KB,上游无响应预算限 30..900 秒", "name": "body", "in": "body", "required": true, @@ -6095,6 +6095,10 @@ }, "streamGuardKB": { "type": "integer" + }, + "upstreamWaitSeconds": { + "description": "UpstreamWaitSeconds 是 responses 直通的上游无响应预算(秒):非流式为单次\n尝试总超时,流式为等待响应头预算;multi-agent/搜索类模型需远超 60s", + "type": "integer" } } }, diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 385a7ea..dcbdb53 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -65,6 +65,11 @@ definitions: type: boolean streamGuardKB: type: integer + upstreamWaitSeconds: + description: |- + UpstreamWaitSeconds 是 responses 直通的上游无响应预算(秒):非流式为单次 + 尝试总超时,流式为等待响应头预算;multi-agent/搜索类模型需远超 60s + type: integer type: object internal_api.attachBootVolumeRequest: properties: @@ -3204,7 +3209,7 @@ paths: - AI 管理 put: parameters: - - description: 全量提交;保险丝阈值限 1..1024 KB + - description: 全量提交;保险丝阈值限 1..1024 KB,上游无响应预算限 30..900 秒 in: body name: body required: true diff --git a/internal/api/aiadmin.go b/internal/api/aiadmin.go index 92968b3..49d7bf9 100644 --- a/internal/api/aiadmin.go +++ b/internal/api/aiadmin.go @@ -3,6 +3,7 @@ package api import ( "net/http" "strconv" + "time" "github.com/gin-gonic/gin" @@ -450,6 +451,9 @@ type aiSettingsResponse struct { // 请求 tools 已包含同名工具时不覆盖 GrokWebSearch bool `json:"grokWebSearch"` GrokXSearch bool `json:"grokXSearch"` + // UpstreamWaitSeconds 是 responses 直通的上游无响应预算(秒):非流式为单次 + // 尝试总超时,流式为等待响应头预算;multi-agent/搜索类模型需远超 60s + UpstreamWaitSeconds int `json:"upstreamWaitSeconds"` } // currentAiSettings 汇总网关运行时设置为响应体。 @@ -457,11 +461,12 @@ 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, + FilterDeprecated: h.gw.FilterDeprecated(), + StreamGuardEnabled: guardOn, + StreamGuardKB: guardKB, + GrokWebSearch: web, + GrokXSearch: x, + UpstreamWaitSeconds: int(h.gw.UpstreamWait() / time.Second), } } @@ -480,7 +485,7 @@ func (h *aiAdminHandler) aiSettings(c *gin.Context) { // // @Summary 更新 AI 网关全局设置 // @Tags AI 管理 -// @Param body body aiSettingsResponse true "全量提交;保险丝阈值限 1..1024 KB" +// @Param body body aiSettingsResponse true "全量提交;保险丝阈值限 1..1024 KB,上游无响应预算限 30..900 秒" // @Success 200 {object} aiSettingsResponse // @Failure 400 {object} map[string]string // @Security BearerAuth @@ -495,6 +500,10 @@ func (h *aiAdminHandler) updateAiSettings(c *gin.Context) { c.JSON(http.StatusBadRequest, gin.H{"error": "streamGuardKB 须在 1..1024"}) return } + if req.UpstreamWaitSeconds < 30 || req.UpstreamWaitSeconds > 900 { + c.JSON(http.StatusBadRequest, gin.H{"error": "upstreamWaitSeconds 须在 30..900"}) + return + } ctx := c.Request.Context() if err := h.gw.SetFilterDeprecated(ctx, req.FilterDeprecated); err != nil { respondError(c, err) @@ -508,5 +517,9 @@ func (h *aiAdminHandler) updateAiSettings(c *gin.Context) { respondError(c, err) return } + if err := h.gw.SetUpstreamWait(ctx, req.UpstreamWaitSeconds); err != nil { + respondError(c, err) + return + } c.JSON(http.StatusOK, h.currentAiSettings()) } diff --git a/internal/oci/client.go b/internal/oci/client.go index 7be374a..f4afc61 100644 --- a/internal/oci/client.go +++ b/internal/oci/client.go @@ -102,10 +102,12 @@ type Client interface { ListGenAiModels(ctx context.Context, cred Credentials, region string) ([]GenAiModel, error) GenAiProbeChat(ctx context.Context, cred Credentials, region, modelOcid, modelName string) (int, error) GenAiEmbed(ctx context.Context, cred Credentials, region, modelOcid string, inputs []string, dimensions *int) ([][]float32, *aiwire.Usage, error) - // GenAiCompatResponses 直通 OpenAI Responses 请求体到 /actions/v1/responses(xAI 服务端工具通路)。 - GenAiCompatResponses(ctx context.Context, cred Credentials, region string, body []byte) ([]byte, error) - // GenAiCompatResponsesStream 流式直通 /actions/v1/responses,建立成功返回 SSE body。 - GenAiCompatResponsesStream(ctx context.Context, cred Credentials, region string, body []byte) (io.ReadCloser, error) + // GenAiCompatResponses 直通 OpenAI Responses 请求体到 /actions/v1/responses(xAI 服务端工具通路); + // wait 是上游无响应预算(整请求总超时),multi-agent/搜索类模型需远超 SDK 默认 60s。 + GenAiCompatResponses(ctx context.Context, cred Credentials, region string, body []byte, wait time.Duration) ([]byte, error) + // GenAiCompatResponsesStream 流式直通 /actions/v1/responses,建立成功返回 SSE body; + // wait 仅约束等待响应头阶段,建立后的流生命周期由 ctx 决定。 + GenAiCompatResponsesStream(ctx context.Context, cred Credentials, region string, body []byte, wait time.Duration) (io.ReadCloser, error) // GenAiCompatSpeech 直通 OpenAI Audio Speech 请求体到 /openai/v1/audio/speech,返回音频与 Content-Type。 GenAiCompatSpeech(ctx context.Context, cred Credentials, region string, body []byte) ([]byte, string, error) // GenAiRerank 文档重排,返回按相关度排序的下标与得分。 diff --git a/internal/oci/genai.go b/internal/oci/genai.go index 6c3a34a..4830b0e 100644 --- a/internal/oci/genai.go +++ b/internal/oci/genai.go @@ -158,7 +158,8 @@ func (c *RealClient) GenAiProbeChat(ctx context.Context, cred Credentials, regio if err != nil { return 0, err } - if _, err = c.GenAiCompatResponses(ctx, cred, region, body); err == nil { + // 探测追求快速失败,沿用 SDK 默认量级的 60s 预算即可 + if _, err = c.GenAiCompatResponses(ctx, cred, region, body, 60*time.Second); err == nil { return http.StatusOK, nil } if status, ok := ServiceStatus(err); ok { diff --git a/internal/oci/genai_responses.go b/internal/oci/genai_responses.go index 36eecb1..7136c12 100644 --- a/internal/oci/genai_responses.go +++ b/internal/oci/genai_responses.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "net/http" + "time" "github.com/oracle/oci-go-sdk/v65/common" ) @@ -13,18 +14,19 @@ import ( // compatResponsesLimit 限制直通响应体大小;web_search 输出含多段引用,给足余量。 const compatResponsesLimit = int64(8 << 20) -// GenAiCompatResponses 实现 Client:把 OpenAI Responses 请求体直通到 OCI -// `/20231130/actions/v1/responses`(IAM 签名)。xAI 服务端工具(web_search / -// x_search / code_interpreter)与 mcp 已被 Oracle 文档正式支持,工具参数与限制 -// 遵循 xAI 规格;调用方须自行校验并改写请求体(store/stream)。 -func (c *RealClient) GenAiCompatResponses(ctx context.Context, cred Credentials, region string, body []byte) ([]byte, error) { - ic, err := c.genAiInferenceClient(cred, region) - if err != nil { - return nil, err +// dispatcherWithTimeout 把 dispatcher 换成指定总超时的拷贝(保留 Transport, +// 代理链路不受影响);timeout=0 表示无总超时(流式读 body 不能有总时限)。 +// 非 *http.Client 的自定义 dispatcher 保持原样,维持既有超时行为。 +func dispatcherWithTimeout(d common.HTTPRequestDispatcher, timeout time.Duration) common.HTTPRequestDispatcher { + hc, ok := d.(*http.Client) + if !ok { + return d } - client := ic.BaseClient - common.UpdateEndpointTemplateForOptions(&client) - common.SetMissingTemplateParams(&client) + return &http.Client{Transport: hc.Transport, Timeout: timeout} +} + +// newCompatResponsesRequest 构造 /actions/v1/responses 直通请求。 +func newCompatResponsesRequest(ctx context.Context, cred Credentials, body []byte) (*http.Request, error) { request, err := http.NewRequestWithContext(ctx, http.MethodPost, "/actions/v1/responses", bytes.NewReader(body)) if err != nil { return nil, fmt.Errorf("build compat responses request: %w", err) @@ -32,6 +34,27 @@ func (c *RealClient) GenAiCompatResponses(ctx context.Context, cred Credentials, request.Header.Set("Content-Type", "application/json") request.Header.Set("CompartmentId", cred.TenancyOCID) request.Header.Set("opc-compartment-id", cred.TenancyOCID) + return request, nil +} + +// GenAiCompatResponses 实现 Client:把 OpenAI Responses 请求体直通到 OCI +// `/20231130/actions/v1/responses`(IAM 签名)。xAI 服务端工具(web_search / +// x_search / code_interpreter)与 mcp 已被 Oracle 文档正式支持,工具参数与限制 +// 遵循 xAI 规格;调用方须自行校验并改写请求体(store/stream)。 +// wait 为整请求总超时:非流式上游要等全部生成完才回响应头,SDK 默认 60s 会掐断慢模型。 +func (c *RealClient) GenAiCompatResponses(ctx context.Context, cred Credentials, region string, body []byte, wait time.Duration) ([]byte, error) { + ic, err := c.genAiInferenceClient(cred, region) + if err != nil { + return nil, err + } + client := ic.BaseClient + common.UpdateEndpointTemplateForOptions(&client) + common.SetMissingTemplateParams(&client) + client.HTTPClient = dispatcherWithTimeout(client.HTTPClient, wait) + request, err := newCompatResponsesRequest(ctx, cred, body) + if err != nil { + return nil, err + } response, err := client.Call(ctx, request) if err != nil { return nil, err @@ -44,10 +67,46 @@ func (c *RealClient) GenAiCompatResponses(ctx context.Context, cred Credentials, return payload, nil } +// cancelReadCloser 在流关闭时同步取消建立阶段派生的 ctx,避免其随流生命周期泄漏。 +type cancelReadCloser struct { + io.ReadCloser + cancel context.CancelFunc +} + +func (c *cancelReadCloser) Close() error { + c.cancel() + return c.ReadCloser.Close() +} + +// httpCaller 抽象 BaseClient.Call,便于对预算逻辑做无签名单测。 +type httpCaller interface { + Call(ctx context.Context, request *http.Request) (*http.Response, error) +} + +// callWithHeaderBudget 以 wait 为等待响应头预算执行调用:预算内未返回则取消 +// 请求(SDK Call 会把 ctx 重绑到请求);响应头到达即解除预算,之后流的生命 +// 周期由 ctx 决定,返回的流 Close 时同步取消派生 ctx。 +func callWithHeaderBudget(ctx context.Context, c httpCaller, req *http.Request, wait time.Duration) (io.ReadCloser, error) { + callCtx, cancel := context.WithCancel(ctx) + timer := time.AfterFunc(wait, cancel) + response, err := c.Call(callCtx, req) + timer.Stop() + if err != nil { + if response != nil && response.Body != nil { + response.Body.Close() + } + cancel() + return nil, err + } + return &cancelReadCloser{ReadCloser: response.Body, cancel: cancel}, nil +} + // GenAiCompatResponsesStream 实现 Client:以流式直通 OCI `/actions/v1/responses`, // 建立成功(2xx)返回 SSE body(调用方负责 Close);建立失败返回 SDK ServiceError, // 与既有渠道切换/熔断错误分类兼容。请求体须由调用方置 stream:true。 -func (c *RealClient) GenAiCompatResponsesStream(ctx context.Context, cred Credentials, region string, body []byte) (io.ReadCloser, error) { +// 总超时置 0(SSE 读 body 不能有总时限);wait 以定时取消模拟等待响应头预算, +// 响应头到达即解除,此后流的生命周期完全由 ctx(下游客户端断开)决定。 +func (c *RealClient) GenAiCompatResponsesStream(ctx context.Context, cred Credentials, region string, body []byte, wait time.Duration) (io.ReadCloser, error) { ic, err := c.genAiInferenceClient(cred, region) if err != nil { return nil, err @@ -55,19 +114,10 @@ func (c *RealClient) GenAiCompatResponsesStream(ctx context.Context, cred Creden client := ic.BaseClient common.UpdateEndpointTemplateForOptions(&client) common.SetMissingTemplateParams(&client) - request, err := http.NewRequestWithContext(ctx, http.MethodPost, "/actions/v1/responses", bytes.NewReader(body)) + client.HTTPClient = dispatcherWithTimeout(client.HTTPClient, 0) + request, err := newCompatResponsesRequest(ctx, cred, body) if err != nil { - return nil, fmt.Errorf("build compat responses stream request: %w", err) - } - request.Header.Set("Content-Type", "application/json") - request.Header.Set("CompartmentId", cred.TenancyOCID) - request.Header.Set("opc-compartment-id", cred.TenancyOCID) - response, err := client.Call(ctx, request) - if err != nil { - if response != nil && response.Body != nil { - response.Body.Close() - } return nil, err } - return response.Body, nil + return callWithHeaderBudget(ctx, client, request, wait) } diff --git a/internal/oci/genai_responses_test.go b/internal/oci/genai_responses_test.go new file mode 100644 index 0000000..2ac28b8 --- /dev/null +++ b/internal/oci/genai_responses_test.go @@ -0,0 +1,141 @@ +package oci + +import ( + "context" + "errors" + "io" + "net/http" + "strings" + "testing" + "time" + + "github.com/oracle/oci-go-sdk/v65/common" +) + +// staticDispatcher 是非 *http.Client 的自定义 dispatcher,用于降级分支。 +type staticDispatcher struct{} + +func (staticDispatcher) Do(*http.Request) (*http.Response, error) { return nil, nil } + +func TestDispatcherWithTimeout(t *testing.T) { + tr := &http.Transport{} + tests := []struct { + name string + in common.HTTPRequestDispatcher + timeout time.Duration + check func(t *testing.T, out common.HTTPRequestDispatcher) + }{ + { + name: "http.Client 换总超时并保留 Transport", in: &http.Client{Transport: tr, Timeout: 60 * time.Second}, + timeout: 300 * time.Second, + check: func(t *testing.T, out common.HTTPRequestDispatcher) { + hc, ok := out.(*http.Client) + if !ok || hc.Timeout != 300*time.Second || hc.Transport != tr { + t.Fatalf("期望拷贝 client 且 Timeout=300s、Transport 保留, 得到 %#v", out) + } + }, + }, + { + name: "timeout=0 表示无总超时", in: &http.Client{Timeout: 60 * time.Second}, timeout: 0, + check: func(t *testing.T, out common.HTTPRequestDispatcher) { + if hc := out.(*http.Client); hc.Timeout != 0 { + t.Fatalf("期望 Timeout=0, 得到 %v", hc.Timeout) + } + }, + }, + { + name: "非 http.Client 原样返回", in: staticDispatcher{}, timeout: 300 * time.Second, + check: func(t *testing.T, out common.HTTPRequestDispatcher) { + if _, ok := out.(staticDispatcher); !ok { + t.Fatalf("期望原样返回自定义 dispatcher, 得到 %#v", out) + } + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { tt.check(t, dispatcherWithTimeout(tt.in, tt.timeout)) }) + } +} + +// ctxReader 模拟真实 HTTP body:请求 ctx 取消后读即失败。 +type ctxReader struct { + ctx context.Context + r io.Reader +} + +func (c *ctxReader) Read(p []byte) (int, error) { + if err := c.ctx.Err(); err != nil { + return 0, err + } + return c.r.Read(p) +} + +// fakeCaller 在 delay 后返回响应;期间 ctx 取消则按真实行为返回 ctx 错误。 +type fakeCaller struct { + delay time.Duration + body string + gotCtx context.Context + failErr error +} + +func (f *fakeCaller) Call(ctx context.Context, _ *http.Request) (*http.Response, error) { + f.gotCtx = ctx + if f.failErr != nil { + return nil, f.failErr + } + select { + case <-time.After(f.delay): + body := &ctxReader{ctx: ctx, r: strings.NewReader(f.body)} + return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(body)}, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +func TestCallWithHeaderBudget(t *testing.T) { + req, _ := http.NewRequest(http.MethodPost, "/actions/v1/responses", nil) + t.Run("预算内返回响应头后长读不受预算影响", func(t *testing.T) { + fc := &fakeCaller{delay: 0, body: "data: hello"} + stream, err := callWithHeaderBudget(context.Background(), fc, req, 30*time.Millisecond) + if err != nil { + t.Fatalf("callWithHeaderBudget = %v", err) + } + defer stream.Close() + time.Sleep(90 * time.Millisecond) // 远超预算,验证响应头到达后预算已解除 + payload, err := io.ReadAll(stream) + if err != nil || string(payload) != "data: hello" { + t.Fatalf("预算解除后读流 = %q, %v; 期望完整 body", payload, err) + } + }) + t.Run("超预算未返回响应头即取消", func(t *testing.T) { + fc := &fakeCaller{delay: time.Minute} + start := time.Now() + _, err := callWithHeaderBudget(context.Background(), fc, req, 30*time.Millisecond) + if !errors.Is(err, context.Canceled) { + t.Fatalf("期望 context.Canceled, 得到 %v", err) + } + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Fatalf("取消耗时 %v, 未受预算约束", elapsed) + } + }) + t.Run("关闭流时取消派生 ctx", func(t *testing.T) { + fc := &fakeCaller{delay: 0, body: "x"} + stream, err := callWithHeaderBudget(context.Background(), fc, req, time.Minute) + if err != nil { + t.Fatalf("callWithHeaderBudget = %v", err) + } + stream.Close() + if fc.gotCtx.Err() == nil { + t.Fatal("Close 后派生 ctx 应已取消") + } + }) + t.Run("建立失败时同样取消派生 ctx", func(t *testing.T) { + fc := &fakeCaller{failErr: errors.New("boom")} + if _, err := callWithHeaderBudget(context.Background(), fc, req, time.Minute); err == nil { + t.Fatal("期望建立失败") + } + if fc.gotCtx.Err() == nil { + t.Fatal("失败路径派生 ctx 应已取消") + } + }) +} diff --git a/internal/oci/proxyhttp.go b/internal/oci/proxyhttp.go index 6312aa0..c0344e5 100644 --- a/internal/oci/proxyhttp.go +++ b/internal/oci/proxyhttp.go @@ -2,6 +2,7 @@ package oci import ( "fmt" + "net" "net/http" "net/url" "time" @@ -23,6 +24,13 @@ type ProxySpec struct { // proxyClientTimeout 与 SDK 默认 HTTPClient 超时保持一致。 const proxyClientTimeout = 60 * time.Second +// 阶段超时对齐 SDK 直连 Transport 模板(transport_template_provider):连不上的 +// 代理快速失败,而不是拖满总超时;responses 直通去掉总超时后这是建立阶段的兜底之一。 +const ( + proxyDialTimeout = 30 * time.Second + proxyTLSHandshakeTimeout = 10 * time.Second +) + // applyProxy 在 SDK client 构造后统一挂出站代理;未关联代理时不动默认配置。 // 所有 New*ClientWithConfigurationProvider 调用点构造成功后都必须经过这里。 func applyProxy(base *common.BaseClient, cred Credentials) { @@ -52,18 +60,23 @@ func HTTPClientFor(p *ProxySpec) *http.Client { // transportFor 构造代理 Transport:http / https 走 CONNECT,socks5 走拨号器。 func transportFor(p *ProxySpec) *http.Transport { addr := fmt.Sprintf("%s:%d", p.Host, p.Port) + dialer := &net.Dialer{Timeout: proxyDialTimeout} if p.Type == "http" || p.Type == "https" { u := &url.URL{Scheme: p.Type, Host: addr} if p.Username != "" { u.User = url.UserPassword(p.Username, p.Password) } - return &http.Transport{Proxy: http.ProxyURL(u)} + return &http.Transport{ + Proxy: http.ProxyURL(u), + DialContext: dialer.DialContext, + TLSHandshakeTimeout: proxyTLSHandshakeTimeout, + } } var auth *proxy.Auth if p.Username != "" { auth = &proxy.Auth{User: p.Username, Password: p.Password} } - d, err := proxy.SOCKS5("tcp", addr, auth, proxy.Direct) + d, err := proxy.SOCKS5("tcp", addr, auth, dialer) if err != nil { return nil } @@ -71,5 +84,8 @@ func transportFor(p *ProxySpec) *http.Transport { if !ok { return nil } - return &http.Transport{DialContext: cd.DialContext} + return &http.Transport{ + DialContext: cd.DialContext, + TLSHandshakeTimeout: proxyTLSHandshakeTimeout, + } } diff --git a/internal/oci/proxyhttp_test.go b/internal/oci/proxyhttp_test.go new file mode 100644 index 0000000..92b2396 --- /dev/null +++ b/internal/oci/proxyhttp_test.go @@ -0,0 +1,34 @@ +package oci + +import ( + "testing" +) + +func TestTransportForStageTimeouts(t *testing.T) { + tests := []struct { + name string + spec *ProxySpec + wantProxy bool // CONNECT 分支应设 Proxy 函数 + }{ + {name: "http CONNECT 代理", spec: &ProxySpec{Type: "http", Host: "127.0.0.1", Port: 8080}, wantProxy: true}, + {name: "https CONNECT 代理", spec: &ProxySpec{Type: "https", Host: "127.0.0.1", Port: 8443}, wantProxy: true}, + {name: "socks5 代理", spec: &ProxySpec{Type: "socks5", Host: "127.0.0.1", Port: 1080, Username: "u", Password: "p"}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + tr := transportFor(tt.spec) + if tr == nil { + t.Fatal("transportFor 返回 nil") + } + if (tr.Proxy != nil) != tt.wantProxy { + t.Fatalf("Proxy 函数存在性 = %v, 期望 %v", tr.Proxy != nil, tt.wantProxy) + } + if tr.DialContext == nil { + t.Fatal("应设置带超时的 DialContext") + } + if tr.TLSHandshakeTimeout != proxyTLSHandshakeTimeout { + t.Fatalf("TLSHandshakeTimeout = %v, 期望 %v", tr.TLSHandshakeTimeout, proxyTLSHandshakeTimeout) + } + }) + } +} diff --git a/internal/service/aigateway.go b/internal/service/aigateway.go index 4558cda..5f75955 100644 --- a/internal/service/aigateway.go +++ b/internal/service/aigateway.go @@ -68,6 +68,9 @@ type AiGatewayService struct { // grokWebSearch / grokXSearch 是 xai. 模型服务端搜索工具默认注入开关 grokWebSearch atomic.Bool grokXSearch atomic.Bool + // upstreamWaitSec 是 responses 直通的上游无响应预算(秒):非流式为单次尝试 + // 总超时,流式为等待响应头预算;multi-agent/搜索类模型远超 SDK 默认 60s + upstreamWaitSec atomic.Int64 } // NewAiGatewayService 组装依赖;调用 StartCleanup 后开始调用日志周期清理。 @@ -78,6 +81,7 @@ func NewAiGatewayService(db *gorm.DB, configs *OciConfigService, client oci.Clie s.streamGuardKB.Store(int64(loadIntSetting(db, settingAiStreamGuardKB, defaultStreamGuardKB))) s.grokWebSearch.Store(loadBoolSetting(db, settingAiGrokWebSearch, true)) s.grokXSearch.Store(loadBoolSetting(db, settingAiGrokXSearch, true)) + s.upstreamWaitSec.Store(int64(loadIntSetting(db, settingAiUpstreamWaitSec, defaultUpstreamWaitSec))) return s } @@ -93,11 +97,17 @@ const ( // 开关,缺省开。 settingAiGrokWebSearch = "ai_grok_web_search" settingAiGrokXSearch = "ai_grok_x_search" + // settingAiUpstreamWaitSec 是 responses 直通的上游无响应预算(秒)。 + settingAiUpstreamWaitSec = "ai_upstream_wait_seconds" ) // defaultStreamGuardKB 是保险丝阈值缺省值,低于实测断流边界留余量。 const defaultStreamGuardKB = 60 +// defaultUpstreamWaitSec 是上游无响应预算缺省值(秒):multi-agent 非流式 +// 实测 100~180s 才回响应头,给足余量;上下限见 SetUpstreamWait。 +const defaultUpstreamWaitSec = 300 + // loadBoolSetting 读 settings 表布尔键,无行或值非法时返回缺省。 func loadBoolSetting(db *gorm.DB, key string, def bool) bool { var row model.Setting @@ -165,6 +175,25 @@ func (s *AiGatewayService) SetStreamGuard(ctx context.Context, on bool, kb int) return nil } +// UpstreamWait 返回 responses 直通的上游无响应预算。 +func (s *AiGatewayService) UpstreamWait() time.Duration { + return time.Duration(s.upstreamWaitSec.Load()) * time.Second +} + +// SetUpstreamWait 持久化并即时生效上游无响应预算;sec 限定 30..900。 +func (s *AiGatewayService) SetUpstreamWait(ctx context.Context, sec int) error { + if sec < 30 || sec > 900 { + return fmt.Errorf("上游无响应预算须在 30..900 秒, 收到 %d", sec) + } + err := s.db.WithContext(ctx). + Save(&model.Setting{Key: settingAiUpstreamWaitSec, Value: strconv.Itoa(sec)}).Error + if err != nil { + return fmt.Errorf("保存上游无响应预算: %w", err) + } + s.upstreamWaitSec.Store(int64(sec)) + return nil +} + // GrokSearch 返回 grok 服务端搜索工具默认注入开关(web_search, x_search)。 func (s *AiGatewayService) GrokSearch() (bool, bool) { return s.grokWebSearch.Load(), s.grokXSearch.Load() diff --git a/internal/service/aigateway_chat.go b/internal/service/aigateway_chat.go index 8c24461..8384815 100644 --- a/internal/service/aigateway_chat.go +++ b/internal/service/aigateway_chat.go @@ -64,7 +64,7 @@ func (s *AiGatewayService) passthroughOnce(ctx context.Context, cand *aiCandidat if err != nil { return nil, err } - return s.client.GenAiCompatResponses(ctx, cred, cand.ch.Region, raw) + return s.client.GenAiCompatResponses(ctx, cred, cand.ch.Region, raw, s.UpstreamWait()) } // RespPassthroughStream 编排流式直通:流建立成功即绑定渠道,建立失败按 switchable @@ -83,7 +83,7 @@ func (s *AiGatewayService) RespPassthroughStream(ctx context.Context, raw []byte if err != nil { return nil, meta, err } - stream, err := s.client.GenAiCompatResponsesStream(ctx, cred, cand.ch.Region, raw) + stream, err := s.client.GenAiCompatResponsesStream(ctx, cred, cand.ch.Region, raw, s.UpstreamWait()) if err == nil { s.markSuccess(ctx, cand.ch.ID) return stream, meta, nil diff --git a/internal/service/aigateway_test.go b/internal/service/aigateway_test.go index 88e70ad..afea6b5 100644 --- a/internal/service/aigateway_test.go +++ b/internal/service/aigateway_test.go @@ -83,7 +83,7 @@ func (f *gatewayStubClient) GenAiApplyGuardrails(ctx context.Context, cred oci.C return f.guardOutcome, f.guardErr } -func (f *gatewayStubClient) GenAiCompatResponses(ctx context.Context, cred oci.Credentials, region string, body []byte) ([]byte, error) { +func (f *gatewayStubClient) GenAiCompatResponses(ctx context.Context, cred oci.Credentials, region string, body []byte, wait time.Duration) ([]byte, error) { f.passCalls++ f.passRegions = append(f.passRegions, region) if len(f.passErrs) > 0 { @@ -96,7 +96,7 @@ func (f *gatewayStubClient) GenAiCompatResponses(ctx context.Context, cred oci.C return f.passPayload, nil } -func (f *gatewayStubClient) GenAiCompatResponsesStream(ctx context.Context, cred oci.Credentials, region string, body []byte) (io.ReadCloser, error) { +func (f *gatewayStubClient) GenAiCompatResponsesStream(ctx context.Context, cred oci.Credentials, region string, body []byte, wait time.Duration) (io.ReadCloser, error) { f.passCalls++ f.passRegions = append(f.passRegions, region) if len(f.passErrs) > 0 { @@ -1124,3 +1124,38 @@ func TestAggregatedModelsFilterDeprecated(t *testing.T) { t.Fatalf("开关开:弃用模型应被过滤, got %+v", items) } } + +func TestUpstreamWaitSetting(t *testing.T) { + gw, _ := newTestGateway(t, &gatewayStubClient{fakeClient: &fakeClient{}}) + ctx := context.Background() + + if got := gw.UpstreamWait(); got != 300*time.Second { + t.Fatalf("缺省上游无响应预算 = %v, 期望 300s", got) + } + tests := []struct { + name string + sec int + wantErr bool + }{ + {name: "下界 30 有效", sec: 30}, + {name: "上界 900 有效", sec: 900}, + {name: "低于下界拒绝", sec: 29, wantErr: true}, + {name: "高于上界拒绝", sec: 901, wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := gw.SetUpstreamWait(ctx, tt.sec) + if (err != nil) != tt.wantErr { + t.Fatalf("SetUpstreamWait(%d) = %v, wantErr=%v", tt.sec, err, tt.wantErr) + } + if !tt.wantErr && gw.UpstreamWait() != time.Duration(tt.sec)*time.Second { + t.Fatalf("UpstreamWait = %v, 期望 %ds", gw.UpstreamWait(), tt.sec) + } + }) + } + // 持久化后新实例应加载已存值(最后一次成功设置为 900) + gw2 := NewAiGatewayService(gw.db, nil, nil) + if got := gw2.UpstreamWait(); got != 900*time.Second { + t.Fatalf("重建服务加载预算 = %v, 期望 900s", got) + } +}