Job Management

Job Management 是管理后台异步任务的模块，包括任务创建、状态查询、取消、重试和结果获取。在 AI Agent 中用于 Embedding 生成、知识库索引、批量评估等。

#type / concept #status / evergreen #tech / backend #tech / architecture

[!info] related notes

所属 MOC: AI Agent Application MOC

相关: Async Task Queue

Job Management

一句话定义

Job Management 是管理后台异步任务的模块。用户触发一个操作（如”索引这个文档”），系统创建一个 Job，用户可以查询进度、取消任务、获取结果。

核心原理

Job 生命周期

pending → running → completed
                  → failed
                  → cancelled

API 设计

// 创建 Job
POST /api/jobs
Body: { "type": "document_indexing", "params": { "doc_id": "xxx" } }
Response: { "job_id": "job_xxx", "status": "pending" }

// 查询 Job 状态
GET /api/jobs/:id
Response: { "job_id": "job_xxx", "status": "running", "progress": 60 }

// 取消 Job
DELETE /api/jobs/:id
Response: { "job_id": "job_xxx", "status": "cancelled" }

// 获取 Job 结果
GET /api/jobs/:id/result
Response: { "result": { "indexed_chunks": 42 } }

Go 实现

type Job struct {
    ID        string    `json:"id"`
    Type      string    `json:"type"`
    Status    string    `json:"status"`
    Progress  int       `json:"progress"`
    Params    any       `json:"params"`
    Result    any       `json:"result,omitempty"`
    Error     string    `json:"error,omitempty"`
    CreatedAt time.Time `json:"created_at"`
    UpdatedAt time.Time `json:"updated_at"`
}

type JobService struct {
    repo  JobRepository
    queue TaskQueue
}

func (s *JobService) Create(ctx context.Context, jobType string, params any) (*Job, error) {
    job := &Job{
        ID:        generateID(),
        Type:      jobType,
        Status:    "pending",
        Params:    params,
        CreatedAt: time.Now(),
    }
    s.repo.Save(ctx, job)
    s.queue.Enqueue(ctx, job)
    return job, nil
}

常见坑

不做进度更新: 用户不知道任务跑到哪了
不做取消支持: 长时间任务无法取消
结果不持久化: 任务完成后结果被清理
不做超时: 任务卡住不结束

参考资料

Async Task Queue