fix: 修复管线context取消、添加FIFO任务队列、移除重复提示词优化
三个修复： 1. runPipelineBg 改用 context.Background()，避免 HTTP 响应返回后 Gin 取消 request context 导致后台管线静默失败 2. 新增 TaskQueue FIFO 串行队列，任务提交后进入 pending 状态排队， 按提交顺序逐个执行，前端轮询显示排队中 3. promptOptimizerNode 移除 RunPromptAgent 调用，提示词优化仅由 前端在提交前通过 /api/v1/prompt/optimize 执行一次，管线内只做 风格合并和技术参数追加
This commit is contained in:
@@ -44,6 +44,11 @@ func main() {
|
||||
// 初始化项目服务
|
||||
service.InitProjectService(db.GetDB())
|
||||
|
||||
// 初始化生成任务队列（FIFO 串行执行）
|
||||
generateQueue := service.NewTaskQueue()
|
||||
handler.SetGenerateQueue(generateQueue)
|
||||
defer generateQueue.Stop()
|
||||
|
||||
r := gin.New()
|
||||
r.Use(mildware.Logger())
|
||||
r.Use(mildware.Recovery())
|
||||
|
||||
@@ -16,6 +16,14 @@ import (
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// generateQueue 全局生成任务队列，由 main 通过 SetGenerateQueue 注入。
|
||||
var generateQueue *service.TaskQueue
|
||||
|
||||
// SetGenerateQueue 设置生成任务队列。
|
||||
func SetGenerateQueue(q *service.TaskQueue) {
|
||||
generateQueue = q
|
||||
}
|
||||
|
||||
// GenerateRequest 素材生成请求。
|
||||
type GenerateRequest struct {
|
||||
ProjectID string `json:"projectId"`
|
||||
@@ -43,7 +51,7 @@ type AssetsResponse struct {
|
||||
}
|
||||
|
||||
// Generate 素材生成接口（异步）。
|
||||
// 立即返回 taskId，后台执行管线，前端通过 GET /tasks/:taskId 轮询进度。
|
||||
// 立即返回 taskId，任务进入 FIFO 队列串行执行，前端通过 GET /tasks/:taskId 轮询进度。
|
||||
func Generate(c *gin.Context) {
|
||||
var req GenerateRequest
|
||||
if err := c.ShouldBindJSON(&req); err != nil {
|
||||
@@ -57,18 +65,29 @@ func Generate(c *gin.Context) {
|
||||
}
|
||||
taskID := fmt.Sprintf("task-%d", time.Now().UnixMilli())
|
||||
|
||||
// 保存任务到数据库
|
||||
// 保存任务到数据库，初始状态为 pending
|
||||
if err := saveTaskToDB(c.Request.Context(), projectID, taskID, req); err != nil {
|
||||
logger.FromCtx(c.Request.Context()).Error("failed to save task", "error", err)
|
||||
c.JSON(http.StatusInternalServerError, model.Fail(http.StatusInternalServerError, "创建任务失败"))
|
||||
return
|
||||
}
|
||||
|
||||
// 返回 taskId
|
||||
c.JSON(http.StatusOK, model.OK(GenerateResponse{TaskID: taskID}))
|
||||
// 加入 FIFO 队列
|
||||
queuePos := 1
|
||||
if generateQueue != nil {
|
||||
queuePos = generateQueue.Enqueue(&service.TaskJob{
|
||||
Ctx: context.Background(),
|
||||
ProjectID: projectID,
|
||||
TaskID: taskID,
|
||||
Execute: func(ctx context.Context) error {
|
||||
runPipelineBg(ctx, projectID, taskID, req)
|
||||
return nil
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
// 后台执行管线
|
||||
go runPipelineBg(c.Request.Context(), projectID, taskID, req)
|
||||
c.JSON(http.StatusOK, model.OK(GenerateResponse{TaskID: taskID}))
|
||||
_ = queuePos
|
||||
}
|
||||
|
||||
// runPipelineBg 后台执行生成管线，更新任务状态。
|
||||
@@ -82,6 +101,7 @@ func runPipelineBg(ctx context.Context, projectID, taskID string, req GenerateRe
|
||||
updateTaskInDB(ctx, taskID, "running", stage, "", progress)
|
||||
})
|
||||
|
||||
// 队列调度后才标记为 running，初始写入时是 pending
|
||||
updateTaskInDB(ctx, taskID, "running", "prompt_builder", "", 5)
|
||||
|
||||
in := service.PipelineInput{
|
||||
@@ -240,9 +260,9 @@ func saveTaskToDB(ctx context.Context, projectID, taskID string, req GenerateReq
|
||||
ProjectID: uint(projectIDUint),
|
||||
Prompt: req.Prompt,
|
||||
AssetType: req.AssetType,
|
||||
Status: "running",
|
||||
Progress: 5,
|
||||
Stage: "prompt_builder",
|
||||
Status: "pending",
|
||||
Progress: 0,
|
||||
Stage: "",
|
||||
RetryCount: 0,
|
||||
CreatedAt: time.Now(),
|
||||
UpdatedAt: time.Now(),
|
||||
|
||||
@@ -14,10 +14,10 @@ import (
|
||||
"github.com/cloudwego/eino/compose"
|
||||
)
|
||||
|
||||
// promptOptimizerNode 节点：调用 PromptAgent 生成规范提示词，合并风格与重试信息。
|
||||
// 输入 PipelineInput，输出最终提示词字符串（直接供 AssetGenerator 消费）。
|
||||
// promptOptimizerNode 节点：合并风格描述、重试信息与技术参数，输出最终提示词。
|
||||
// 提示词优化已由前端在提交前完成，管线内不再重复调用 PromptAgent。
|
||||
var promptOptimizerNode = compose.InvokableLambda(func(ctx context.Context, in PipelineInput) (string, error) {
|
||||
// 合并风格描述，注入原始 Prompt 中
|
||||
// 合并风格描述
|
||||
styleDesc := buildStyleDescription(in.ProjectStyle, in.TaskStyle)
|
||||
if styleDesc != "" {
|
||||
if in.Prompt != "" {
|
||||
@@ -36,26 +36,11 @@ var promptOptimizerNode = compose.InvokableLambda(func(ctx context.Context, in P
|
||||
}
|
||||
}
|
||||
|
||||
if len(in.Tags) == 0 && in.Prompt == "" {
|
||||
return "", fmt.Errorf("pipeline: Prompt and Tags are both empty")
|
||||
if in.Prompt == "" {
|
||||
return "", fmt.Errorf("pipeline: prompt is empty")
|
||||
}
|
||||
|
||||
// 有标签时调用 PromptAgent 优化提示词
|
||||
if len(in.Tags) > 0 {
|
||||
agentIn := PromptAgentInput{
|
||||
Tags: in.Tags,
|
||||
AssetType: in.AssetType,
|
||||
Prompt: in.Prompt,
|
||||
UserNote: in.UserNote,
|
||||
}
|
||||
output, err := RunPromptAgent(ctx, agentIn)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("prompt agent: %w", err)
|
||||
}
|
||||
return output.Prompt, nil
|
||||
}
|
||||
|
||||
// 无标签时直接使用原始 Prompt，补上技术参数段
|
||||
// 追加技术参数段，不再调用 PromptAgent
|
||||
return appendTechNotes(in.Prompt, in.AssetType, in.Params), nil
|
||||
})
|
||||
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
"gen2d/internal/logger"
|
||||
)
|
||||
|
||||
// TaskJob 队列中的任务。
|
||||
type TaskJob struct {
|
||||
Ctx context.Context
|
||||
ProjectID string
|
||||
TaskID string
|
||||
Execute func(ctx context.Context) error
|
||||
}
|
||||
|
||||
// TaskQueue 串行 FIFO 任务队列。
|
||||
type TaskQueue struct {
|
||||
mu sync.Mutex
|
||||
jobs []*TaskJob
|
||||
ready chan struct{}
|
||||
stop chan struct{}
|
||||
stopped bool
|
||||
}
|
||||
|
||||
// NewTaskQueue 创建任务队列并启动调度协程。
|
||||
func NewTaskQueue() *TaskQueue {
|
||||
q := &TaskQueue{
|
||||
jobs: make([]*TaskJob, 0),
|
||||
ready: make(chan struct{}, 1),
|
||||
stop: make(chan struct{}),
|
||||
}
|
||||
go q.run()
|
||||
return q
|
||||
}
|
||||
|
||||
// Enqueue 将任务加入队尾，返回队列中的位置（1-based）。
|
||||
func (q *TaskQueue) Enqueue(job *TaskJob) int {
|
||||
q.mu.Lock()
|
||||
defer q.mu.Unlock()
|
||||
q.jobs = append(q.jobs, job)
|
||||
pos := len(q.jobs)
|
||||
select {
|
||||
case q.ready <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
return pos
|
||||
}
|
||||
|
||||
// QueueLen 返回当前队列长度。
|
||||
func (q *TaskQueue) QueueLen() int {
|
||||
q.mu.Lock()
|
||||
defer q.mu.Unlock()
|
||||
return len(q.jobs)
|
||||
}
|
||||
|
||||
// Stop 优雅关闭队列。
|
||||
func (q *TaskQueue) Stop() {
|
||||
q.mu.Lock()
|
||||
defer q.mu.Unlock()
|
||||
if !q.stopped {
|
||||
q.stopped = true
|
||||
close(q.stop)
|
||||
}
|
||||
}
|
||||
|
||||
func (q *TaskQueue) run() {
|
||||
for {
|
||||
select {
|
||||
case <-q.stop:
|
||||
return
|
||||
case <-q.ready:
|
||||
q.processNext()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (q *TaskQueue) processNext() {
|
||||
q.mu.Lock()
|
||||
if len(q.jobs) == 0 {
|
||||
q.mu.Unlock()
|
||||
return
|
||||
}
|
||||
job := q.jobs[0]
|
||||
q.jobs = q.jobs[1:]
|
||||
// 如果队列还有任务，重新发信号
|
||||
if len(q.jobs) > 0 {
|
||||
select {
|
||||
case q.ready <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
q.mu.Unlock()
|
||||
|
||||
l := logger.With("task_id", job.TaskID, "project_id", job.ProjectID)
|
||||
l.Info("task queue executing job", "queue_remaining", len(q.jobs))
|
||||
|
||||
if err := job.Execute(job.Ctx); err != nil {
|
||||
l.Error("task job failed", "error", err)
|
||||
} else {
|
||||
l.Info("task job completed")
|
||||
}
|
||||
}
|
||||
@@ -120,7 +120,7 @@ export const useGenerationStore = create<GenerationState>((set) => ({
|
||||
|
||||
function stageLabel(stage?: string): string {
|
||||
switch (stage) {
|
||||
case 'prompt_builder': return '优化提示词...'
|
||||
case 'prompt_builder': return '构建提示词...'
|
||||
case 'asset_generator': return '生成素材中...'
|
||||
case 'quality_supervisor': return '质检中...'
|
||||
case 'format_adapter': return '格式转换中...'
|
||||
|
||||
Reference in New Issue
Block a user