diff --git a/.gitignore b/.gitignore index 61069b7..b1c7a03 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,3 @@ -/.idea/* \ No newline at end of file +/.idea/* +/.superpowers/ +/docs/superpowers/ diff --git a/config.yml b/config.yml index 2188df4..eaa61c7 100644 --- a/config.yml +++ b/config.yml @@ -10,7 +10,7 @@ database: host: "192.168.0.83" port: "15432" user: "postgres" - pass: "Bjang09@686^*^" + pass: "Q!P@z#M$1@686^*^.." name: "model-gateway" prefix: "" # (可选)表名前缀 role: "master" # (可选)数据库主从角色(master/slave)，默认为master。如果不使用应用主从机制请不配置或留空即可。 @@ -28,10 +28,10 @@ database: timeMaintainDisabled: false # (可选)是否完全关闭时间更新特性，为true时CreatedAt/UpdatedAt/DeletedAt都将失效 model_gateway: - type: "pgsql" - host: "localhost" - port: "5432" + host: "192.168.0.83" + port: "15432" user: "postgres" - pass: "123456" + pass: "Q!P@z#M$1@686^*^.." name: "model-gateway" prefix: "model_gateway_" role: "master" @@ -50,17 +50,17 @@ database: redis: default: - address: localhost:6379 + address: 192.168.0.83:6379 db: 0 consul: - address: localhost:8500 + address: 192.168.0.83:8500 jaeger: - addr: localhost:4318 + addr: 192.168.0.83:4318 nats: - addr: localhost + addr: 192.168.0.83 port: 4222 # schema_mapping 自动构建专用 LLM（OpenAI 兼容；密钥不要写死进代码） diff --git a/consts/public/table_name.go b/consts/public/table_name.go index 7980793..8b7dbdc 100644 --- a/consts/public/table_name.go +++ b/consts/public/table_name.go @@ -5,9 +5,9 @@ const ( ) const ( - TableNameModelManage = "model_manage" - TableNameModelSession = "model_session" - TableNameModelTaskStart = "model_task_start" - TableNameModelTaskEnd = "model_task_end" - TableNameErrorMemory = "error_memory" + TableNameModelManage = "model_manage" + TableNameModelSession = "model_session" + TableNameModelTaskStart = "model_task_start" + TableNameModelTaskEnd = "model_task_end" + TableNameModelErrorMemory = "model_error_memory" ) diff --git a/controller/error_memory_controller.go b/controller/error_memory_controller.go deleted file mode 100644 index a6f077e..0000000 --- a/controller/error_memory_controller.go +++ /dev/null @@ -1,25 +0,0 @@ -package controller - -import ( - "context" - "model-gateway/model/dto" - "model-gateway/service" - - "gitea.redpowerfuture.com/red-future/common/beans" -) - -// ErrorMemory 错误重试记忆控制器 -var ErrorMemory = new(errorMemory) - -type errorMemory struct{} - -// List 错误重试记忆列表 -func (c *errorMemory) List(ctx context.Context, req *dto.GetErrorMemoryListReq) (res *dto.GetErrorMemoryListRes, err error) { - return service.ErrorMemory.List(ctx, req) -} - -// Delete 删除错误重试记忆 -func (c *errorMemory) Delete(ctx context.Context, req *dto.DeleteErrorMemoryReq) (res *beans.ResponseEmpty, err error) { - err = service.ErrorMemory.Delete(ctx, req) - return -} diff --git a/controller/model_error_memory_controller.go b/controller/model_error_memory_controller.go new file mode 100644 index 0000000..40302d9 --- /dev/null +++ b/controller/model_error_memory_controller.go @@ -0,0 +1,25 @@ +package controller + +import ( + "context" + "model-gateway/model/dto" + "model-gateway/service" + + "gitea.redpowerfuture.com/red-future/common/beans" +) + +// ModelErrorMemory 错误重试记忆控制器 +var ModelErrorMemory = new(modelErrorMemory) + +type modelErrorMemory struct{} + +// List 错误重试记忆列表 +func (c *modelErrorMemory) List(ctx context.Context, req *dto.GetErrorMemoryListReq) (res *dto.GetErrorMemoryListRes, err error) { + return service.ModelErrorMemory.List(ctx, req) +} + +// Delete 删除错误重试记忆 +func (c *modelErrorMemory) Delete(ctx context.Context, req *dto.DeleteErrorMemoryReq) (res *beans.ResponseEmpty, err error) { + err = service.ModelErrorMemory.Delete(ctx, req) + return +} diff --git a/dao/error_memory_dao_test.go b/dao/error_memory_dao_test.go deleted file mode 100644 index c1c7c6c..0000000 --- a/dao/error_memory_dao_test.go +++ /dev/null @@ -1,41 +0,0 @@ -package dao - -import ( - "context" - "fmt" - "testing" - "time" - - "model-gateway/consts/public" - - _ "gitea.redpowerfuture.com/red-future/common/consul" - "gitea.redpowerfuture.com/red-future/common/db/gfdb" - _ "github.com/gogf/gf/contrib/drivers/pgsql/v2" - "github.com/gogf/gf/v2/net/gtrace" -) - -// TestErrorMemoryDaoGetByKeyMiss 回归：错误记忆未命中必须返回 (nil, nil)， -// 而非把空记录丢给 r.Struct(&res) 上浮 sql.ErrNoRows —— 否则 service.shouldRetryWithMemory -// 会把一切查询错误当 fail-closed，导致 分析→落库→重试 整条链路变死代码。 -// 依赖真实 PG；不可达时跳过（不在无 PG 环境硬失败）。 -func TestErrorMemoryDaoGetByKeyMiss(t *testing.T) { - ctx, span := gtrace.NewSpan(context.Background(), "TestErrorMemoryDaoGetByKeyMiss") - defer span.End() - - // 探测连接：不可用则跳过 - if _, err := gfdb.DB(ctx, public.DbNameModelGateway). - Model(ctx, public.TableNameErrorMemory). - NoTenantId(ctx). - Count(); err != nil { - t.Skipf("DB不可用,跳过: %v", err) - } - - key := fmt.Sprintf("test-miss-%d", time.Now().UnixNano()) - row, err := ErrorMemory.GetByKey(ctx, key) - if err != nil { - t.Fatalf("GetByKey 未命中不应报错: %v", err) - } - if row != nil { - t.Fatalf("GetByKey 未命中应返回 nil, got %+v", row) - } -} diff --git a/dao/error_memory_dao.go b/dao/model_error_memory_dao.go similarity index 52% rename from dao/error_memory_dao.go rename to dao/model_error_memory_dao.go index c7a7b1f..58e2cd5 100644 --- a/dao/error_memory_dao.go +++ b/dao/model_error_memory_dao.go @@ -8,17 +8,17 @@ import ( "gitea.redpowerfuture.com/red-future/common/db/gfdb" ) -var ErrorMemory = &errorMemoryDao{} +var ModelErrorMemory = &modelErrorMemoryDao{} -type errorMemoryDao struct{} +type modelErrorMemoryDao struct{} // GetByKey 按记忆键查询（未命中返回 (nil, nil)） // 错误记忆为全局表：NoTenantId 绕过租户过滤，跨租户共享；r.IsEmpty() 兜底 miss 契约， // 避免对空记录 r.Struct(&res) 上浮 sql.ErrNoRows 导致调用方 fail-closed。 -func (d *errorMemoryDao) GetByKey(ctx context.Context, key string) (res *entity.ErrorMemory, err error) { - r, err := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameErrorMemory). +func (d *modelErrorMemoryDao) GetByKey(ctx context.Context, key string) (res *entity.ModelErrorMemory, err error) { + r, err := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameModelErrorMemory). NoTenantId(ctx). - Where(entity.ErrorMemoryCol.MemoryKey, key). + Where(entity.ModelErrorMemoryCol.MemoryKey, key). One() if err != nil { return @@ -31,18 +31,18 @@ func (d *errorMemoryDao) GetByKey(ctx context.Context, key string) (res *entity. } // Upsert 存在则更新 retryable/reason/analyzed_by，不存在则插入 -func (d *errorMemoryDao) Upsert(ctx context.Context, m *entity.ErrorMemory) (err error) { - model := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameErrorMemory) +func (d *modelErrorMemoryDao) Upsert(ctx context.Context, m *entity.ModelErrorMemory) (err error) { + model := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameModelErrorMemory) // Count 同样全局化：跨租户已存在的记忆键需命中更新分支，而非重复插入 - n, err := model.NoTenantId(ctx).Where(entity.ErrorMemoryCol.MemoryKey, m.MemoryKey).Count() + n, err := model.NoTenantId(ctx).Where(entity.ModelErrorMemoryCol.MemoryKey, m.MemoryKey).Count() if err != nil { return } if n > 0 { - _, err = model.Where(entity.ErrorMemoryCol.MemoryKey, m.MemoryKey).Data(map[string]any{ - entity.ErrorMemoryCol.Retryable: m.Retryable, - entity.ErrorMemoryCol.Reason: m.Reason, - entity.ErrorMemoryCol.AnalyzedBy: m.AnalyzedBy, + _, err = model.Where(entity.ModelErrorMemoryCol.MemoryKey, m.MemoryKey).Data(map[string]any{ + entity.ModelErrorMemoryCol.Retryable: m.Retryable, + entity.ModelErrorMemoryCol.Reason: m.Reason, + entity.ModelErrorMemoryCol.AnalyzedBy: m.AnalyzedBy, }).Update() return } @@ -51,20 +51,20 @@ func (d *errorMemoryDao) Upsert(ctx context.Context, m *entity.ErrorMemory) (err } // List 分页查询（按 id 倒序）；全局表，管理端列表展示所有租户记忆 -func (d *errorMemoryDao) List(ctx context.Context, page, pageSize int) (list []entity.ErrorMemory, total int64, err error) { - model := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameErrorMemory).NoTenantId(ctx) +func (d *modelErrorMemoryDao) List(ctx context.Context, page, pageSize int) (list []entity.ModelErrorMemory, total int64, err error) { + model := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameModelErrorMemory).NoTenantId(ctx) n, err := model.Count() if err != nil { return } total = int64(n) - err = model.Page(page, pageSize).OrderDesc(entity.ErrorMemoryCol.Id).Scan(&list) + err = model.Page(page, pageSize).OrderDesc(entity.ModelErrorMemoryCol.Id).Scan(&list) return } // Delete 按 id 删除（软删除） -func (d *errorMemoryDao) Delete(ctx context.Context, id int64) (err error) { - _, err = gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameErrorMemory). - Where(entity.ErrorMemoryCol.Id, id).Delete() +func (d *modelErrorMemoryDao) Delete(ctx context.Context, id int64) (err error) { + _, err = gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameModelErrorMemory). + Where(entity.ModelErrorMemoryCol.Id, id).Delete() return } diff --git a/docs/superpowers/plans/2026-09-01-llm-error-retry-memory.md b/docs/superpowers/plans/2026-09-01-llm-error-retry-memory.md deleted file mode 100644 index 1290283..0000000 --- a/docs/superpowers/plans/2026-09-01-llm-error-retry-memory.md +++ /dev/null @@ -1,1178 +0,0 @@ -# 模型错误 LLM 分析重试 + 持久记忆 实现计划 - -> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. - -**Goal:** 请求上游模型失败时,不再用固定错误码清单判定重试,改为调用对话模型分析错误类型决定是否重试,并把"错误→结论"持久化到 PostgreSQL 记忆表,命中记忆不再调分析模型。 - -**Architecture:** 新增统一的 `shouldRetryWithMemory(ctx, modelInfo, code, msg, rawBody) bool` 判定入口,内部:构造记忆键 → 查 `model_gateway_error_memory` → 命中返回存储结论;未命中经 singleflight 合并后调分析模型(对话模型)解析 `{retryable,reason}` → UPSERT 落库 → 返回。同步/流式(缓冲)/异步任务启动三路接入,保留现有 `modelCallMaxRetries` 退避预算。分析模型不新增配置段:失败模型本身 `chat_model=true` 优先复用,否则取当前用户对话模型;任一环节失败均 fail-closed 不重试。 - -**Tech Stack:** Go 1.26+, GoFrame v2(gfdb/Model 链), PostgreSQL(gfdb), `golang.org/x/sync/singleflight`, OpenAI 兼容 chat 接口。 - -## Global Constraints - -- **Spec:** `docs/superpowers/specs/2026-09-01-llm-error-retry-memory-design.md`(已提交 `49c1cb1`)。 -- **⚠️ WIP 前置:** model-gateway 工作区有**未提交的 staged WIP 重构**(涉及 `service/session_sync.go`、`service/session_stream.go`、`service/model_task_end_service.go`、`service/pricing_client.go` 等)。执行前**必须**先与用户确认 WIP 去留(建议先提交 WIP)。本计划每个 task 的提交统一用 `git commit --only <本task文件>` 只提交本 task 文件,避免夹带;但 Task 7 改到 `session_sync.go`/`session_stream.go` 时,工作区版本同时含 WIP+本任务改动,`--only` 会一并提交 —— **该两步前先让 WIP 落库**。 -- **分析调用不复用 `httpclient.ModelHttpNormalRequest`**(其 `modelDoRaw` 响应头超时 30min 且不区分 HTTP 状态码)—— 用独立 `http.Client{Timeout: 15s}`(对齐 `schema_mapping_service.callLLM` 的写法)。这是对 spec §3 的一处**技术修正**,spec 已同步更新。 -- 固定错误码清单 `isRetryableErrorCode` **删除**;`modelCallMaxRetries=10`、`retryWait` 指数退避保留。 -- 记忆条目**永久有效**(无 TTL),靠管理端点手动清理;不做自动过期/连续失败降级(YAGNI)。 -- 表名/列名以 entity orm 为准;`model_gateway_` 前缀;列常量进 `consts/public/table_name.go`、entity 常量进 `model/entity/*`. -- 分析消息截断 2000 字符;`temperature=0`;`max_tokens=256`。 -- 测试:纯函数(归一化/记忆键/解析/请求体/单飞去重)写真实单测;DAO/三路接入以 `go build ./...` + 手动验证。 - ---- - -### Task 1: 错误重试记忆表 + entity + consts - -**Files:** -- Modify: `update.sql`(追加建表 DDL) -- Create: `model/entity/error_memory.go` -- Modify: `consts/public/table_name.go` - -**Interfaces:** -- Consumes: `beans.SQLBaseDO` / `beans.SQLBaseCol`(common/beans,基列 `Id/TenantId/Creator/CreatedAt/Updater/UpdatedAt/DeletedAt`) -- Produces: `entity.ErrorMemory` + `entity.ErrorMemoryCol`;`public.TableNameErrorMemory = "error_memory"` - -- [ ] **Step 1: 追加建表 DDL 到 `update.sql` 末尾** - -```sql --- ========================= --- 错误重试记忆：LLM 分析上游模型错误是否可重试的持久知识库 --- memory_key = SHA-256(upstream|error_code|归一化消息)，唯一；命中直接复用，永久有效 --- ========================= -CREATE TABLE IF NOT EXISTS model_gateway_error_memory ( - id BIGSERIAL PRIMARY KEY, - tenant_id BIGINT DEFAULT 0, - creator VARCHAR(64) DEFAULT '', - created_at TIMESTAMPTZ DEFAULT now(), - updater VARCHAR(64) DEFAULT '', - updated_at TIMESTAMPTZ DEFAULT now(), - deleted_at TIMESTAMPTZ DEFAULT NULL, - memory_key CHAR(64) NOT NULL, - upstream VARCHAR(512) NOT NULL DEFAULT '', - error_code VARCHAR(128) NOT NULL DEFAULT '', - msg_fingerprint CHAR(32) NOT NULL DEFAULT '', - retryable BOOLEAN NOT NULL DEFAULT false, - reason VARCHAR(512) NOT NULL DEFAULT '', - analyzed_by VARCHAR(128) NOT NULL DEFAULT '' -); -CREATE UNIQUE INDEX IF NOT EXISTS uk_error_memory_memory_key - ON model_gateway_error_memory (memory_key) - WHERE deleted_at IS NULL; -``` - -- [ ] **Step 2: 新建 `model/entity/error_memory.go`** - -```go -package entity - -import "gitea.redpowerfuture.com/red-future/common/beans" - -type errorMemoryCol struct { - beans.SQLBaseCol - MemoryKey string - Upstream string - ErrorCode string - MsgFingerprint string - Retryable string - Reason string - AnalyzedBy string -} - -var ErrorMemoryCol = errorMemoryCol{ - SQLBaseCol: beans.DefSQLBaseCol, - MemoryKey: "memory_key", - Upstream: "upstream", - ErrorCode: "error_code", - MsgFingerprint: "msg_fingerprint", - Retryable: "retryable", - Reason: "reason", - AnalyzedBy: "analyzed_by", -} - -// ErrorMemory 错误重试记忆（LLM 分析结论持久化，永久有效） -type ErrorMemory struct { - beans.SQLBaseDO `orm:",inline"` - MemoryKey string `orm:"memory_key" json:"memoryKey" dc:"记忆键=SHA-256(upstream|code|归一化消息)"` - Upstream string `orm:"upstream" json:"upstream" dc:"失败上游BaseURL"` - ErrorCode string `orm:"error_code" json:"errorCode" dc:"错误码"` - MsgFingerprint string `orm:"msg_fingerprint" json:"msgFingerprint" dc:"归一化消息md5"` - Retryable bool `orm:"retryable" json:"retryable" dc:"是否可重试"` - Reason string `orm:"reason" json:"reason" dc:"分析原因"` - AnalyzedBy string `orm:"analyzed_by" json:"analyzedBy" dc:"分析模型名"` -} -``` - -- [ ] **Step 3: `consts/public/table_name.go` 增加表名常量** - -在 `TableNameModelTaskEnd` 后加一行: - -```go -TableNameErrorMemory = "error_memory" -``` - -- [ ] **Step 4: 编译验证** - -Run: `cd model-gateway && go build ./...` -Expected: PASS(无编译错误) - -- [ ] **Step 5: 提交** - -```bash -cd model-gateway -git add update.sql model/entity/error_memory.go consts/public/table_name.go -git commit --only update.sql model/entity/error_memory.go consts/public/table_name.go -m "feat: 新增错误重试记忆表/entity/consts" -``` - ---- - -### Task 2: 错误消息归一化 + 记忆键构造(纯函数 + 单测) - -**Files:** -- Create: `service/error_memory.go` -- Create: `service/error_memory_test.go` - -**Interfaces:** -- Consumes: 无 -- Produces: `normalizeErrorMsg(msg string) string`、`buildMemoryKey(upstream, code, msg string) string`、`msgFingerprint(msg string) string`(Task 6 消费) - -- [ ] **Step 1: 写失败测试 `service/error_memory_test.go`** - -```go -package service - -import "testing" - -func TestNormalizeErrorMsg(t *testing.T) { - cases := []struct{ in, want string }{ - {"rate limit exceeded for req-abc123", "rate limit exceeded for {reqid}"}, - {"timeout at 2026-09-01T09:00:00Z req_88f1a2", "timeout at {time} {reqid}"}, - {"uuid 0f8fad5b-d9cb-469f-a165-70867728950e remains", "uuid {uuid} remains"}, - {"err code 12345678 quota exceeded", "err code {num} quota exceeded"}, - {"clean message unchanged", "clean message unchanged"}, - } - for _, c := range cases { - if got := normalizeErrorMsg(c.in); got != c.want { - t.Fatalf("normalizeErrorMsg(%q) = %q, want %q", c.in, got, c.want) - } - } -} - -func TestBuildMemoryKey(t *testing.T) { - k1 := buildMemoryKey("https://a.com", "429", "rate limit for req-a1b2c3") - k2 := buildMemoryKey("https://a.com", "429", "rate limit for req-d4e5f6") // 同因不同请求ID - if k1 != k2 { - t.Fatalf("同因不同请求ID应同键: %s != %s", k1, k2) - } - k3 := buildMemoryKey("https://b.com", "429", "rate limit for req-a1b2c3") // 不同上游 - if k1 == k3 { - t.Fatalf("不同上游应不同键") - } - k4 := buildMemoryKey("https://a.com", "500", "rate limit for req-a1b2c3") // 不同错误码 - if k1 == k4 { - t.Fatalf("不同错误码应不同键") - } - if len(k1) != 64 { - t.Fatalf("memory_key 应为 SHA-256 十六进制64位, got %d", len(k1)) - } -} - -func TestMsgFingerprint(t *testing.T) { - if msgFingerprint("a req-a1b2c3") != msgFingerprint("a req-d4e5f6") { - t.Fatalf("同因指纹应一致") - } -} -``` - -- [ ] **Step 2: 运行测试确认失败** - -Run: `cd model-gateway && go test ./service/ -run 'TestNormalizeErrorMsg|TestBuildMemoryKey|TestMsgFingerprint' -v` -Expected: FAIL(函数未定义) - -- [ ] **Step 3: 实现 `service/error_memory.go`** - -```go -package service - -import ( - "crypto/md5" - "crypto/sha256" - "encoding/hex" - "regexp" - "strings" -) - -var ( - reUUID = regexp.MustCompile(`[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}`) - reISOTime = regexp.MustCompile(`\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(\.\d+)?(Z|[+-]\d{2}:?\d{2})?`) - reUnixMs = regexp.MustCompile(`\b1[4-9]\d{12}\b`) // unix 毫秒级时间戳 - reRequestID = regexp.MustCompile(`\b(req[-_]?|request[-_]?|rid[-_:]?)[-_:]?[0-9a-zA-Z-]{4,}\b`) - reLongNum = regexp.MustCompile(`\b\d{4,}\b`) // 连续≥4位数字 -) - -// normalizeErrorMsg 归一化错误消息：剔除易变片段（UUID/时间戳/请求ID/连续数字）， -// 使同因不同实例的错误命中同一记忆键。 -func normalizeErrorMsg(msg string) string { - m := msg - m = reUUID.ReplaceAllString(m, "{uuid}") - m = reISOTime.ReplaceAllString(m, "{time}") - m = reUnixMs.ReplaceAllString(m, "{ts}") - m = reRequestID.ReplaceAllString(m, "{reqid}") - m = reLongNum.ReplaceAllString(m, "{num}") - return strings.TrimSpace(m) -} - -// buildMemoryKey 构造记忆键 = SHA-256(upstream|error_code|归一化消息)。 -// 含失败上游维度：不同上游的同类错误互不串扰。 -func buildMemoryKey(upstream, code, msg string) string { - raw := strings.Join([]string{upstream, code, normalizeErrorMsg(msg)}, "|") - sum := sha256.Sum256([]byte(raw)) - return hex.EncodeToString(sum[:]) -} - -// msgFingerprint 归一化消息的 md5（观测/展示用）。 -func msgFingerprint(msg string) string { - sum := md5.Sum([]byte(normalizeErrorMsg(msg))) - return hex.EncodeToString(sum[:]) -} -``` - -- [ ] **Step 4: 运行测试确认通过** - -Run: `cd model-gateway && go test ./service/ -run 'TestNormalizeErrorMsg|TestBuildMemoryKey|TestMsgFingerprint' -v` -Expected: PASS - -- [ ] **Step 5: 提交** - -```bash -cd model-gateway -git commit --only service/error_memory.go service/error_memory_test.go -m "feat: 错误消息归一化与记忆键构造" -``` - ---- - -### Task 3: 分析响应 JSON 解析(纯函数 + 单测) - -**Files:** -- Create: `service/error_analysis.go`(本 task 只含 `parseAnalysisResponse`;Task 5 同文件追加) -- Create: `service/error_analysis_test.go` - -**Interfaces:** -- Consumes: 无 -- Produces: `parseAnalysisResponse(content string) (retryable bool, reason string, err error)`(Task 5 消费) - -- [ ] **Step 1: 写失败测试 `service/error_analysis_test.go`** - -```go -package service - -import "testing" - -func TestParseAnalysisResponse(t *testing.T) { - cases := []struct { - name string - in string - wantRetry bool - wantErr bool - }{ - {"纯JSON", `{"retryable": true, "reason": "限流"}`, true, false}, - {"带json代码块", "```json\n{\"retryable\": false, \"reason\": \"参数错误\"}\n```", false, false}, - {"带前后文字", `分析结果: {"retryable": true, "reason": "瞬时故障"} 完毕`, true, false}, - {"非法响应", `抱歉，我无法分析`, false, true}, - {"空串", ``, false, true}, - } - for _, c := range cases { - retry, _, err := parseAnalysisResponse(c.in) - if c.wantErr { - if err == nil { - t.Fatalf("[%s] 期望错误, got retry=%v", c.name, retry) - } - continue - } - if err != nil { - t.Fatalf("[%s] 不应报错: %v", c.name, err) - } - if retry != c.wantRetry { - t.Fatalf("[%s] retryable = %v, want %v", c.name, retry, c.wantRetry) - } - } -} - -func TestParseAnalysisResponseReason(t *testing.T) { - _, reason, err := parseAnalysisResponse(`{"retryable": true, "reason": "服务过载"}`) - if err != nil { - t.Fatalf("unexpected err: %v", err) - } - if reason != "服务过载" { - t.Fatalf("reason = %q, want 服务过载", reason) - } -} -``` - -- [ ] **Step 2: 运行测试确认失败** - -Run: `cd model-gateway && go test ./service/ -run TestParseAnalysisResponse -v` -Expected: FAIL(函数未定义) - -- [ ] **Step 3: 实现 `service/error_analysis.go`(本 task 部分)** - -```go -package service - -import ( - "encoding/json" - "fmt" - "strings" -) - -// parseAnalysisResponse 解析分析模型输出的判定 JSON。 -// 容错：剥 ```json 代码块/首尾空白/多余文字，取首个 {...}。 -func parseAnalysisResponse(content string) (retryable bool, reason string, err error) { - s := strings.TrimSpace(content) - s = strings.TrimPrefix(s, "```json") - s = strings.TrimSuffix(s, "```") - s = strings.TrimSpace(s) - start, end := strings.IndexByte(s, '{'), strings.LastIndexByte(s, '}') - if start < 0 || end <= start { - return false, "", fmt.Errorf("分析响应中未找到JSON对象: %q", content) - } - var obj struct { - Retryable bool `json:"retryable"` - Reason string `json:"reason"` - } - if err = json.Unmarshal([]byte(s[start:end+1]), &obj); err != nil { - return false, "", fmt.Errorf("分析响应JSON解析失败: %v", err) - } - return obj.Retryable, obj.Reason, nil -} -``` - -- [ ] **Step 4: 运行测试确认通过** - -Run: `cd model-gateway && go test ./service/ -run TestParseAnalysisResponse -v` -Expected: PASS - -- [ ] **Step 5: 提交** - -```bash -cd model-gateway -git commit --only service/error_analysis.go service/error_analysis_test.go -m "feat: 分析响应JSON解析" -``` - ---- - -### Task 4: 错误重试记忆 DAO + DTO - -**Files:** -- Create: `dao/error_memory_dao.go` -- Create: `model/dto/error_memory_dao_dto.go` - -**Interfaces:** -- Consumes: `entity.ErrorMemory`/`entity.ErrorMemoryCol`(Task 1)、`public.TableNameErrorMemory`(Task 1) -- Produces: `dao.ErrorMemory.GetByKey(ctx, key) (*entity.ErrorMemory, error)`、`Upsert(ctx, *entity.ErrorMemory) error`、`List(ctx, page, pageSize) ([]entity.ErrorMemory, int64, error)`、`Delete(ctx, id) error`;DTO `GetErrorMemoryListReq/Res`、`DeleteErrorMemoryReq`、`ErrorMemoryItem` - -- [ ] **Step 1: 新建 `model/dto/error_memory_dao_dto.go`** - -```go -package dto - -import ( - "gitea.redpowerfuture.com/red-future/common/beans" - "github.com/gogf/gf/v2/frame/g" -) - -// GetErrorMemoryListReq 错误重试记忆列表 -type GetErrorMemoryListReq struct { - g.Meta `path:"/errorMemory/list" method:"get" tags:"错误记忆" summary:"错误重试记忆列表" dc:"查看错误→可重试结论记忆"` - *beans.Page `json:"page"` -} - -type GetErrorMemoryListRes struct { - List []ErrorMemoryItem `json:"list" dc:"记忆条目"` - Total int64 `json:"total" dc:"总数"` -} - -type ErrorMemoryItem struct { - Id int64 `json:"id"` - MemoryKey string `json:"memoryKey"` - Upstream string `json:"upstream"` - ErrorCode string `json:"errorCode"` - MsgFingerprint string `json:"msgFingerprint"` - Retryable bool `json:"retryable"` - Reason string `json:"reason"` - AnalyzedBy string `json:"analyzedBy"` -} - -// DeleteErrorMemoryReq 删除错误重试记忆 -type DeleteErrorMemoryReq struct { - g.Meta `path:"/errorMemory/delete" method:"post" tags:"错误记忆" summary:"删除错误重试记忆" dc:"手动清理永久记忆条目"` - Id int64 `json:"id" v:"required#id不能为空" dc:"记忆ID"` -} -``` - -- [ ] **Step 2: 新建 `dao/error_memory_dao.go`**(未命中处理对齐 `model_manage_dao.Get` 的 `One()+Struct` 语义) - -```go -package dao - -import ( - "context" - "model-gateway/consts/public" - "model-gateway/model/entity" - - "gitea.redpowerfuture.com/red-future/common/db/gfdb" -) - -var ErrorMemory = &errorMemoryDao{} - -type errorMemoryDao struct{} - -// GetByKey 按记忆键查询（未命中返回 (nil, nil)） -func (d *errorMemoryDao) GetByKey(ctx context.Context, key string) (res *entity.ErrorMemory, err error) { - r, err := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameErrorMemory). - Where(entity.ErrorMemoryCol.MemoryKey, key). - One() - if err != nil { - return - } - err = r.Struct(&res) - return -} - -// Upsert 存在则更新 retryable/reason/analyzed_by，不存在则插入 -func (d *errorMemoryDao) Upsert(ctx context.Context, m *entity.ErrorMemory) (err error) { - model := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameErrorMemory) - n, err := model.Where(entity.ErrorMemoryCol.MemoryKey, m.MemoryKey).Count() - if err != nil { - return - } - if n > 0 { - _, err = model.Where(entity.ErrorMemoryCol.MemoryKey, m.MemoryKey).Data(map[string]any{ - entity.ErrorMemoryCol.Retryable: m.Retryable, - entity.ErrorMemoryCol.Reason: m.Reason, - entity.ErrorMemoryCol.AnalyzedBy: m.AnalyzedBy, - }).Update() - return - } - _, err = model.Insert(m) - return -} - -// List 分页查询（按 id 倒序） -// 注:GoFrame v2 的 Model.Count() 返回 int,而公开签名 total 为 int64,须显式转换(实现时按此修正)。 -func (d *errorMemoryDao) List(ctx context.Context, page, pageSize int) (list []entity.ErrorMemory, total int64, err error) { - model := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameErrorMemory) - var n int - if n, err = model.Count(); err != nil { - return - } - total = int64(n) - err = model.Page(page, pageSize).OrderDesc(entity.ErrorMemoryCol.Id).Scan(&list) - return -} - -// Delete 按 id 删除（软删除） -func (d *errorMemoryDao) Delete(ctx context.Context, id int64) (err error) { - _, err = gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameErrorMemory). - Where(entity.ErrorMemoryCol.Id, id).Delete() - return -} -``` - -- [ ] **Step 3: 编译验证** - -Run: `cd model-gateway && go build ./...` -Expected: PASS - -- [ ] **Step 4: 提交** - -```bash -cd model-gateway -git commit --only dao/error_memory_dao.go model/dto/error_memory_dao_dto.go -m "feat: 错误重试记忆DAO与DTO" -``` - ---- - -### Task 5: 对话模型错误分析调用 - -**Files:** -- Modify: `service/error_analysis.go`(追加 `analysisSystemPrompt`、`truncateStr`、`buildAnalysisBody`、`resolveAnalysisModel`、`callAnalysisLLM`) -- Modify: `service/error_analysis_test.go`(追加 body 构造与截断单测) - -**Interfaces:** -- Consumes: `parseAnalysisResponse`(Task 3)、`dto.GetChatModelReq/Res`、`service.ModelManage.GetChatModel`、`entity.ModelManage` -- Produces: `buildAnalysisBody(modelName, code, msg, body string) map[string]any`、`resolveAnalysisModel(ctx, *entity.ModelManage) (*entity.ModelManage, bool)`、`callAnalysisLLM(ctx, model, code, msg, body string) (bool, string, error)`、`truncateStr(s string, max int) string` - -- [ ] **Step 1: 追加失败测试到 `service/error_analysis_test.go`** - -```go -func TestBuildAnalysisBody(t *testing.T) { - body := buildAnalysisBody("doubao-lite", "429", "too many", strings.Repeat("x", 5000)) - msg := body["messages"].([]map[string]string)[1] - if len(msg["content"]) >= 2000+len("429")+len("too many")+100 { - t.Fatalf("响应体应被截断到2000字符, got %d", len(msg["content"])) - } - if body["model"] != "doubao-lite" { - t.Fatalf("model 字段错误") - } - if body["temperature"] != 0 { - t.Fatalf("temperature 应为0") - } -} - -func TestTruncateStr(t *testing.T) { - if truncateStr("abc", 5) != "abc" { - t.Fatalf("短串不应截断") - } - if truncateStr("abcdef", 3) != "abc" { - t.Fatalf("截断错误") - } -} -``` - -(需在文件头 import 增加 `"strings"`。) - -- [ ] **Step 2: 运行测试确认失败** - -Run: `cd model-gateway && go test ./service/ -run 'TestBuildAnalysisBody|TestTruncateStr' -v` -Expected: FAIL(函数未定义) - -- [ ] **Step 3: 追加实现到 `service/error_analysis.go`** - -在 `parseAnalysisResponse` 后追加常量、辅助函数与三个函数。**下面的 `import` 块并入文件顶部已有的 import 块**(Go 的 import 必须在文件顶部,勿放在函数之间): - -```go -import ( - "bytes" - "context" - "encoding/json" - "fmt" - "io" - "net/http" - "strings" - "time" - - "model-gateway/model/dto" - "model-gateway/model/entity" -) - -const ( - analysisTimeout = 15 * time.Second - analysisMaxBody = 2000 - analysisMaxTokens = 256 -) - -const analysisSystemPrompt = `你是 AI 模型网关的错误分析器。上游 AI 模型调用返回了一个错误，你需要判断该错误是否"值得指数退避后重试"。 - -## 值得重试（retryable: true） -- 限流：429、rate limit、请求过密、并发超限 -- 服务端瞬时故障：5xx、服务过载、上游临时不可用、连接抖动 -- 资源暂时不足：quota 暂时受限 - -## 不值得重试（retryable: false） -- 请求/参数错误：400、invalid_argument、格式错误 -- 鉴权失败：401、403、invalid_api_key、签名错误 -- 模型不存在：404、model_not_found -- 余额不足：insufficient_quota -- 内容违规：内容安全拦截 -- 明确的永久性配置错误 - -## 输出 -只输出一个 JSON 对象，不要任何多余文字、解释或代码块标记： -{"retryable": true 或 false, "reason": "不超过20字的简要原因"}` - -// truncateStr 按字节截断到 max（中文可能截半个字符，仅用于分析输入，可接受） -func truncateStr(s string, max int) string { - if len(s) <= max { - return s - } - return s[:max] -} - -// buildAnalysisBody 构造分析请求体（OpenAI 兼容 messages 格式），纯函数便于单测。 -func buildAnalysisBody(modelName, code, msg, body string) map[string]any { - user := fmt.Sprintf("错误码: %s\n错误消息: %s\n错误响应体: %s", code, msg, truncateStr(body, analysisMaxBody)) - return map[string]any{ - "model": modelName, - "messages": []map[string]string{ - {"role": "system", "content": analysisSystemPrompt}, - {"role": "user", "content": user}, - }, - "max_tokens": analysisMaxTokens, - "temperature": 0, - } -} - -// resolveAnalysisModel 选择分析模型：失败模型自身是对话模型则复用，否则取当前用户对话模型。 -// 取不到返回 ok=false，调用方 fail-closed 不重试。 -func resolveAnalysisModel(ctx context.Context, modelInfo *entity.ModelManage) (model *entity.ModelManage, ok bool) { - if modelInfo != nil && modelInfo.ChatModel != nil && *modelInfo.ChatModel { - return modelInfo, true - } - chat, err := ModelManage.GetChatModel(ctx, &dto.GetChatModelReq{}) - if err == nil && chat != nil && chat.ModelManage != nil { - return chat.ModelManage, true - } - return nil, false -} - -// callAnalysisLLM 调分析模型（对话模型）判定错误是否可重试。 -// 独立短超时 http.Client；非 200 / 解析失败 / 超时 → 返回 err，调用方 fail-closed。 -func callAnalysisLLM(ctx context.Context, model *entity.ModelManage, code, msg, body string) (retryable bool, reason string, err error) { - reqBody, err := json.Marshal(buildAnalysisBody(model.ModelName, code, msg, body)) - if err != nil { - return false, "", fmt.Errorf("marshal分析请求失败: %w", err) - } - httpMethod := model.HttpMethod - if httpMethod == "" { - httpMethod = http.MethodPost - } - httpReq, err := http.NewRequestWithContext(ctx, httpMethod, strings.TrimRight(model.BaseURL, "/"), bytes.NewBuffer(reqBody)) - if err != nil { - return false, "", fmt.Errorf("创建分析请求失败: %w", err) - } - for k, v := range model.RequestHeadMapping { - httpReq.Header.Set(k, v) - } - httpReq.Header.Set("Authorization", "Bearer "+model.ApiKey) - httpReq.Header.Set("Content-Type", "application/json") - - client := &http.Client{Timeout: analysisTimeout} - resp, err := client.Do(httpReq) - if err != nil { - return false, "", fmt.Errorf("分析请求失败: %w", err) - } - defer resp.Body.Close() - respBody, err := io.ReadAll(resp.Body) - if err != nil { - return false, "", fmt.Errorf("读取分析响应失败: %w", err) - } - if resp.StatusCode != http.StatusOK { - return false, "", fmt.Errorf("分析接口非200: status=%d body=%s", resp.StatusCode, truncateStr(string(respBody), 500)) - } - var apiResp struct { - Choices []struct { - Message struct { - Content string `json:"content"` - } `json:"message"` - } `json:"choices"` - } - if err = json.Unmarshal(respBody, &apiResp); err != nil { - return false, "", fmt.Errorf("解析分析响应失败: %w", err) - } - if len(apiResp.Choices) == 0 { - return false, "", fmt.Errorf("分析响应无choices") - } - return parseAnalysisResponse(apiResp.Choices[0].Message.Content) -} -``` - -- [ ] **Step 4: 运行测试确认通过** - -Run: `cd model-gateway && go test ./service/ -run 'TestParseAnalysisResponse|TestBuildAnalysisBody|TestTruncateStr' -v` -Expected: PASS - -- [ ] **Step 5: 编译验证 + 提交** - -```bash -cd model-gateway -go build ./... && git add service/error_analysis.go service/error_analysis_test.go -git commit --only service/error_analysis.go service/error_analysis_test.go -m "feat: 对话模型错误分析调用" -``` - ---- - -### Task 6: 统一判定入口 shouldRetryWithMemory + singleflight - -**Files:** -- Modify: `go.mod` / `go.sum`(加 `golang.org/x/sync`) -- Create: `service/error_memory_service.go`(含 `analyzeOnce`、`shouldRetryWithMemory`;Task 8 同文件追加 List/Delete) - -**Interfaces:** -- Consumes: `buildMemoryKey`/`msgFingerprint`(Task 2)、`parseAnalysisResponse`(Task 3)、`dao.ErrorMemory`(Task 4)、`resolveAnalysisModel`/`callAnalysisLLM`(Task 5) -- Produces: `shouldRetryWithMemory(ctx, modelInfo *entity.ModelManage, code, msg, rawBody string) bool`(Task 7 消费)、`analyzeOnce(key string, fn func() (bool, string)) (bool, string)` - -- [ ] **Step 1: 加 singleflight 依赖** - -Run: `cd model-gateway && go get golang.org/x/sync@v0.19.0 && go mod tidy` -Expected: PASS(模块缓存已有 v0.19.0,离线可解析) - -- [ ] **Step 2: 写失败测试 `service/error_memory_service_test.go`** - -```go -package service - -import ( - "sync" - "testing" - "time" -) - -// analyzeOnce 应合并并发同键分析请求，只执行一次 fn -// 注:singleflight 仅对 in-flight 请求去重,故 fn 内须 sleep(50ms) 拉长执行窗口, -// 否则 10 个 goroutine 未必都在首个完成前启动,calls==1 断言会间歇失败(计划原版因此非确定,已修)。 -func TestAnalyzeOnceDedup(t *testing.T) { - var mu sync.Mutex - calls := 0 - fn := func() (bool, string) { - mu.Lock() - calls++ - mu.Unlock() - time.Sleep(50 * time.Millisecond) - return true, "dedup" - } - var wg sync.WaitGroup - for i := 0; i < 10; i++ { - wg.Add(1) - go func() { - defer wg.Done() - if retry, _ := analyzeOnce("k", fn); !retry { - t.Errorf("期望 retryable=true") - } - }() - } - wg.Wait() - if calls != 1 { - t.Fatalf("singleflight应只执行一次fn, got %d", calls) - } -} -``` - -- [ ] **Step 3: 运行测试确认失败** - -Run: `cd model-gateway && go test ./service/ -run TestAnalyzeOnceDedup -v` -Expected: FAIL(analyzeOnce 未定义) - -- [ ] **Step 4: 实现 `service/error_memory_service.go`** - -```go -package service - -import ( - "context" - - "model-gateway/dao" - "model-gateway/model/entity" - - "github.com/gogf/gf/v2/frame/g" - "golang.org/x/sync/singleflight" -) - -var analysisGroup singleflight.Group - -// shouldRetryWithMemory 统一重试判定：查持久记忆，未命中则调分析模型并落库。 -// 记忆/分析/DB 任一环节失败均 fail-closed 不重试。 -func shouldRetryWithMemory(ctx context.Context, modelInfo *entity.ModelManage, code, msg, rawBody string) (retry bool) { - if modelInfo == nil || (code == "" && msg == "") { - return false - } - key := buildMemoryKey(modelInfo.BaseURL, code, msg) - if row, err := dao.ErrorMemory.GetByKey(ctx, key); err != nil { - g.Log().Errorf(ctx, "查询错误重试记忆失败: %v", err) - return false - } else if row != nil { - return row.Retryable - } - retry, _ = analyzeOnce(key, func() (bool, string) { - model, ok := resolveAnalysisModel(ctx, modelInfo) - if !ok { - g.Log().Warningf(ctx, "无可用分析模型(对话模型)，错误不重试: code=%s", code) - return false, "" - } - r, reason, err := callAnalysisLLM(ctx, model, code, msg, rawBody) - if err != nil { - g.Log().Warningf(ctx, "错误分析失败，fail-closed不重试: %v", err) - return false, "" - } - row := &entity.ErrorMemory{ - MemoryKey: key, - Upstream: modelInfo.BaseURL, - ErrorCode: code, - MsgFingerprint: msgFingerprint(msg), - Retryable: r, - Reason: reason, - AnalyzedBy: model.ModelName, - } - if err := dao.ErrorMemory.Upsert(ctx, row); err != nil { - g.Log().Errorf(ctx, "错误重试记忆落库失败: %v", err) - } - return r, reason - }) - return retry -} - -// analyzeOnce 按记忆键合并并发分析请求（singleflight）。 -// 注：fn 失败时结果在本次突发内共享（后续新错误会重新分析）。 -func analyzeOnce(key string, fn func() (bool, string)) (bool, string) { - v, err, _ := analysisGroup.Do(key, func() (any, error) { - retry, reason := fn() - return []any{retry, reason}, nil - }) - if err != nil { - return false, "" - } - vals := v.([]any) - return vals[0].(bool), vals[1].(string) -} -``` - -- [ ] **Step 5: 运行测试确认通过 + 编译** - -Run: `cd model-gateway && go test ./service/ -run TestAnalyzeOnceDedup -v && go build ./...` -Expected: PASS + PASS - -- [ ] **Step 6: 提交** - -```bash -cd model-gateway -git commit --only go.mod go.sum service/error_memory_service.go service/error_memory_service_test.go -m "feat: 重试判定入口shouldRetryWithMemory + singleflight" -``` - ---- - -### Task 7: 三路接入 LLM+记忆重试判定 - -> ⚠️ 本 task 修改的 `service/session_sync.go`、`service/session_stream.go` 含用户 WIP 重构(Global Constraints)。**执行前先让 WIP 落库/经用户确认**。 - -**Files:** -- Modify: `service/session_sync.go`(重试判定替换) -- Modify: `service/session_stream.go`(`streamRetryCodeOfError` → `streamErrorInfoOfError`;两处判定替换) -- Modify: `service/model_task_start_service.go`(新增重试循环) -- Modify: `service/retry.go`(删除 `isRetryableErrorCode`) - -**Interfaces:** -- Consumes: `shouldRetryWithMemory`(Task 6) -- Produces: 无(行为改造) - -- [ ] **Step 1: 删除 `service/retry.go` 的固定错误码清单** - -删除 `isRetryableErrorCode` 函数及其注释(保留 `modelCallMaxRetries`、`retryWait`、`firstText`、`extractChunkText`)。若删除后 `strings`/`gconv` import 变未使用,同步清理。 - -- [ ] **Step 2: 改同步路径 `service/session_sync.go`** - -把 `CreateSession` 中(约 L58-68): - -```go - if errCode != "" { - - if attempt < modelCallMaxRetries && isRetryableErrorCode(errCode) { - attempt++ - wait := time.Duration(1<= 0 { - codeStr := strings.TrimSpace(e[idx+len("状态码异常: "):]) - if comma := strings.IndexByte(codeStr, ','); comma >= 0 { - codeStr = codeStr[:comma] - } - return codeStr, "" - } - return "", "" -} -``` - -(2) `CreateSessionStreamOnce` HTTP 错误重试点(约 L33-48)改为: - -```go - streamReader, err := httpclient.ModelHttpStreamRequest(ctx, nil, modelInfo.BaseURL, modelInfo.RequestHeadMapping, modelInfo.HttpMethod, newRequestParams) - if err != nil { - if attempt < modelCallMaxRetries { - if code, msg := streamErrorInfoOfError(err); shouldRetryWithMemory(ctx, modelInfo, code, msg, "") { - attempt++ - wait := time.Duration(1< ⚠️ 该文件已存在且含 Task 6 内容(`analysisGroup`/`analyzeOnce`/`shouldRetryWithMemory`)。**不要整文件覆盖**——只**追加**以下内容到文件尾部(`var ErrorMemory`/`errorMemoryService`/`List`/`Delete`),不再重复 `package service`;`model-gateway/model/dto` import 并入顶部既有 import 块(`context`/`dao` 已存在)。 - -```go -var ErrorMemory = &errorMemoryService{} - -type errorMemoryService struct{} - -// List 错误重试记忆列表 -func (s *errorMemoryService) List(ctx context.Context, req *dto.GetErrorMemoryListReq) (res *dto.GetErrorMemoryListRes, err error) { - page, size := 1, 20 - if req.Page != nil && req.Page.PageNum > 0 { - page = int(req.Page.PageNum) - } - if req.Page != nil && req.Page.PageSize > 0 { - size = int(req.Page.PageSize) - } - list, total, err := dao.ErrorMemory.List(ctx, page, size) - if err != nil { - return nil, err - } - res = &dto.GetErrorMemoryListRes{Total: total, List: make([]dto.ErrorMemoryItem, 0, len(list))} - for _, m := range list { - res.List = append(res.List, dto.ErrorMemoryItem{ - Id: m.Id, - MemoryKey: m.MemoryKey, - Upstream: m.Upstream, - ErrorCode: m.ErrorCode, - MsgFingerprint: m.MsgFingerprint, - Retryable: m.Retryable, - Reason: m.Reason, - AnalyzedBy: m.AnalyzedBy, - }) - } - return res, nil -} - -// Delete 删除错误重试记忆（手动纠错永久记忆） -func (s *errorMemoryService) Delete(ctx context.Context, req *dto.DeleteErrorMemoryReq) (err error) { - return dao.ErrorMemory.Delete(ctx, req.Id) -} -``` - -> ⚠️ `service.ErrorMemory`(本服务)与 `dao.ErrorMemory` 同名不同包,合法。但 `shouldRetryWithMemory` 中引用的 `dao.ErrorMemory` 需保持包前缀,勿混。 - -- [ ] **Step 2: 新建 `controller/error_memory_controller.go`** - -```go -package controller - -import ( - "context" - "model-gateway/model/dto" - "model-gateway/service" - - "gitea.redpowerfuture.com/red-future/common/beans" -) - -// ErrorMemory 错误重试记忆控制器 -var ErrorMemory = new(errorMemory) - -type errorMemory struct{} - -// List 错误重试记忆列表 -func (c *errorMemory) List(ctx context.Context, req *dto.GetErrorMemoryListReq) (res *dto.GetErrorMemoryListRes, err error) { - return service.ErrorMemory.List(ctx, req) -} - -// Delete 删除错误重试记忆 -func (c *errorMemory) Delete(ctx context.Context, req *dto.DeleteErrorMemoryReq) (res *beans.ResponseEmpty, err error) { - err = service.ErrorMemory.Delete(ctx, req) - return -} -``` - -- [ ] **Step 3: `main.go` 注册控制器** - -把路由注册改为: - -```go - http.RouteRegister([]interface{}{ - controller.ModelCall, - controller.ModelManage, - controller.ErrorMemory, - }) -``` - -- [ ] **Step 4: 编译验证** - -Run: `cd model-gateway && go build ./...` -Expected: PASS - -- [ ] **Step 5: 手动验证端点(需运行环境)** - -- `GET /errorMemory/list?page[pageNum]=1&page[pageSize]=20` → 返回记忆条目 -- `POST /errorMemory/delete` body `{"id": 1}` → 删除成功,再次 list 不含该条 - -- [ ] **Step 6: 提交** - -```bash -cd model-gateway -git commit --only service/error_memory_service.go controller/error_memory_controller.go main.go -m "feat: 错误记忆管理端点" -``` - ---- - -## Self-Review - -- **Spec 覆盖**:spec 的 8 项决策(LLM 全判断/DB 知识库/键构成/二元输出/永久/对话模型/三路/内联+单飞)分别落在 Task 7、Task 1/4、Task 2、Task 3/5、Task 1/8、Task 5、Task 7、Task 6。管理端点 Task 8;测试策略分布各 Task 单测。无缺口。 -- **占位符扫描**:全部步骤含实际代码,无 TBD/TODO/"适当处理"类占位。 -- **类型一致性**:`shouldRetryWithMemory` 签名在 Task 6 定义、Task 7 三处调用一致(`code/msg string`、`rawBody string`);`buildMemoryKey(upstream, code, msg)` 与 `msgFingerprint(msg)` 命名全链一致;`streamErrorInfoOfError(err) (code, msg string)` 在 Task 7 定义与使用一致;DAO 方法签名 Task 4→Task 6/8 一致。`analysisSystemPrompt`/`analysisTimeout` 等常量单一定义处。 -- **已知行为**:singleflight 失败结果在突发内共享(后续新错误重新分析);永久记忆靠端点清理;task_start 重试可能重复建任务——均为 spec 已记录取舍。 diff --git a/docs/superpowers/specs/2026-09-01-llm-error-retry-memory-design.md b/docs/superpowers/specs/2026-09-01-llm-error-retry-memory-design.md deleted file mode 100644 index d256e67..0000000 --- a/docs/superpowers/specs/2026-09-01-llm-error-retry-memory-design.md +++ /dev/null @@ -1,173 +0,0 @@ -# 模型网关 LLM 驱动重试决策 + 持久错误记忆 设计 - -> **日期**:2026-09-01 -> **范围**:model-gateway(仅本服务) -> **状态**:已确认 - -## Goal - -请求上游模型失败时,不再用固定错误码清单决定是否重试,改为**调用对话模型分析错误类型**决定是否重试,并持久化"错误 → 结论"的记忆(知识库),命中记忆时不再调用分析模型。 - -## 背景与现状 - -当前重试判定(`service/retry.go:isRetryableErrorCode`)基于固定错误码集合 `429/500/501/502/503/InvalidParameter/limit_requests/limit_tokens/rate_limit_exceeded`。判定点两处: - -- 同步:`service/session_sync.go:60`(`CreateSession`) -- 流式:`service/session_stream.go:88`,及 `streamRetryCodeOfError` 中对 5xx 的硬编码分支 - -异步任务启动 `service/model_task_start_service.go`(`CreateTask`)目前**没有重试**。 - -错误解析已配置驱动:`parseModelError`(按模型 `ErrorMessageMapping` 提取 code/message)。 - -## 设计决策(已确认) - -| # | 决策 | -|---|---| -| 1 | **完全交给 LLM 判断**:删除固定错误码清单,所有错误(记忆未命中时)都调分析模型决定是否重试 | -| 2 | **持久知识库**:结论存 PostgreSQL 新表,跨实例共享、重启不丢 | -| 3 | **记忆键** = 失败上游 BaseURL + 错误码 + 归一化消息指纹(同因同键、跨实例命中) | -| 4 | **二元输出**:`{retryable: bool, reason: str}` | -| 5 | **永久有效**:条目无 TTL,直到同键重新分析覆盖或经管理端点手动删除 | -| 6 | **分析模型 = 对话模型**(不新增配置段):失败模型自身 `chat_model=true` 优先复用;否则取当前用户对话模型(`GetChatModel`);都没有 → fail-closed 不重试 | -| 7 | **三路接入**:同步 / 流式 / 异步任务启动 | -| 8 | **内联分析 + singleflight 去重**:未命中时同步调分析模型;同键并发 miss 合并为一次分析 | - -## 架构与数据流 - -``` -请求失败(parseModelError 得 code+msg) - └─ shouldRetryWithMemory(ctx, modelInfo, code, msg) → bool: - 1) 归一化消息 → 记忆键 = SHA-256(upstream|code|fingerprint) - 2) 查 model_gateway_error_memory - ├─ 命中 → 返回存储的 retryable - └─ 未命中 → singleflight 合并后调分析模型 - → 解析 {retryable, reason} → UPSERT 落库 → 返回 - 3) retryable=true → 现有指数退避重试(modelCallMaxRetries=10 预算) - retryable=false → 记 ErrorMsg 走现有终态逻辑(不重试) -``` - -## 记忆知识库(新表 `model_gateway_error_memory`) - -表前缀 `model_gateway_`,DDL 追加到 `update.sql`,DAO/entity/consts 按现有惯例新建。 - -| 列 | 类型 | 说明 | -|---|---|---| -| `id` | BIGSERIAL PK | | -| `memory_key` | CHAR(64) | SHA-256(upstream\|code\|fingerprint),**唯一索引** | -| `upstream` | VARCHAR | 失败上游 BaseURL | -| `error_code` | VARCHAR | 解析出的错误码 | -| `msg_fingerprint` | CHAR(32) | 归一化消息 md5 | -| `retryable` | BOOLEAN | LLM 结论 | -| `reason` | VARCHAR | LLM 给出的简短原因(可观测) | -| `analyzed_by` | VARCHAR | 分析所用模型名 | -| `created_at` / `updated_at` | 自动 | 永久有效,无 TTL | - -### 归一化 `normalizeErrorMsg` - -正则剔除易变片段,使同因不同实例命中同一键: - -- UUID(8-4-4-4-12 hex) -- ISO 8601 时间戳 / unix 秒与毫秒数字 -- `req-xxx` / `request-xxx` 类请求 ID -- 连续 ≥4 位数字(去掉具体数值,保留位置标记) - -### DAO / entity / consts - -按 `dao/model_task_start_dao.go` 模式:包名 `dao`,变量 `var ErrorMemory = &errorMemoryDao{}`;表名/库名常量进 `consts/public`;entity 带列常量(`entity.ErrorMemoryCol.*`)。 - -## 分析模型与调用(不新增 config) - -### 分析模型选择 - -1. 失败请求的 `modelInfo` 本身 `chat_model=true` → 直接用它(复用其 baseURL/apiKey/HttpMethod) -2. 否则 → `GetChatModel`(当前用户对话模型,现有语义 Creator + chat_model=true) -3. 都没有 → **fail-closed 不重试** + 日志 - -依赖:重试循环内 `ctx` 需携带用户信息(与现有 X-User-Info 约定一致)。 - -### 调用方式 - -**用独立短超时 `http.Client{Timeout: 15s}`** 对分析模型发 OpenAI messages 格式(技术修正:`httpclient.ModelHttpNormalRequest` 底层 `modelDoRaw` 响应头超时 30min 且不区分 HTTP 状态码,用作重试判定会拖死调用,故不复用它): - -```json -{ - "model": "<分析模型 modelName>", - "messages": [ - {"role": "system", "content": "<判定错误是否可重试的系统提示词>"}, - {"role": "user", "content": "{error_code, error_message, 截断错误响应体}"} - ], - "max_tokens": 256, - "temperature": 0 -} -``` - -### Prompt(要点) - -- system:定义任务——判断上游模型错误是否值得指数退避重试;考虑限流/瞬时故障/配置/参数/鉴权等类型;输出严格 JSON。 -- user:携带 `error_code`、`error_message`、截断的错误响应体(≤2000 字符)。 -- 输出解析容错:剥 ```json 代码块 / 前后空白 / 首个 `{...}` 提取。 - -### 失败处理 - -分析调用超时(15s)/失败/解析失败 → **fail-closed 不重试** + 日志。分析失败不影响记忆表。 - -## 三路接入 - -| 路径 | 文件 | 改动 | -|---|---|---| -| 同步 | `service/session_sync.go:60` | `isRetryableErrorCode(errCode)` → `shouldRetryWithMemory(...)` | -| 流式 | `service/session_stream.go:88` | 同上替换;`streamRetryCodeOfError` 中硬编码 5xx 分支收敛(移除) | -| 异步启动 | `service/model_task_start_service.go` | 新增重试循环(现无重试),错误时走同一判定 | - -统一入口落在 `service/retry.go`: - -- 删除 `isRetryableErrorCode` 固定清单 -- 新增 `shouldRetryWithMemory(ctx, modelInfo, code, msg) (retry bool)`(内部:查记忆 → miss 则分析 → 落库) -- 保留 `modelCallMaxRetries=10` 与 `retryWait` 指数退避 -- 新增单飞:`golang.org/x/sync/singleflight.Group` 按记忆键合并并发分析(go.sum 已有传递版本,加为直接依赖即可;避免自实现 keyed mutex) - -### 管理端点(手动清理记忆) - -- `GET /errorMemory/list` — 分页查看记忆条目 -- `POST /errorMemory/delete` — 按 id 删除条目(永久记忆的手动纠错途径) - -## 边界与降级 - -| 场景 | 行为 | -|---|---| -| 记忆查询失败(DB 异常) | 不重试 + 日志,不阻塞主流程 | -| 分析模型调用失败/超时/解析失败 | 不重试(fail-closed)+ 日志 | -| 同键并发 miss | singleflight 合并,只调一次分析模型 | -| 消息过长 | 截断到 2000 字符再送分析 | -| 分析模型取不到 | 不重试 + 日志 | -| 重试仍失败(同键) | 烧满 `modelCallMaxRetries` 预算;永久记忆条目保留,靠管理端点手动清理 | - -## 已知取舍 - -- **永久记忆**:上游行为变化后旧结论可能过时。接受,靠管理端点手动纠错;自动"连续失败 N 次降级"明确**不做**(YAGNI,留作未来)。 -- **异步 task_start 重试**:重试=重新调用创建任务;首次调用已报错大概率未建任务,接受"响应丢失但任务已建 → 重复建任务"的既有语义(与同步/流式行为一致)。 - -## 测试策略 - -**单元测试** -- `normalizeErrorMsg`:UUID / 时间戳 / 请求 ID / 连续数字被剔除;稳定文本不变 -- 记忆键构造:同 code+同归一化消息同键;不同上游不同键 -- 分析响应解析:纯 JSON / ```json 代码块 / 前后缀 / 非法响应(返回失败) - -**集成测试** -- 记忆命中:直接复用存储结论,不调分析模型 -- 未命中:调分析模型 → 落库 → 按结论重试/不重试 -- 分析失败:不重试,不落库 -- 并发同键:多个 goroutine 只触发一次分析调用 -- 管理端点:list / delete - -**三路径** -- 同步 / 流式:重试预算与退避行为保持现状,仅判定来源替换 -- 异步 task_start:新增重试循环行为 - -## 非目标 - -- 不做固定错误码快路径(决策 1 已排除) -- 不做记忆自动过期 / 命中计数降级(决策 5 + YAGNI) -- 不新增 config 段(决策 6) -- 不接 ai-agent / prompts-core 做分析(分析在 model-gateway 内完成) diff --git a/go.mod b/go.mod index c5bb233..fb64bd3 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module model-gateway go 1.26.1 require ( - gitea.redpowerfuture.com/red-future/common v0.0.32 + gitea.redpowerfuture.com/red-future/common v0.0.33 github.com/bjang03/gmq v0.0.3 github.com/gogf/gf/contrib/drivers/pgsql/v2 v2.10.2 github.com/gogf/gf/contrib/nosql/redis/v2 v2.10.2 @@ -11,8 +11,6 @@ require ( golang.org/x/sync v0.19.0 ) -replace gitea.redpowerfuture.com/red-future/common v0.0.32 => ../common - require ( github.com/BurntSushi/toml v1.5.0 // indirect github.com/armon/go-metrics v0.4.1 // indirect diff --git a/main.go b/main.go index f746aa8..efd44ac 100644 --- a/main.go +++ b/main.go @@ -36,7 +36,7 @@ func main() { http.RouteRegister([]interface{}{ controller.ModelCall, controller.ModelManage, - controller.ErrorMemory, + controller.ModelErrorMemory, }) gmq.GmqRegister(public.GmqMsgPluginsName, &mq.NatsConn{ diff --git a/model/dto/model_call_dto.go b/model/dto/model_call_dto.go index 55cb970..c8ff709 100644 --- a/model/dto/model_call_dto.go +++ b/model/dto/model_call_dto.go @@ -17,6 +17,8 @@ type ModelCallReq struct { type ModelCallRes struct { TaskId int64 `json:"id" dc:"任务ID"` + ModelId int64 `json:"modelId" dc:"生效模型ID（引用行=解析后的系统模型ID，计价按此）"` + MediaType string `json:"mediaType" dc:"输入媒体类型（shop词汇: text/audio/video）"` TotalTokens int64 `json:"totalTokens" dc:"总token"` PromptTokens int64 `json:"promptTokens" dc:"输入token"` CompletionTokens int64 `json:"completionTokens" dc:"输出token"` @@ -61,6 +63,8 @@ type ModelCallStreamReq struct { type ModelMsg struct { TaskID int64 `json:"id" dc:"任务ID"` + ModelId int64 `json:"modelId" dc:"生效模型ID（引用行=解析后的系统模型ID，计价按此）"` + MediaType string `json:"mediaType" dc:"输入媒体类型（shop词汇: text/audio/video）"` TotalTokens int64 `json:"totalTokens" dc:"总token"` PromptTokens int64 `json:"promptTokens" dc:"输入token"` CompletionTokens int64 `json:"completionTokens" dc:"输出token"` diff --git a/model/dto/error_memory_dao_dto.go b/model/dto/model_error_memory_dto.go similarity index 100% rename from model/dto/error_memory_dao_dto.go rename to model/dto/model_error_memory_dto.go diff --git a/model/dto/model_manage_dto_test.go b/model/dto/model_manage_dto_test.go deleted file mode 100644 index 1e73601..0000000 --- a/model/dto/model_manage_dto_test.go +++ /dev/null @@ -1,24 +0,0 @@ -package dto - -import ( - "testing" - - "model-gateway/model/entity" - - "github.com/gogf/gf/v2/util/gconv" -) - -// 引用行字段落地验证：DTO → entity 映射（CreateModelManageReq.RefSystemModelId → entity.RefSystemModelId） -func TestRefSystemModelIdRoundtrip(t *testing.T) { - req := CreateModelManageReq{ModelName: "gpt-4o", RefSystemModelId: 100} - var e entity.ModelManage - if err := gconv.Struct(&req, &e); err != nil { - t.Fatalf("gconv dto->entity: %v", err) - } - if e.RefSystemModelId != 100 { - t.Fatalf("refSystemModelId mismatch: got %d want 100", e.RefSystemModelId) - } - if e.ModelName != "gpt-4o" { - t.Fatalf("modelName mismatch: got %s", e.ModelName) - } -} diff --git a/model/dto/model_task_start_dto.go b/model/dto/model_task_start_dto.go index 528987a..0e0c285 100644 --- a/model/dto/model_task_start_dto.go +++ b/model/dto/model_task_start_dto.go @@ -19,7 +19,7 @@ type CreateModelTaskStartReq struct { MsgTopic string `json:"msgTopic" dc:"消息主题（可选，用于后续业务通知）"` RequestPath string `json:"requestPath" dc:"请求参数保存路径"` OriginalRequestPath string `json:"originalRequestPath" dc:"原始请求参数保存路径"` - MediaType string `json:"mediaType" dc:"输入媒体类型快照（audio/no_video/has_video，创建任务时按请求体推导）"` + MediaType string `json:"mediaType" dc:"输入媒体类型快照（audio/video，空=无媒体引用；shop 计费词汇，创建任务时按请求体推导）"` } type CreateModelTaskStartRes struct { diff --git a/model/entity/error_memory.go b/model/entity/model_error_memory.go similarity index 85% rename from model/entity/error_memory.go rename to model/entity/model_error_memory.go index c71df0e..426593f 100644 --- a/model/entity/error_memory.go +++ b/model/entity/model_error_memory.go @@ -2,7 +2,7 @@ package entity import "gitea.redpowerfuture.com/red-future/common/beans" -type errorMemoryCol struct { +type modelErrorMemoryCol struct { beans.SQLBaseCol MemoryKey string Upstream string @@ -13,7 +13,7 @@ type errorMemoryCol struct { AnalyzedBy string } -var ErrorMemoryCol = errorMemoryCol{ +var ModelErrorMemoryCol = modelErrorMemoryCol{ SQLBaseCol: beans.DefSQLBaseCol, MemoryKey: "memory_key", Upstream: "upstream", @@ -24,8 +24,8 @@ var ErrorMemoryCol = errorMemoryCol{ AnalyzedBy: "analyzed_by", } -// ErrorMemory 错误重试记忆（LLM 分析结论持久化，永久有效） -type ErrorMemory struct { +// ModelErrorMemory 错误重试记忆（LLM 分析结论持久化，永久有效） +type ModelErrorMemory struct { beans.SQLBaseDO `orm:",inline"` MemoryKey string `orm:"memory_key" json:"memoryKey" dc:"记忆键=SHA-256(upstream|code|归一化消息)"` Upstream string `orm:"upstream" json:"upstream" dc:"失败上游BaseURL"` diff --git a/model/entity/model_task_start.go b/model/entity/model_task_start.go index ad60f28..cc6fd18 100644 --- a/model/entity/model_task_start.go +++ b/model/entity/model_task_start.go @@ -49,6 +49,6 @@ type ModelTaskStart struct { OriginalResponseParams map[string]any `orm:"original_response_params" json:"originalResponseParams" dc:"原始响应结果"` DurationSeconds int64 `orm:"duration_seconds" json:"durationSeconds" dc:"耗时（秒）"` TaskId string `orm:"task_id" json:"taskId" dc:"任务ID"` - MediaType string `orm:"media_type" json:"mediaType" dc:"输入媒体类型快照（audio/no_video/has_video，创建任务时按请求体推导）"` + MediaType string `orm:"media_type" json:"mediaType" dc:"输入媒体类型快照（audio/video，空=无媒体引用；shop 计费词汇，创建任务时按请求体推导）"` ErrorMsg string `orm:"error_msg" json:"errorMsg" dc:"错误消息"` } diff --git a/service/error_analysis.go b/service/error_analysis.go index 8d311e2..55dcc5f 100644 --- a/service/error_analysis.go +++ b/service/error_analysis.go @@ -12,6 +12,8 @@ import ( "model-gateway/model/dto" "model-gateway/model/entity" + + "github.com/gogf/gf/v2/frame/g" ) // parseAnalysisResponse 解析分析模型输出的判定 JSON。 @@ -71,8 +73,11 @@ func truncateStr(s string, max int) string { } // buildAnalysisBody 构造分析请求体（OpenAI 兼容 messages 格式），纯函数便于单测。 -func buildAnalysisBody(modelName, code, msg, body string) map[string]any { +func buildAnalysisBody(ctx context.Context, modelName, code, msg, body string) map[string]any { user := fmt.Sprintf("错误码: %s\n错误消息: %s\n错误响应体: %s", code, msg, truncateStr(body, analysisMaxBody)) + // 打印错误信息 + g.Log().Debugf(ctx, "分析请求体: %s", user) + return map[string]any{ "model": modelName, "messages": []map[string]string{ @@ -100,7 +105,7 @@ func resolveAnalysisModel(ctx context.Context, modelInfo *entity.ModelManage) (m // callAnalysisLLM 调分析模型（对话模型）判定错误是否可重试。 // 独立短超时 http.Client；非 200 / 解析失败 / 超时 → 返回 err，调用方 fail-closed。 func callAnalysisLLM(ctx context.Context, model *entity.ModelManage, code, msg, body string) (retryable bool, reason string, err error) { - reqBody, err := json.Marshal(buildAnalysisBody(model.ModelName, code, msg, body)) + reqBody, err := json.Marshal(buildAnalysisBody(ctx, model.ModelName, code, msg, body)) if err != nil { return false, "", fmt.Errorf("marshal分析请求失败: %w", err) } diff --git a/service/error_analysis_test.go b/service/error_analysis_test.go deleted file mode 100644 index 94a5124..0000000 --- a/service/error_analysis_test.go +++ /dev/null @@ -1,69 +0,0 @@ -package service - -import ( - "strings" - "testing" -) - -func TestParseAnalysisResponse(t *testing.T) { - cases := []struct { - name string - in string - wantRetry bool - wantErr bool - }{ - {"纯JSON", `{"retryable": true, "reason": "限流"}`, true, false}, - {"带json代码块", "```json\n{\"retryable\": false, \"reason\": \"参数错误\"}\n```", false, false}, - {"带前后文字", `分析结果: {"retryable": true, "reason": "瞬时故障"} 完毕`, true, false}, - {"非法响应", `抱歉，我无法分析`, false, true}, - {"空串", ``, false, true}, - } - for _, c := range cases { - retry, _, err := parseAnalysisResponse(c.in) - if c.wantErr { - if err == nil { - t.Fatalf("[%s] 期望错误, got retry=%v", c.name, retry) - } - continue - } - if err != nil { - t.Fatalf("[%s] 不应报错: %v", c.name, err) - } - if retry != c.wantRetry { - t.Fatalf("[%s] retryable = %v, want %v", c.name, retry, c.wantRetry) - } - } -} - -func TestParseAnalysisResponseReason(t *testing.T) { - _, reason, err := parseAnalysisResponse(`{"retryable": true, "reason": "服务过载"}`) - if err != nil { - t.Fatalf("unexpected err: %v", err) - } - if reason != "服务过载" { - t.Fatalf("reason = %q, want 服务过载", reason) - } -} - -func TestBuildAnalysisBody(t *testing.T) { - body := buildAnalysisBody("doubao-lite", "429", "too many", strings.Repeat("x", 5000)) - msg := body["messages"].([]map[string]string)[1] - if len(msg["content"]) >= 2000+len("429")+len("too many")+100 { - t.Fatalf("响应体应被截断到2000字符, got %d", len(msg["content"])) - } - if body["model"] != "doubao-lite" { - t.Fatalf("model 字段错误") - } - if body["temperature"] != 0 { - t.Fatalf("temperature 应为0") - } -} - -func TestTruncateStr(t *testing.T) { - if truncateStr("abc", 5) != "abc" { - t.Fatalf("短串不应截断") - } - if truncateStr("abcdef", 3) != "abc" { - t.Fatalf("截断错误") - } -} diff --git a/service/error_memory_service_test.go b/service/error_memory_service_test.go deleted file mode 100644 index 02cd3a0..0000000 --- a/service/error_memory_service_test.go +++ /dev/null @@ -1,35 +0,0 @@ -package service - -import ( - "sync" - "testing" - "time" -) - -// analyzeOnce 应合并并发同键分析请求，只执行一次 fn -func TestAnalyzeOnceDedup(t *testing.T) { - var mu sync.Mutex - calls := 0 - fn := func() (bool, string) { - mu.Lock() - calls++ - mu.Unlock() - // 短暂阻塞确保所有并发调用在 singleflight 窗口内到达 Do，仅首次执行 fn - time.Sleep(50 * time.Millisecond) - return true, "dedup" - } - var wg sync.WaitGroup - for i := 0; i < 10; i++ { - wg.Add(1) - go func() { - defer wg.Done() - if retry, _ := analyzeOnce("k", fn); !retry { - t.Errorf("期望 retryable=true") - } - }() - } - wg.Wait() - if calls != 1 { - t.Fatalf("singleflight应只执行一次fn, got %d", calls) - } -} diff --git a/service/error_memory_test.go b/service/error_memory_test.go deleted file mode 100644 index 947bd2e..0000000 --- a/service/error_memory_test.go +++ /dev/null @@ -1,43 +0,0 @@ -package service - -import "testing" - -func TestNormalizeErrorMsg(t *testing.T) { - cases := []struct{ in, want string }{ - {"rate limit exceeded for req-abc123", "rate limit exceeded for {reqid}"}, - {"timeout at 2026-09-01T09:00:00Z req_88f1a2", "timeout at {time} {reqid}"}, - {"uuid 0f8fad5b-d9cb-469f-a165-70867728950e remains", "uuid {uuid} remains"}, - {"err code 12345678 quota exceeded", "err code {num} quota exceeded"}, - {"clean message unchanged", "clean message unchanged"}, - } - for _, c := range cases { - if got := normalizeErrorMsg(c.in); got != c.want { - t.Fatalf("normalizeErrorMsg(%q) = %q, want %q", c.in, got, c.want) - } - } -} - -func TestBuildMemoryKey(t *testing.T) { - k1 := buildMemoryKey("https://a.com", "429", "rate limit for req-a1b2c3") - k2 := buildMemoryKey("https://a.com", "429", "rate limit for req-d4e5f6") // 同因不同请求ID - if k1 != k2 { - t.Fatalf("同因不同请求ID应同键: %s != %s", k1, k2) - } - k3 := buildMemoryKey("https://b.com", "429", "rate limit for req-a1b2c3") // 不同上游 - if k1 == k3 { - t.Fatalf("不同上游应不同键") - } - k4 := buildMemoryKey("https://a.com", "500", "rate limit for req-a1b2c3") // 不同错误码 - if k1 == k4 { - t.Fatalf("不同错误码应不同键") - } - if len(k1) != 64 { - t.Fatalf("memory_key 应为 SHA-256 十六进制64位, got %d", len(k1)) - } -} - -func TestMsgFingerprint(t *testing.T) { - if msgFingerprint("a req-a1b2c3") != msgFingerprint("a req-d4e5f6") { - t.Fatalf("同因指纹应一致") - } -} diff --git a/service/model_call_service.go b/service/model_call_service.go index c4a9d02..4d84d6e 100644 --- a/service/model_call_service.go +++ b/service/model_call_service.go @@ -197,7 +197,7 @@ func (s *modelCallService) saveModelRequestParams(ctx context.Context, now time. return 0, nil, fmt.Errorf("上传模型解析请求参数文件失败：%v", err) } - // 3) 保存模型请求信息（快照媒体类型供任务完成时换算费用；模型计费配置任务完成时按 modelId 现查） + // 3) 保存模型请求信息（快照媒体类型=shop 计费词汇 audio/video，空=无媒体引用，任务完成时直接用于算费；模型计费配置任务完成时按 modelId 现查） if *modelInfo.ResponseType == *model.ResponseTypeAsync.Code() { id, err = dao.ModelTaskStart.Insert(ctx, &dto.CreateModelTaskStartReq{ ModelId: req.ModelId, diff --git a/service/error_memory_service.go b/service/model_error_memory_service.go similarity index 81% rename from service/error_memory_service.go rename to service/model_error_memory_service.go index 750fd06..6a264fd 100644 --- a/service/error_memory_service.go +++ b/service/model_error_memory_service.go @@ -20,7 +20,7 @@ func shouldRetryWithMemory(ctx context.Context, modelInfo *entity.ModelManage, c return false } key := buildMemoryKey(modelInfo.BaseURL, code, msg) - if row, err := dao.ErrorMemory.GetByKey(ctx, key); err != nil { + if row, err := dao.ModelErrorMemory.GetByKey(ctx, key); err != nil { g.Log().Errorf(ctx, "查询错误重试记忆失败: %v", err) return false } else if row != nil { @@ -37,7 +37,7 @@ func shouldRetryWithMemory(ctx context.Context, modelInfo *entity.ModelManage, c g.Log().Warningf(ctx, "错误分析失败，fail-closed不重试: %v", err) return false, "" } - row := &entity.ErrorMemory{ + row := &entity.ModelErrorMemory{ MemoryKey: key, Upstream: modelInfo.BaseURL, ErrorCode: code, @@ -46,7 +46,7 @@ func shouldRetryWithMemory(ctx context.Context, modelInfo *entity.ModelManage, c Reason: reason, AnalyzedBy: model.ModelName, } - if err := dao.ErrorMemory.Upsert(ctx, row); err != nil { + if err := dao.ModelErrorMemory.Upsert(ctx, row); err != nil { g.Log().Errorf(ctx, "错误重试记忆落库失败: %v", err) } return r, reason @@ -68,12 +68,12 @@ func analyzeOnce(key string, fn func() (bool, string)) (bool, string) { return vals[0].(bool), vals[1].(string) } -var ErrorMemory = &errorMemoryService{} +var ModelErrorMemory = &modelErrorMemoryService{} -type errorMemoryService struct{} +type modelErrorMemoryService struct{} // List 错误重试记忆列表 -func (s *errorMemoryService) List(ctx context.Context, req *dto.GetErrorMemoryListReq) (res *dto.GetErrorMemoryListRes, err error) { +func (s *modelErrorMemoryService) List(ctx context.Context, req *dto.GetErrorMemoryListReq) (res *dto.GetErrorMemoryListRes, err error) { page, size := 1, 20 if req.Page != nil && req.Page.PageNum > 0 { page = int(req.Page.PageNum) @@ -81,7 +81,7 @@ func (s *errorMemoryService) List(ctx context.Context, req *dto.GetErrorMemoryLi if req.Page != nil && req.Page.PageSize > 0 { size = int(req.Page.PageSize) } - list, total, err := dao.ErrorMemory.List(ctx, page, size) + list, total, err := dao.ModelErrorMemory.List(ctx, page, size) if err != nil { return nil, err } @@ -102,6 +102,6 @@ func (s *errorMemoryService) List(ctx context.Context, req *dto.GetErrorMemoryLi } // Delete 删除错误重试记忆（手动纠错永久记忆） -func (s *errorMemoryService) Delete(ctx context.Context, req *dto.DeleteErrorMemoryReq) (err error) { - return dao.ErrorMemory.Delete(ctx, req.Id) +func (s *modelErrorMemoryService) Delete(ctx context.Context, req *dto.DeleteErrorMemoryReq) (err error) { + return dao.ModelErrorMemory.Delete(ctx, req.Id) } diff --git a/service/model_task_end_service.go b/service/model_task_end_service.go index e4cbb9b..ec6b1f6 100644 --- a/service/model_task_end_service.go +++ b/service/model_task_end_service.go @@ -314,6 +314,8 @@ LOOP: } // 调 shop-user-trade 按用量算费（媒体类型取任务创建时的快照；subject=解析后的系统模型 id） + docMsg.ModelId = modelInfo.Id // 引用行=系统模型 id，供 per_token 结算按系统模型计价 + docMsg.MediaType = item.MediaType docMsg.Cost = calcModelCost(asyncCtx, modelInfo.Id, buildModelUsage(docMsg.PromptTokens, docMsg.CompletionTokens, 0, item.MediaType, docMsg.Duration)) } diff --git a/service/pricing_client.go b/service/pricing_client.go index 12c0f60..ef268d8 100644 --- a/service/pricing_client.go +++ b/service/pricing_client.go @@ -36,18 +36,6 @@ func pricingURL(sub string) string { return "shop-user-trade/pricing/controller/" + sub } -// mgMediaTypeToShop 媒体类型词汇映射：model-gateway（audio/no_video/has_video）→ shop-user-trade（text/audio/video/image） -func mgMediaTypeToShop(mg string) string { - switch mg { - case "audio": - return "audio" - case "has_video": - return "video" - default: // no_video - return "text" - } -} - // walletURL 组装 shop-user-trade 钱包接口地址（accountController → account/controller，与 pricing 同 RouteRegister 推导规则） func walletURL(sub string) string { return "shop-user-trade/account/controller/" + sub @@ -87,7 +75,9 @@ func modelBillable(ctx context.Context, modelId int64) error { return nil } -// buildModelUsage 组装算费用量 JSON 对象（ChargeUsage 形状）。mediaType 为 model-gateway 词汇（DetectMediaType/异步快照）。 +// buildModelUsage 组装算费用量 JSON 对象（ChargeUsage 形状）。mediaType 为 shop 计费词汇 +// （audio/video，空=无媒体引用走默认价；DetectMediaType/异步快照已直接为该词汇，不再二次转换）。 +// per_char 模型把输出字数映射到 completionTokens（TokenMapping），随该字段传给 shop /calc 计价。 func buildModelUsage(prompt, completion, cached int64, mediaType string, durationSec int64) map[string]any { if durationSec < 0 { durationSec = 0 @@ -96,7 +86,7 @@ func buildModelUsage(prompt, completion, cached int64, mediaType string, duratio "promptTokens": prompt, "completionTokens": completion, "cachedTokens": cached, - "mediaType": mgMediaTypeToShop(mediaType), + "mediaType": mediaType, "durationSec": durationSec, } } diff --git a/service/session_stream.go b/service/session_stream.go index ae12c25..2d733cd 100644 --- a/service/session_stream.go +++ b/service/session_stream.go @@ -130,6 +130,8 @@ LOOP: updateModelSessionReq.DurationSeconds = int64(time.Since(startTime).Seconds()) // 调 shop-user-trade 按用量算费（不本地换算；调用前门禁已保证配置存在，失败→0 不阻塞） mediaType := modelUtils.DetectMediaType(modelInfo.RequestBusinessFieldMapping, newRequestParams) + docMsg.ModelId = modelInfo.Id // 引用行=系统模型 id，供 per_token 结算按系统模型计价 + docMsg.MediaType = mediaType docMsg.Cost = calcModelCost(ctx, modelInfo.Id, buildModelUsage(docMsg.PromptTokens, docMsg.CompletionTokens, 0, mediaType, 0)) updateModelSessionReq.TotalCost = docMsg.Cost @@ -241,6 +243,8 @@ func (s *modelSessionService) CreateSessionStream(ctx context.Context, w http.Re // 流结束：调 shop-user-trade 按用量算费（不本地换算；调用前门禁已保证配置存在，失败→0 不阻塞） mediaType := modelUtils.DetectMediaType(modelInfo.RequestBusinessFieldMapping, newRequestParams) + docMsg.ModelId = modelInfo.Id // 引用行=系统模型 id，供 per_token 结算按系统模型计价 + docMsg.MediaType = mediaType docMsg.Cost = calcModelCost(ctx, modelInfo.Id, buildModelUsage(docMsg.PromptTokens, docMsg.CompletionTokens, 0, mediaType, 0)) diff --git a/service/session_sync.go b/service/session_sync.go index 721be3a..84dc3ab 100644 --- a/service/session_sync.go +++ b/service/session_sync.go @@ -120,6 +120,8 @@ LOOP: updateModelSessionReq.DurationSeconds = int64(time.Since(startTime).Seconds()) // 9.5) 调 shop-user-trade 按用量算费（不本地换算；调用前门禁已保证配置存在，失败→0 不阻塞） mediaType := modelUtils.DetectMediaType(modelInfo.RequestBusinessFieldMapping, newRequestParams) + docMsg.ModelId = modelInfo.Id // 引用行=系统模型 id，供 per_token 结算按系统模型计价 + docMsg.MediaType = mediaType docMsg.Cost = calcModelCost(ctx, modelInfo.Id, buildModelUsage(docMsg.PromptTokens, docMsg.CompletionTokens, 0, mediaType, 0)) updateModelSessionReq.TotalCost = docMsg.Cost diff --git a/service/utils/media_type.go b/service/utils/media_type.go index 57ab5bb..a1ca0b7 100644 --- a/service/utils/media_type.go +++ b/service/utils/media_type.go @@ -1,21 +1,23 @@ package utils -// DetectMediaType 按模型业务字段映射从请求体推导输入媒体类型（替代硬编码的 media.type 路径）： +// DetectMediaType 按模型业务字段映射从请求体推导输入媒体类型（替代硬编码的 media.type 路径）。 +// 直接返回 shop-user-trade 计费词汇（audio/video，对齐 ChargeUsage.MediaType）， +// 无需二次转换（原 mgMediaTypeToShop 已删除）： // - reference_audio 映射路径在请求体中有值 → "audio" -// - reference_video 映射路径在请求体中有值 → "has_video" -// - 否则 → "no_video" +// - reference_video 映射路径在请求体中有值 → "video" +// - 否则 → ""（无媒体引用，shop 侧 pickModelPrice 查不到 mediaPrices 键 → 落默认价） // // 判定完全由模型配置（RequestBusinessFieldMapping，业务字段名见 ChatFieldsReq/VideoFields）驱动， // 无请求结构硬编码；映射路径值即 GetByPathAll 路径（如 input.media?type=audio&url=#）。 -// 媒体类型仅供 shop-user-trade 算费用量（见 service/pricing_client.go mgMediaTypeToShop）。 +// 媒体类型仅供 shop-user-trade 算费用量（见 service/pricing_client.go buildModelUsage）。 func DetectMediaType(reqBizMapping map[string]string, reqParams map[string]any) string { if hasMediaValue(reqBizMapping, reqParams, "reference_audio") { return "audio" } if hasMediaValue(reqBizMapping, reqParams, "reference_video") { - return "has_video" + return "video" } - return "no_video" + return "" } // hasMediaValue 业务字段映射路径在请求体中是否命中值 diff --git a/update.sql b/update.sql index 83de068..b2550d4 100644 --- a/update.sql +++ b/update.sql @@ -312,7 +312,7 @@ COMMENT ON COLUMN model_gateway_session.total_cost ALTER TABLE model_gateway_model_task_start ADD COLUMN IF NOT EXISTS media_type VARCHAR(32) DEFAULT NULL; COMMENT ON COLUMN model_gateway_model_task_start.media_type - IS '输入媒体类型快照（audio/no_video/has_video，创建任务时按请求体参考媒体字段推导）'; + IS '输入媒体类型快照（audio/video，空=无媒体引用；shop 计费词汇，创建任务时按请求体参考媒体字段推导）'; ALTER TABLE model_gateway_model_task_end ADD COLUMN IF NOT EXISTS total_cost NUMERIC DEFAULT 0; @@ -363,14 +363,15 @@ ALTER TABLE model_gateway_model_manage COMMENT ON COLUMN model_gateway_model_manage.error_message_mapping IS '错误消息映射：{code,message} 的 schema 树（type/attrs/value/defaultValue），解析模型错误响应，defaultValue 为成功码'; + -- ========================= -- 错误重试记忆：LLM 分析上游模型错误是否可重试的持久知识库 -- memory_key = SHA-256(upstream|error_code|归一化消息)，唯一；命中直接复用，永久有效 -- ========================= -CREATE TABLE IF NOT EXISTS model_gateway_error_memory ( - id BIGSERIAL PRIMARY KEY, - tenant_id BIGINT DEFAULT 0, - creator VARCHAR(64) DEFAULT '', +CREATE TABLE IF NOT EXISTS model_gateway_model_error_memory ( + id BIGSERIAL PRIMARY KEY, + tenant_id BIGINT DEFAULT 0, + creator VARCHAR(64) DEFAULT '', created_at TIMESTAMPTZ DEFAULT now(), updater VARCHAR(64) DEFAULT '', updated_at TIMESTAMPTZ DEFAULT now(), @@ -382,7 +383,7 @@ CREATE TABLE IF NOT EXISTS model_gateway_error_memory ( retryable BOOLEAN NOT NULL DEFAULT false, reason VARCHAR(512) NOT NULL DEFAULT '', analyzed_by VARCHAR(128) NOT NULL DEFAULT '' -); + ); CREATE UNIQUE INDEX IF NOT EXISTS uk_error_memory_memory_key - ON model_gateway_error_memory (memory_key) + ON model_gateway_model_error_memory (memory_key) WHERE deleted_at IS NULL; \ No newline at end of file