diff --git a/gateway/model.go b/gateway/model.go index e1391c9..a57a74c 100644 --- a/gateway/model.go +++ b/gateway/model.go @@ -132,12 +132,11 @@ func GetModelInfoById(ctx context.Context, req *GetModelInfoByIdReq) (res *GetMo return } -// ModelCallResult 调模型网关生成一段内容并等待结果。 -// 视频等异步模型为"提交+等待"一体:内部订阅 GMQ 直到结果返回。 -func ModelCallResult(ctx context.Context, modelId int64, responseType model.ResponseType, sessionId string, requestParams map[string]any, businessParams map[string]any) (responseParams *ModelCallRes, err error) { +// SubmitModelCall 提交模型调用(异步模型自动生成唯一 msgTopic),返回提交响应与消息主题。 +// taskId 在 res.TaskId;同步模型 msgTopic 为空串。结果等待由 WaitModelCallResult 完成。 +func SubmitModelCall(ctx context.Context, modelId int64, responseType model.ResponseType, sessionId string, requestParams map[string]any, businessParams map[string]any) (res *ModelCallRes, msgTopic string, err error) { // 异步模型必须绑定消息主题接收结果:自动生成唯一主题, // 带业务标识(bizName/modelId/sessionId)便于排查,每次调用唯一避免并发串结果 - msgTopic := "" if responseType != nil && *responseType == *model.ResponseTypeAsync.Code() { msgTopic = modelCallTopic(g.Cfg().MustGet(ctx, "server.name").String(), modelId, sessionId) } @@ -150,62 +149,77 @@ func ModelCallResult(ctx context.Context, modelId int64, responseType model.Resp BusinessParams: businessParams, MsgTopic: msgTopic, } - // 2. 克隆 commonHttp 客户端(保留 Consul 服务发现),显式设置超时和 ResponseHeaderTimeout + // 克隆 commonHttp 客户端(保留 Consul 服务发现),显式设置超时和 ResponseHeaderTimeout client := commonHttp.Httpclient.Clone() client.SetTimeout(30 * time.Minute) if tr, ok := client.Transport.(*http.Transport); ok { tr.ResponseHeaderTimeout = 30 * time.Minute } - res := new(ModelCallRes) + res = new(ModelCallRes) err = commonHttp.Post(ctx, "model-gateway/model/call/modelCall", requestHeaders(ctx), res, &req) if err != nil { - return nil, err + return nil, "", err } if g.IsEmpty(res.TaskId) || !g.IsEmpty(res.ErrorMsg) { - return nil, fmt.Errorf("创建模型任务失败:%v", res.ErrorMsg) + return nil, "", fmt.Errorf("创建模型任务失败:%v", res.ErrorMsg) } - // 3. 订阅模型结果(异步模型) - if responseType != nil && *responseType == *model.ResponseTypeAsync.Code() { - resultCh := make(chan *ModelCallRes, 1) - errCh := make(chan error, 1) + return res, msgTopic, nil +} - _, err = gmq.GetGmq(public.GmqMsgPluginsName).GmqSubscribe(ctx, &mq.NatsSubMessage{ - SubMessage: types.SubMessage{ - Topic: msgTopic, - ConsumerName: fmt.Sprintf("model-call-result-%s", uuid.NewString()), - AutoAck: false, - AutoUnsubscribe: true, - FetchCount: 1, - HandleFunc: func(ctx context.Context, msg any) error { - r := new(ModelCallRes) - if err := gconv.Struct(msg, r); err != nil { - errCh <- err - return nil - } - if g.IsEmpty(r.TaskId) || !g.IsEmpty(r.ErrorMsg) { - errCh <- fmt.Errorf("创建模型任务失败:%v", r.ErrorMsg) - return nil - } - resultCh <- r +// WaitModelCallResult 订阅 msgTopic 等待异步模型结果。 +// 重订阅可拿回已发布结果:gmq NATS 侧 Retention=LimitsPolicy+MaxAge=7天(消息不随 ack 删除、保留 7 天), +// 订阅用 DeliverPolicy=DeliverAllPolicy(新订阅从 stream 序头重放),崩溃后重订阅同一 msgTopic 即可恢复。 +func WaitModelCallResult(ctx context.Context, msgTopic string) (responseParams *ModelCallRes, err error) { + resultCh := make(chan *ModelCallRes, 1) + errCh := make(chan error, 1) + + _, err = gmq.GetGmq(public.GmqMsgPluginsName).GmqSubscribe(ctx, &mq.NatsSubMessage{ + SubMessage: types.SubMessage{ + Topic: msgTopic, + ConsumerName: fmt.Sprintf("model-call-result-%s", uuid.NewString()), + AutoAck: false, + AutoUnsubscribe: true, + FetchCount: 1, + HandleFunc: func(ctx context.Context, msg any) error { + r := new(ModelCallRes) + if err := gconv.Struct(msg, r); err != nil { + errCh <- err return nil - }, + } + if g.IsEmpty(r.TaskId) || !g.IsEmpty(r.ErrorMsg) { + errCh <- fmt.Errorf("创建模型任务失败:%v", r.ErrorMsg) + return nil + } + resultCh <- r + return nil }, - Durable: true, - }) - if err != nil { - return - } - - select { - case responseParams = <-resultCh: - case err = <-errCh: - return nil, err - case <-ctx.Done(): - return nil, ctx.Err() - } + }, + Durable: true, + }) + if err != nil { return } + select { + case responseParams = <-resultCh: + case err = <-errCh: + return nil, err + case <-ctx.Done(): + return nil, ctx.Err() + } + return +} + +// ModelCallResult 调模型网关生成一段内容并等待结果(内部复用 SubmitModelCall + WaitModelCallResult)。 +// 同步模型提交即返回;异步模型为"提交+等待"一体。 +func ModelCallResult(ctx context.Context, modelId int64, responseType model.ResponseType, sessionId string, requestParams map[string]any, businessParams map[string]any) (responseParams *ModelCallRes, err error) { + res, msgTopic, err := SubmitModelCall(ctx, modelId, responseType, sessionId, requestParams, businessParams) + if err != nil { + return nil, err + } + if responseType != nil && *responseType == *model.ResponseTypeAsync.Code() { + return WaitModelCallResult(ctx, msgTopic) + } return res, nil } diff --git a/gateway/model_split_test.go b/gateway/model_split_test.go new file mode 100644 index 0000000..6cca231 --- /dev/null +++ b/gateway/model_split_test.go @@ -0,0 +1,15 @@ +package gateway + +import ( + "context" + "testing" + + "ai-agent/workflow/consts/model" +) + +// TestSplitSignatures 只做编译期签名验证:拆分后的函数存在且签名符合 memo 包装契约。 +// 不做实际调用(依赖 live model-gateway + Consul),实际行为由端到端走查覆盖。 +func TestSplitSignatures(t *testing.T) { + var _ func(context.Context, int64, model.ResponseType, string, map[string]any, map[string]any) (*ModelCallRes, string, error) = SubmitModelCall + var _ func(context.Context, string) (*ModelCallRes, error) = WaitModelCallResult +}