09 - 任务编排：下发、执行与产物交付

一句话：所有入口都收敛成一次 console 入站，task_id 由下发方生成做幂等、session_id 决定上下文边界，同一会话串行执行以避免上下文互相污染；任务状态外置到控制面注册表，实例暂存时下发退化为 waking 而非失败，代价是全链路依赖 SSE 长连接与心跳保活，产物还要从本地工作区二次取回、上传对象存储再按平台格式回发。

一、设计目标

员工通过 Web 控制台、企业 IM、定时任务或上层系统向数字分身下发任务。系统需要保证：

任务可追踪：每个任务有唯一 ID，全链路可查
执行可观测：任务执行过程中的 SSE 事件流实时回传
产物可交付：执行完成后产物自动回传到原会话（IM 或 Web）
异常可恢复：实例暂存时任务触发唤醒而非失败

二、任务下发链路

2.1 入站规范

所有任务统一通过 POST /api/console/chat 入站，请求体遵循 AgentRequest 规范：

json

{
  "channel": "console",
  "user_id": "zhangsan",
  "session_id": "user:zhangsan:default",
  "task_id": "task_abc123",
  "input": [
    {"type": "text", "content": "帮我分析这份 CSV 数据，生成可视化报告"}
  ],
  "request_context": {
    "approval_level": "medium",
    "source": "feishu",
    "source_chat_id": "oc_xxx",
    "trace_id": "550e8400-e29b-41d4-a716-446655440000",
    "metadata": {
      "priority": "normal",
      "timeout_seconds": 300
    }
  }
}

字段	必填	说明
`channel`	是	入站渠道标识，统一为 `"console"`
`user_id`	是	员工标识，网关从 SSO 提取
`session_id`	是	会话标识。新值创建新会话，已有值续接上下文
`task_id`	是	任务唯一标识（幂等键），由下发方生成
`input`	是	指令内容（文本/多模态分段）
`request_context`	否	透传上下文（审批策略、来源、trace_id）

2.2 多入口下发

2.3 会话路由规则

入口	session_id 生成规则	上下文策略
Web 控制台	`user:{user_id}:default`	与 IM 单聊共享上下文（默认）
IM 单聊	`user:{user_id}:default`	与 Web 共享上下文（默认）
IM 群聊	`group:{platform}:{chat_id}`	群内独立上下文
定时任务	`cron:{task_name}:{schedule}`	每次执行独立上下文（或续接）
上层系统	自定义（由调用方指定）	由调用方管理

员工可在 Web 偏好设置中开启端隔离模式，此时 Web 和 IM 使用独立 session_id，上下文互不串扰。默认关闭。

三、SSE 事件流

3.1 事件类型

实例执行任务时，通过 SSE 返回实时事件流：

event: task_accepted
data: {"task_id":"task_abc123","status":"accepted","estimated_seconds":60}

event: thinking
data: {"task_id":"task_abc123","content":"正在分析 CSV 数据结构..."}

event: tool_call
data: {"task_id":"task_abc123","tool":"data_analysis","params":{"file":"data.csv"}}

event: delta
data: {"task_id":"task_abc123","content":"根据分析，销售额环比增长","index":0}

event: delta
data: {"task_id":"task_abc123","content":"15.3%，主要增长来自...","index":1}

event: approval_required
data: {"task_id":"task_abc123","request_id":"req_xyz789","tool":"send_email","risk_level":"high"}

event: artifact
data: {"task_id":"task_abc123","path":"projects/task_abc123/report.html","size":15234,"type":"text/html"}

event: task_complete
data: {"task_id":"task_abc123","status":"completed","artifacts":[{"path":"report.html","size":15234},{"path":"chart.png","size":89012}],"total_tokens":3500}

event: task_error
data: {"task_id":"task_abc123","error":"timeout","message":"任务执行超时（300s）"}

3.2 网关 SSE 转发

网关对 SSE 事件流做 chunked 重编码转发，保证浏览器打字机效果：

实例 → 网关：SSE 事件流（text/event-stream）
网关处理：
  1. 设置 Transfer-Encoding: chunked
  2. 设置 Connection: keep-alive
  3. 逐块转发（64KB buffer）
  4. 注入心跳（每 30s 发送 ":ping\n\n" 防假活）
  5. 记录审计事件（每个 event 的 type + size）
网关 → 客户端：chunked SSE 事件流

四、产物交付

4.1 产物类型与交付方式

产物类型	大小	交付方式	IM 消息格式
文本/Markdown	< 4KB	直接内嵌回复	text 消息
图片/截图	< 10MB	上传 IM + 原会话回传	image 消息
文档/报告	< 10MB	上传 IM + 原会话回传	file 消息
大文件	> 10MB	MinIO presigned URL	卡片消息（含链接）
多文件	任意	打包 zip + MinIO URL	卡片消息（含下载链接）

4.2 产物交付流程

4.3 产物交付失败处理

交付失败处理策略（三级降级）：

Level 1: 上传 IM 失败
  → 重试 ×3（指数退避：1s, 2s, 4s）
  → 仍失败 → 降级到 Level 2

