red-future/model-gateway
4
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor(gateway): 重构任务回调和数据结构

Browse Source
This commit is contained in:
WangLiZhao
2026-06-22 15:08:56 +08:00
parent 4cc44bf57c
commit 3ccce74465
7 changed files with 81 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controller/model_gateway_task_controller.go
+5
View File
@@ -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)
model/dto/model_gateway_task_dto.go
+14 -4
View File
@@ -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 任务列表分页查询
model/entity/model_gateway_logs_op.go
+13 -13
View File
@@ -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"`
}
model/entity/model_gateway_task.go
+12 -27
View File
@@ -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"`
}
service/gateway/gateway_http_service.go
+14 -8
View File
@@ -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字节",
service/task/task_service.go
+16 -14
View File
@@ -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
service/task/worker.go
+7 -9
View File
@@ -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)