From 3ccce7446504440434e0e4c7c795c3883837d7c6 Mon Sep 17 00:00:00 2001 From: WangLiZhao <1838393649@qq.com> Date: Mon, 22 Jun 2026 15:08:56 +0800 Subject: [PATCH] =?UTF-8?q?refactor(gateway):=20=E9=87=8D=E6=9E=84?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E5=9B=9E=E8=B0=83=E5=92=8C=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E7=BB=93=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- controller/model_gateway_task_controller.go | 5 +++ model/dto/model_gateway_task_dto.go | 18 +++++++--- model/entity/model_gateway_logs_op.go | 26 +++++++------- model/entity/model_gateway_task.go | 39 +++++++-------------- service/gateway/gateway_http_service.go | 22 +++++++----- service/task/task_service.go | 30 ++++++++-------- service/task/worker.go | 16 ++++----- 7 files changed, 81 insertions(+), 75 deletions(-) diff --git a/controller/model_gateway_task_controller.go b/controller/model_gateway_task_controller.go index 1ca0b0c..909f67b 100644 --- a/controller/model_gateway_task_controller.go +++ b/controller/model_gateway_task_controller.go @@ -17,6 +17,11 @@ func (c *task) CreateTask(ctx context.Context, req *dto.CreateTaskReq) (res *dto return taskService.ModelGatewayTask.Create(ctx, req) } +// JobTask 定时任务：循环执行待处理任务 +func (c *task) JobTask(ctx context.Context, req *dto.JobTaskReq) (res *dto.JobTaskRes, err error) { + return taskService.ModelGatewayTask.JobTask(ctx, req) +} + // GetTaskResult 获取单条任务结果（返回 *dto.GetTaskResultRes） func (c *task) GetTaskResult(ctx context.Context, req *dto.GetTaskResultReq) (res *dto.GetTaskResultRes, err error) { return taskService.ModelGatewayTask.GetResult(ctx, req.TaskID) diff --git a/model/dto/model_gateway_task_dto.go b/model/dto/model_gateway_task_dto.go index 8d7b0a1..e2e779a 100644 --- a/model/dto/model_gateway_task_dto.go +++ b/model/dto/model_gateway_task_dto.go @@ -18,6 +18,17 @@ type CreateTaskReq struct { type CreateTaskRes struct { TaskID string `json:"taskId" dc:"任务ID"` } +type JobTaskReq struct { + g.Meta `path:"/jobTask" method:"post" tags:"任务管理" summary:"定时任务" dc:"循环执行待处理任务，按间隔时间和批次大小处理"` + Interval int `json:"interval" dc:"循环间隔(秒)"` + BatchSize int `json:"batchSize" dc:"每批执行条数"` +} + +type JobTaskRes struct { + TotalProcessed int `json:"totalProcessed" dc:"总处理数"` + SuccessCount int `json:"successCount" dc:"成功数"` + FailCount int `json:"failCount" dc:"失败数"` +} type ModelTaskCallbackReq struct { g.Meta `path:"/modelCallback" method:"post" tags:"异步任务" summary:"模型任务回调通知"` @@ -73,10 +84,9 @@ type GetTaskBatchRes struct { } type GetTaskBatchItem struct { - TaskID string `json:"taskId" dc:"任务ID"` - State int `json:"state" dc:"任务状态"` - OssFile string `json:"ossFile" dc:"结果文件OSS地址"` - TextResult map[string]any `json:"textResult" dc:"文本结果"` + TaskID string `json:"taskId" dc:"任务ID"` + State int `json:"state" dc:"任务状态"` + OssFile string `json:"ossFile" dc:"结果文件OSS地址"` } // ListTaskReq 任务列表分页查询 diff --git a/model/entity/model_gateway_logs_op.go b/model/entity/model_gateway_logs_op.go index 1cf8a2d..9d3b3df 100644 --- a/model/entity/model_gateway_logs_op.go +++ b/model/entity/model_gateway_logs_op.go @@ -40,17 +40,17 @@ var ModelGatewayLogsOpCol = modelGatewayLogsOpCol{ // ModelGatewayLogsOp 操作日志 type ModelGatewayLogsOp struct { beans.SQLBaseDO `orm:",inline"` - IP string `orm:"ip" json:"ip"` - UserAgent string `orm:"user_agent" json:"userAgent"` - APIPath string `orm:"api_path" json:"apiPath"` - HttpMethod string `orm:"http_method" json:"httpMethod"` - BizName string `orm:"biz_name" json:"bizName"` - ModelName string `orm:"model_name" json:"modelName"` - TaskID string `orm:"task_id" json:"taskId"` - OpType string `orm:"op_type" json:"opType"` - Success int `orm:"success" json:"success"` - ErrorMsg string `orm:"error_msg" json:"errorMsg"` - CostMs int64 `orm:"cost_ms" json:"costMs"` - RequestPayload *RequestPayload `orm:"request_payload" json:"requestPayload"` - ResponsePayload map[string]any `orm:"response_payload" json:"responsePayload"` + IP string `orm:"ip" json:"ip"` + UserAgent string `orm:"user_agent" json:"userAgent"` + APIPath string `orm:"api_path" json:"apiPath"` + HttpMethod string `orm:"http_method" json:"httpMethod"` + BizName string `orm:"biz_name" json:"bizName"` + ModelName string `orm:"model_name" json:"modelName"` + TaskID string `orm:"task_id" json:"taskId"` + OpType string `orm:"op_type" json:"opType"` + Success int `orm:"success" json:"success"` + ErrorMsg string `orm:"error_msg" json:"errorMsg"` + CostMs int64 `orm:"cost_ms" json:"costMs"` + RequestPayload map[string]any `orm:"request_payload" json:"requestPayload"` + ResponsePayload map[string]any `orm:"response_payload" json:"responsePayload"` } diff --git a/model/entity/model_gateway_task.go b/model/entity/model_gateway_task.go index baaa13e..986839a 100644 --- a/model/entity/model_gateway_task.go +++ b/model/entity/model_gateway_task.go @@ -11,14 +11,11 @@ type modelGatewayTaskCol struct { BizName string CallbackURL string State string - Phase string ErrorMsg string ResultFile string - TextResult string ExpendTokens string DurationSeconds string RetryCount string - TmpFile string RequestPayload string EpicycleId string } @@ -30,14 +27,11 @@ var ModelGatewayTaskCol = modelGatewayTaskCol{ BizName: "biz_name", CallbackURL: "callback_url", State: "state", - Phase: "phase", ErrorMsg: "error_msg", ResultFile: "result_file", - TextResult: "text_result", ExpendTokens: "expend_tokens", DurationSeconds: "duration_seconds", RetryCount: "retry_count", - TmpFile: "tmp_file", RequestPayload: "request_payload", EpicycleId: "epicycle_id", } @@ -45,21 +39,18 @@ var ModelGatewayTaskCol = modelGatewayTaskCol{ // ModelGatewayTask 模型网关任务 type ModelGatewayTask struct { beans.SQLBaseDO `orm:",inline"` - ModelName string `orm:"model_name" json:"modelName"` - TaskID string `orm:"task_id" json:"taskId"` - BizName string `orm:"biz_name" json:"bizName"` - CallbackURL string `orm:"callback_url" json:"callbackUrl"` - State int `orm:"state" json:"state"` - Phase int `orm:"phase" json:"phase"` - ErrorMsg string `orm:"error_msg" json:"errorMsg"` - ResultFile *ResultFile `orm:"result_file" json:"resultFile"` - TextResult map[string]any `orm:"text_result" json:"text"` - ExpendTokens int64 `orm:"expend_tokens" json:"expendTokens"` - DurationSeconds int64 `orm:"duration_seconds" json:"durationSeconds"` - RetryCount int `orm:"retry_count" json:"retryCount"` - TmpFile string `orm:"tmp_file" json:"tmpFile"` - RequestPayload *RequestPayload `orm:"request_payload" json:"requestPayload"` - EpicycleId int64 `orm:"epicycle_id" json:"epicycleId"` + ModelName string `orm:"model_name" json:"modelName"` + TaskID string `orm:"task_id" json:"taskId"` + BizName string `orm:"biz_name" json:"bizName"` + CallbackURL string `orm:"callback_url" json:"callbackUrl"` + State int `orm:"state" json:"state"` + ErrorMsg string `orm:"error_msg" json:"errorMsg"` + ResultFile *ResultFile `orm:"result_file" json:"resultFile"` + ExpendTokens int64 `orm:"expend_tokens" json:"expendTokens"` + DurationSeconds int64 `orm:"duration_seconds" json:"durationSeconds"` + RetryCount int `orm:"retry_count" json:"retryCount"` + RequestPayload map[string]any `orm:"request_payload" json:"requestPayload"` + EpicycleId int64 `orm:"epicycle_id" json:"epicycleId"` } // ResultFile OSS 结果文件 @@ -68,9 +59,3 @@ type ResultFile struct { FileType string `json:"fileType"` FileSize int64 `json:"fileSize"` } - -// RequestPayload 请求参数结构体 -type RequestPayload struct { - Headers map[string]string `json:"headers"` - Body map[string]any `json:"body"` -} diff --git a/service/gateway/gateway_http_service.go b/service/gateway/gateway_http_service.go index 3782545..96d7eff 100644 --- a/service/gateway/gateway_http_service.go +++ b/service/gateway/gateway_http_service.go @@ -86,7 +86,6 @@ type CallbackPayload struct { // TriggerCallback 任务的回调 func TriggerCallback(ctx context.Context, t *entity.ModelGatewayTask) { - //headers := util.ForwardHeaders(ctx) headers := make(map[string]string) if r := g.RequestFromCtx(ctx); r != nil { for k, v := range r.Request.Header { @@ -123,14 +122,17 @@ func TriggerCallback(ctx context.Context, t *entity.ModelGatewayTask) { // PromptsCallbackPayload 提示词回调请求体 type PromptsCallbackPayload struct { - EpicycleId int64 `json:"epicycleId"` - Messages map[string]any `json:"messages"` + EpicycleId int64 `json:"epicycleId"` + TaskId string `json:"task_id"` + State int `json:"state"` + ErrorMsg string `json:"error_msg"` + OssFile string `json:"oss_file"` + FileType string `json:"file_type"` } // TriggerPromptsCallback 任务成功后的提示词回调 -func TriggerPromptsCallback(ctx context.Context, t *entity.ModelGatewayTask, epicycleId int64) { +func TriggerPromptsCallback(ctx context.Context, t *entity.ModelGatewayTask) { callbackURL := "prompts-core/session/callback" - //headers := util.ForwardHeaders(ctx) headers := make(map[string]string) if r := g.RequestFromCtx(ctx); r != nil { for k, v := range r.Request.Header { @@ -141,12 +143,16 @@ func TriggerPromptsCallback(ctx context.Context, t *entity.ModelGatewayTask, epi } var resp struct{} payload := PromptsCallbackPayload{ - EpicycleId: epicycleId, - Messages: t.TextResult, + EpicycleId: t.EpicycleId, + TaskId: t.TaskID, + State: t.State, + ErrorMsg: t.ErrorMsg, + OssFile: t.ResultFile.OssFile, + FileType: t.ResultFile.FileType, } jsonData, err := json.Marshal(payload) if err != nil { - g.Log().Warningf(ctx, "[提示词回调] JSON序列化失败 epicycleId=%d 错误=%v", epicycleId, err) + g.Log().Warningf(ctx, "[提示词回调] JSON序列化失败 epicycleId=%d 错误=%v", t.EpicycleId, err) return } g.Log().Infof(ctx, "[提示词回调] 开始发送 epicycleId=%d 回调地址=%s 请求头数量=%d 消息体大小=%d字节", diff --git a/service/task/task_service.go b/service/task/task_service.go index f8905b0..1a628e6 100644 --- a/service/task/task_service.go +++ b/service/task/task_service.go @@ -64,16 +64,13 @@ func (s *taskService) Create(ctx context.Context, req *dto.CreateTaskReq) (res * // 3) 构建任务实体 task := &entity.ModelGatewayTask{ - ModelName: model.ModelName, - TaskID: taskID, - State: public.TaskStatusRunning, - BizName: req.BizName, - CallbackURL: req.CallbackUrl, - RequestPayload: &entity.RequestPayload{ - Body: req.RequestPayload, - Headers: util.ParseHeadMsgHeaders(model.HeadMsg), - }, - EpicycleId: req.EpicycleId, + ModelName: model.ModelName, + TaskID: taskID, + State: public.TaskStatusRunning, + BizName: req.BizName, + CallbackURL: req.CallbackUrl, + RequestPayload: req.RequestPayload, + EpicycleId: req.EpicycleId, } // 4) 插入任务记录 @@ -112,6 +109,12 @@ func (s *taskService) Create(ctx context.Context, req *dto.CreateTaskReq) (res * return &dto.CreateTaskRes{TaskID: taskID}, nil } +// JobTask 定时任务：循环执行待处理任务 +func (s *taskService) JobTask(ctx context.Context, req *dto.JobTaskReq) (res *dto.JobTaskRes, err error) { + + return nil, err +} + // GetResult 获取任务结果 func (s *taskService) GetResult(ctx context.Context, taskID string) (res *dto.GetTaskResultRes, err error) { t, err := dao.ModelGatewayTask.Get(ctx, &entity.ModelGatewayTask{ @@ -161,10 +164,9 @@ func (s *taskService) GetBatch(ctx context.Context, req *dto.GetTaskBatchReq) (r continue } items = append(items, dto.GetTaskBatchItem{ - TaskID: t.TaskID, - State: t.State, - OssFile: t.ResultFile.OssFile, - TextResult: t.TextResult, + TaskID: t.TaskID, + State: t.State, + OssFile: t.ResultFile.OssFile, }) } return &dto.GetTaskBatchRes{List: items}, nil diff --git a/service/task/worker.go b/service/task/worker.go index 20dd537..f25bfab 100644 --- a/service/task/worker.go +++ b/service/task/worker.go @@ -6,6 +6,7 @@ import ( "encoding/json" "fmt" "io" + "model-gateway/model/dto" "net/http" "strings" "sync" @@ -15,7 +16,6 @@ import ( "model-gateway/common/util" "model-gateway/consts/public" "model-gateway/dao" - "model-gateway/model/dto" "model-gateway/model/entity" "model-gateway/service/gateway" @@ -33,7 +33,7 @@ type asyncWorker struct { // handleOne 执行一次完整的任务 func (w *asyncWorker) handleOne(ctx context.Context, task *entity.ModelGatewayTask, model *entity.ModelGatewayModel, req *dto.CreateTaskReq) { var ( - body = task.RequestPayload.Body + body = task.RequestPayload maxRetry = model.RetryTimes startTime = time.Now() rawBytes []byte @@ -83,9 +83,8 @@ func (w *asyncWorker) handleOne(ctx context.Context, task *entity.ModelGatewayTa // ============================================ // 2) 解析校验 + 响应映射（可重试） // ============================================ - result, err = w.parseAndRetry(ctx, result, task, model, req, maxRetry, startTime) + result, err = w.parseAndRetry(ctx, result, task, model, maxRetry, req) if err != nil { - task.TextResult = result w.failTask(ctx, task, startTime, err.Error()) return } @@ -119,7 +118,6 @@ func (w *asyncWorker) handleOne(ctx context.Context, task *entity.ModelGatewayTa FileType: oss.FileFormat, FileSize: int64(oss.FileSize), } - task.TextResult = result if _, err = dao.ModelGatewayTask.Update(ctx, task); err != nil { g.Log().Errorf(ctx, "[handleOne] 更新DB失败 taskId=%s err=%v", task.TaskID, err) @@ -127,8 +125,8 @@ func (w *asyncWorker) handleOne(ctx context.Context, task *entity.ModelGatewayTa } go gateway.TriggerCallback(util.AsyncCtx(ctx), task) - if req.EpicycleId != 0 { - go gateway.TriggerPromptsCallback(util.AsyncCtx(ctx), task, req.EpicycleId) + if task.EpicycleId != 0 { + go gateway.TriggerPromptsCallback(util.AsyncCtx(ctx), task) } g.Log().Infof(ctx, "[handleOne] 成功 taskId=%s duration=%ds fileType=%s", @@ -208,7 +206,7 @@ func (w *asyncWorker) callModel(ctx context.Context, task *entity.ModelGatewayTa } // parseAndRetry 解析模型返回结果，并重试 -func (w *asyncWorker) parseAndRetry(ctx context.Context, body map[string]any, task *entity.ModelGatewayTask, model *entity.ModelGatewayModel, req *dto.CreateTaskReq, maxRetry int, startTime time.Time) (map[string]any, error) { +func (w *asyncWorker) parseAndRetry(ctx context.Context, body map[string]any, task *entity.ModelGatewayTask, model *entity.ModelGatewayModel, maxRetry int, req *dto.CreateTaskReq) (map[string]any, error) { var lastErr error for attempt := 0; attempt <= maxRetry; attempt++ { if attempt > 0 { @@ -259,7 +257,7 @@ func (w *asyncWorker) parseAndRetry(ctx context.Context, body map[string]any, ta task.RetryCount++ _, _ = dao.ModelGatewayTask.Update(ctx, task) - body = injectErrorMessage(task.RequestPayload.Body, lastErr) + body = injectErrorMessage(task.RequestPayload, lastErr) rawData, callErr := InvokeModel(ctx, model, body) if callErr != nil { g.Log().Warningf(ctx, "[执行任务][重调模型失败] taskId=%s attempt=%d/%d err=%v", task.TaskID, attempt, maxRetry, callErr)