feat(cloud-agent): 单步输出上限与秒级墙钟,超时/空输出可自恢复
画布 Agent 一轮要跑很多步模型调用,此前只有"按任务类型的分钟级超时":某一步卡住只能等 整轮判死;更根本的是每一步请求根本不带输出上限,不写 max_tokens 时上游按剩余上下文放行, 思考模型可以把单步拖到几分钟,且偶发把整个输出预算花在推理上、正文与工具调用皆空。 - 新增运行时策略 task.agentStepTimeoutSeconds 与 task.agentStepMaxOutputTokens:秒级墙钟 与单步输出上限,都支持环境变量给部署级默认值、管理端热更新;旧配置 JSON 缺字段按默认值 回填,无需数据迁移。 - 单步墙钟到点中止这一步时不再把整轮判死:以可识别标记记为该步失败,运行期关掉上游思考 重试同一步一次(model_failure_recovered / step_timeout_retried),重试仍超时才以可读原因 (model_step_timeout)结束本轮。 - 空输出催办用尽后新增一级升级:关思考 + 放大输出预算重试同一步 (model_failure_recovered / empty_output_escalated),而不是直接判死。放大档在原值上翻倍 并以 131072 封顶,策略值为 0 时用 32768 兜底,让重试仍然有界。 - 输出上限真正下发到上游:textOptions.maxOutputTokens 成为任务信封字段,Agent 的两条请求 路径(内置协议与声明式协议)组装请求体时写入 max_tokens / max_output_tokens,与既有的 input.MaxOutputTokens 取较小值,避免策略上限超过模型声明的物理上限。 - task_worker:执行时限改用带画布 Agent 分支的 taskExecutionTimeout;执行时限到点本身不再 被当成"租约丢失",否则任务会停在 running 等租约过期后被反复重跑(实测一次调用被重跑成 三次上游请求)。续租仍用上游已合入的"不继承父 context 取消"的独立 context。
This commit is contained in:
1 parent
6f45240599
commit
63c9e32259
11 files changed
+609
-11
No files matched your search
@@ -137,7 +137,7 @@ func (w *taskWorkerCoordinator) processClaimedTask(task *model.Task, globalSlot
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), taskExecutionTimeoutWithPolicy(task.Type, policy.Task))
|
||||
ctx, cancel := context.WithTimeout(context.Background(), taskExecutionTimeout(task, policy.Task))
|
||||
defer cancel()
|
||||
leaseDone := make(chan struct{})
|
||||
leaseLost := make(chan error, 1)
|
||||
@@ -227,11 +227,17 @@ func (w *taskWorkerCoordinator) processClaimedTask(task *model.Task, globalSlot
|
||||
if code, _ := ChannelSlotFailureDetails(err); code != "" {
|
||||
channelSlotFailedBeforeRequest = true
|
||||
}
|
||||
// 续租使用独立 context;即使执行同时超时,真实租约失效仍必须阻止写入。
|
||||
// 续租使用独立 context,真实租约失效必须阻止写入;但执行时限到点
|
||||
// (单步墙钟或任务超时)本身不是"租约丢失":续租可能正好被同一个时限打断,
|
||||
// 此时必须继续走失败收尾,否则任务停在 running 被反复重跑。
|
||||
deadlineExpired := errors.Is(ctx.Err(), context.DeadlineExceeded)
|
||||
select {
|
||||
case leaseErr := <-leaseLost:
|
||||
_ = s.log(task.UserID, task.ID, "warn", "任务租约失效,等待其他 worker 恢复", leaseErr.Error())
|
||||
return leaseErr
|
||||
if !deadlineExpired {
|
||||
_ = s.log(task.UserID, task.ID, "warn", "任务租约失效,等待其他 worker 恢复", leaseErr.Error())
|
||||
return leaseErr
|
||||
}
|
||||
_ = s.log(task.UserID, task.ID, "warn", "任务执行时限到点,按失败收尾(不视为租约丢失)", leaseErr.Error())
|
||||
default:
|
||||
}
|
||||
decryptedInput, decryptErr := s.decryptTaskInputJSON(task.InputJSON)
|
||||
@@ -252,8 +258,14 @@ func (w *taskWorkerCoordinator) processClaimedTask(task *model.Task, globalSlot
|
||||
if newAPIChannel2TaskSyncExpired(*task, err, time.Now()) {
|
||||
err = errors.New("上游任务长时间未同步,已停止自动查询,请确认渠道任务状态后重试。")
|
||||
}
|
||||
if errors.Is(err, context.DeadlineExceeded) {
|
||||
err = errors.New(taskTimeoutMessage(task.Type))
|
||||
if errors.Is(err, context.DeadlineExceeded) || deadlineExpired {
|
||||
// 画布 Agent 的单步超时是可恢复事件(运行期会关思考重试同一步),
|
||||
// 因此必须与"任务执行超时"区分开,否则只能整轮判死。
|
||||
if cloudAgentModelOperation(task) {
|
||||
err = errors.New(cloudAgentStepTimeoutError + ",已中止这一步")
|
||||
} else {
|
||||
err = errors.New(taskTimeoutMessage(task.Type))
|
||||
}
|
||||
}
|
||||
s.noteAgentMemoryCompactTask(*task, nil, err)
|
||||
return terminal.handleExecutionFailure(task, err, providerSucceeded, channelSlotFailedBeforeRequest)
|
||||
@@ -307,6 +319,21 @@ func taskLeaseRenewContext(parent context.Context) (context.Context, context.Can
|
||||
return context.WithTimeout(context.WithoutCancel(parent), 5*time.Second)
|
||||
}
|
||||
|
||||
// taskExecutionTimeout 解析一次任务的执行墙钟:画布 Agent 的单步调用可以配秒级超时
|
||||
// (AgentStepTimeoutSeconds),没配时沿用文本任务超时。秒级粒度是必要的——一轮里每一步
|
||||
// 都是分钟级的调用,分钟粒度改不动"某一步卡住"的体验。
|
||||
// 超时的表现是任务错误里带 cloudAgentStepTimeoutError 标记,运行期据此关思考重试同一步,
|
||||
// 而不是把整轮判死(见 cloud_agent_step_timeout.go)。
|
||||
func taskExecutionTimeout(task *model.Task, policy RuntimeTaskPolicy) time.Duration {
|
||||
if task != nil && cloudAgentModelOperation(task) && policy.AgentStepTimeoutSeconds > 0 {
|
||||
return time.Duration(policy.AgentStepTimeoutSeconds) * time.Second
|
||||
}
|
||||
if task == nil {
|
||||
return time.Duration(policy.DefaultTimeoutMinutes) * time.Minute
|
||||
}
|
||||
return taskExecutionTimeoutWithPolicy(task.Type, policy)
|
||||
}
|
||||
|
||||
func taskExecutionTimeoutWithPolicy(taskType string, policy RuntimeTaskPolicy) time.Duration {
|
||||
switch {
|
||||
case strings.HasPrefix(taskType, "canvas_video") || strings.HasPrefix(taskType, "video_"):
|
||||
|
||||
Reference in new issue
Block a user