mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-11-12 12:17:23 +00:00
fix(codearts): 上游限流改为可声明重试(内核通用能力)+ 会话/连接两处真因
用户连续报「codeart 还是不可用 / empty stream」。把上游原始响应打出来逐条对齐,
三个症状同源,且都不是"模型坏了":
| 场景 | 上游实际返回 |
|---|---|
| 流式、短时间连续请求 | HTTP **200** + 只带 error 的帧 ⇒ 内核判 `empty stream` |
| 非流式、同一条件 | HTTP 400 `TM.00001041: 并发会话数已达上限(3个)` |
| 突发请求 | HTTP 429 `TPM limit … 500000 tokens per minute` |
实测:最后一个请求结束后约 **44~60 秒**释放;同一账号连续请求 3~4 次即占满。
参考实现(示例插件 llm-adapter.ts)的做法是**每 10s 轮询后重试 chat**(最多 30
分钟),所以用户那边"看起来没事"。
## 我们的改法(内核只做通用机制,厂商知识留在插件)
1. **内核新增声明式重试策略** `source.retry`(纯数据:状态码列表 + 响应体子串 +
间隔 + 次数)。内核不认识任何厂商;命中即在同一网关请求内每 10s 重试一次。
流式只在**握手阶段**重试(首帧尚未下发,不会重复输出),非流式整体重试。
2. **插件为 codearts 声明**:400/429 + 命中 `TM.00001041` / `TPM limit` /
`并发会话数已达上限` → 10s × 最多 18 次(≈3 分钟)。
3. **适配器补两个会话细节**:`Connection: close`(keep-alive 空闲连接会被上游
计入并发会话)与稳定 `Session-Id`(同一账号恒定,避免每次请求新建会话)。
4. 错误文案补可行动提示(并发会话上限 / TPM 限流,约 1 分钟释放)。
## 验证(生产)
- 连续 6 次**流式**:6/6 成功(撞上限时自动等 10s 再试,客户端只感觉慢):
`+0s / +12s / +34s / +86s / +100s / +144s`
- 连续 5 次**非流式**:5/5 成功(`+0 / +11 / +20 / +28 / +38s`)
- 修复前:第 4 个请求起必然 TM.00001041 / empty stream。
> 注:`TM.00001005 无正常可用套餐` 属于账号权益问题(不重试、如实报出),
> 与限流要分开处理。
(cherry picked from commit 6454b094bc)
This commit is contained in:
@ -1156,7 +1156,7 @@ func (p *Provider) Chat(ctx context.Context, req *types.ChatRequest) (*types.Uni
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
raw, status, err := p.do(ctx, chatURL, body, hdrs)
|
||||
raw, status, err := p.doWithRetry(ctx, chatURL, body, hdrs)
|
||||
if err != nil {
|
||||
// a client disconnect or cancelled context is neither a success nor
|
||||
// a failure for scheduling purposes — only upstream errors count
|
||||
@ -1227,8 +1227,48 @@ func (p *Provider) ChatStream(ctx context.Context, req *types.ChatRequest) (<-ch
|
||||
}
|
||||
rc := make(chan respOrErr, 1)
|
||||
go func() {
|
||||
resp, err := p.doRawStream(ctx, streamURL, body, hdrs)
|
||||
rc <- respOrErr{resp, err}
|
||||
// Retry only the HANDSHAKE: a policy-matched rejection (e.g. a
|
||||
// per-account concurrency cap answered with a business 4xx) can clear
|
||||
// within seconds, and nothing has been sent to the client yet — the
|
||||
// kernel holds the first chunk back. Once a 200 arrives and streaming
|
||||
// starts, no retry happens (that would duplicate output).
|
||||
rp := p.cfg.Retry
|
||||
attempts := 1
|
||||
if rp != nil && rp.MaxAttempts > 1 {
|
||||
attempts = rp.MaxAttempts
|
||||
}
|
||||
for i := 0; i < attempts; i++ {
|
||||
resp, err := p.doRawStream(ctx, streamURL, body, hdrs)
|
||||
if err != nil {
|
||||
rc <- respOrErr{resp, err}
|
||||
return
|
||||
}
|
||||
if resp.StatusCode == 200 {
|
||||
rc <- respOrErr{resp, nil}
|
||||
return
|
||||
}
|
||||
// A rejection can only be classified after the body is read: the
|
||||
// policy matches on body substrings, and a source may use the same
|
||||
// status (400) for unrelated errors that must NOT be retried.
|
||||
raw, _ := io.ReadAll(resp.Body)
|
||||
resp.Body.Close()
|
||||
if !rp.ShouldRetry(resp.StatusCode, string(raw)) {
|
||||
// Not the retryable body: hand the rejection back verbatim so
|
||||
// the caller reports the real upstream error.
|
||||
resp2 := &http.Response{StatusCode: resp.StatusCode, Body: io.NopCloser(strings.NewReader(string(raw)))}
|
||||
rc <- respOrErr{resp2, nil}
|
||||
return
|
||||
}
|
||||
if i == attempts-1 {
|
||||
resp2 := &http.Response{StatusCode: resp.StatusCode, Body: io.NopCloser(strings.NewReader(string(raw)))}
|
||||
rc <- respOrErr{resp2, nil}
|
||||
return
|
||||
}
|
||||
if serr := sleepCtx(ctx, rp.DelayDuration()); serr != nil {
|
||||
rc <- respOrErr{nil, serr}
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
inner := make(chan types.UnifiedChunk, 64)
|
||||
@ -1560,6 +1600,50 @@ func (p *Provider) do(ctx context.Context, url, body string, hdr http.Header) (s
|
||||
return string(raw), resp.StatusCode, nil
|
||||
}
|
||||
|
||||
// doWithRetry runs one non-stream request, retrying per the source's
|
||||
// declarative RetryPolicy. The policy is pure data (status list + body
|
||||
// substrings), so the kernel stays vendor-agnostic: a provider that answers a
|
||||
// per-account concurrency cap with a business 4xx can clear by itself while the
|
||||
// client just waits a few seconds instead of seeing an error.
|
||||
func (p *Provider) doWithRetry(ctx context.Context, url, body string, hdr http.Header) (string, int, error) {
|
||||
rp := p.cfg.Retry
|
||||
attempts := 1
|
||||
if rp != nil && rp.MaxAttempts > 1 {
|
||||
attempts = rp.MaxAttempts
|
||||
}
|
||||
var raw string
|
||||
var status int
|
||||
for i := 0; i < attempts; i++ {
|
||||
r, st, err := p.do(ctx, url, body, hdr)
|
||||
if err != nil {
|
||||
return "", 0, err
|
||||
}
|
||||
raw, status = r, st
|
||||
if status == 200 || !rp.ShouldRetry(status, raw) || i == attempts-1 {
|
||||
break
|
||||
}
|
||||
if serr := sleepCtx(ctx, rp.DelayDuration()); serr != nil {
|
||||
return "", 0, serr
|
||||
}
|
||||
}
|
||||
return raw, status, nil
|
||||
}
|
||||
|
||||
// sleepCtx sleeps for d unless the request context ends first.
|
||||
func sleepCtx(ctx context.Context, d time.Duration) error {
|
||||
if d <= 0 {
|
||||
return ctx.Err()
|
||||
}
|
||||
t := time.NewTimer(d)
|
||||
defer t.Stop()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-t.C:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
func (p *Provider) doRaw(ctx context.Context, url, body string, hdr http.Header) (*http.Response, error) {
|
||||
httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader([]byte(body)))
|
||||
if err != nil {
|
||||
|
||||
Reference in New Issue
Block a user