Level 2: 降级为对象存储链接
  → 生成 MinIO presigned URL
  → 通过 IM 文本消息发送链接
  → 仍失败 → 降级到 Level 3

Level 3: 通知员工到控制台查看
  → 发送 IM 文本消息：
     "你的任务已完成，产物可在控制台查看：
      https://agent.example.com/tasks/{task_id}"
  → 同时记录告警（交付通道异常）

五、任务状态管理

5.1 任务状态机

5.2 任务存储

sql

CREATE TABLE task_registry (
    task_id         VARCHAR(64) PRIMARY KEY,
    user_id         VARCHAR(64) NOT NULL,
    instance_id     VARCHAR(64) NOT NULL,
    session_id      VARCHAR(128) NOT NULL,
    
    -- 任务内容
    input           JSONB NOT NULL,
    request_context JSONB DEFAULT '{}',
    
    -- 状态
    status          VARCHAR(20) NOT NULL DEFAULT 'submitted',
    -- submitted/accepted/running/waiting_approval/completed/failed/timeout/rejected/waking
    
    -- 产物
    artifacts       JSONB DEFAULT '[]',
    artifact_count  INTEGER DEFAULT 0,
    total_tokens    INTEGER DEFAULT 0,
    
    -- 时间
    submitted_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    accepted_at     TIMESTAMPTZ,
    completed_at    TIMESTAMPTZ,
    timeout_at      TIMESTAMPTZ,
    
    -- 追踪
    trace_id        VARCHAR(64),
    source          VARCHAR(20),        -- web/feishu/wecom/dingtalk/cron/api
    
    -- 错误
    error_code      VARCHAR(50),
    error_message   TEXT
);

CREATE INDEX idx_task_user ON task_registry(user_id, submitted_at DESC);
CREATE INDEX idx_task_status ON task_registry(status) WHERE status NOT IN ('completed', 'failed', 'timeout', 'rejected');

5.3 任务超时处理

超时类型	阈值	处置
任务执行超时	`request_context.timeout_seconds`（默认 300s）	实例侧中止 → 返回 task_error 事件
审批等待超时	按风险等级（30min / 1h）	自动拒绝 → 返回 task_error
实例唤醒超时	60s	返回 wake_failed → 通知员工稍后重试
SSE 连接超时	600s 无事件	网关发送心跳；客户端 EventSource 自动重连

六、任务幂等与去重

6.1 幂等保证

每个任务必须有唯一的 task_id。控制面在处理任务前检查幂等：

python

async def submit_task(user_id: str, task_id: str, request: TaskRequest):
    # 幂等检查
    existing = await task_store.get(task_id)
    if existing:
        if existing.user_id != user_id:
            raise PermissionError("Task belongs to another user")
        return existing  # 返回已有任务状态（幂等）
    
    # 创建新任务
    task = TaskRecord(
        task_id=task_id,
        user_id=user_id,
        status="submitted",
        ...
    )
    await task_store.create(task)
    return task

6.2 去重策略

来源	去重键	窗口
Web 控制台	task_id（前端生成 UUID）	永久
企业 IM	event_id（IM 平台消息 ID）	5 分钟
定时任务	`{cron_name}:{schedule_time}`	24 小时
上层系统	task_id（调用方生成）	永久

七、并发控制

7.1 同一会话串行

同一 session_id 的任务串行执行——前一个任务完成后才处理下一个。避免上下文混乱：

python

async def dispatch_task(task: TaskRecord):
    session_key = task.session_id
    
    # 获取会话锁
    lock = redis_client.lock(f"session_lock:{session_key}", timeout=600)
    
    async with lock:
        # 转发到实例
        response = await forward_to_instance(task)
        
        # 处理 SSE 事件流
        async for event in response.sse_stream():
            yield event
            if event.type == "task_complete":
                break

7.2 不同会话并行

不同 session_id 的任务可以并行执行，互不影响。实例侧 Agent 运行时支持多会话并行。

7.3 并发限制

维度	限制	说明
单实例并发会话	≤ 5	防止单实例过载
单用户并发任务	≤ 3	防止单用户占用过多资源
全局并发任务	≤ 节点数 × 30	集群容量保护

超限请求返回 429 + 排队提示：

json

{
  "error": "concurrency_limit_exceeded",
  "message": "当前任务排队中，预计等待 15 秒",
  "queue_position": 2,
  "retry_after": 15
}

09 - 任务编排：下发、执行与产物交付 ​

一、设计目标 ​

二、任务下发链路 ​

2.1 入站规范 ​

2.2 多入口下发 ​

2.3 会话路由规则 ​

三、SSE 事件流 ​

3.1 事件类型 ​

3.2 网关 SSE 转发 ​

四、产物交付 ​

4.1 产物类型与交付方式 ​

4.2 产物交付流程 ​

4.3 产物交付失败处理 ​

五、任务状态管理 ​

5.1 任务状态机 ​

5.2 任务存储 ​

5.3 任务超时处理 ​

六、任务幂等与去重 ​

6.1 幂等保证 ​

6.2 去重策略 ​

七、并发控制 ​

7.1 同一会话串行 ​

7.2 不同会话并行 ​

7.3 并发限制 ​