共计 3801 个字符，预计需要花费 10 分钟才能阅读完成。
当 AI Agent 遇上长周期任务
最近在给客户设计 AI 客服工单系统时，发现传统服务架构在面对需要跨天的对话场景时频繁翻车。每次排查问题都看到熟悉的报错三连：
- 异步调用丢失：凌晨 3 点触发的邮件发送 API 调用，因为 Pod 重启直接蒸发
- 上下文断裂：用户第 5 次询问 ” 上个月说的折扣还能用吗 ” 时，系统一脸茫然
- 错误恢复困难：OCR 识别失败后，整个流程带着半成品数据继续往下跑
这些正是 AI Agent 在复杂业务流中的典型痛点——它们需要像人类一样记住上下文，但现有微服务架构更像金鱼记忆。
为什么是 Cadence？
对比过主流工作流引擎后，Cadence（现更名为 Temporal）的两个特性直接命中要害：
- 持久化虚拟内存：整个工作流的状态（包括局部变量）会自动保存，即使进程崩溃也能从断点恢复
- 确定性重放：通过事件溯源机制，可以精确复现任何异常现场
和 Airflow 这类批处理调度器不同，Cadence 专为 长时间运行 、 需要人工干预 的场景设计。举个典型对比：
| 特性 | Airflow | Cadence |
|---|---|---|
| 任务中断恢复 | 需手动重试 | 自动续跑 |
| 执行时长限制 | 默认 48 小时 | 无硬性限制 |
| 人工介入接口 | 无 | 内置 Signal 机制 |
| 状态存储方式 | 数据库快照 | 事件流 + 检查点 |
实战：给 AI Agent 装上持久化内存
基础工作流定义（Go 版本）
// 定义工作流接口
type OrderHandlingWorkflow interface {Execute(ctx workflow.Context, orderID string) (string, error)
}
// 实现工作流
type orderHandlingWorkflow struct{}
func (w *orderHandlingWorkflow) Execute(ctx workflow.Context, orderID string) (string, error) {
// 持久化配置：自动记录执行进度
ao := workflow.ActivityOptions{
ScheduleToStartTimeout: time.Hour,
StartToCloseTimeout: time.Hour * 24,
HeartbeatTimeout: time.Minute * 5, // 关键！避免僵尸任务
RetryPolicy: &cadence.RetryPolicy{
InitialInterval: time.Second,
BackoffCoefficient: 2.0,
MaximumInterval: time.Minute * 10,
ExpirationInterval: time.Hour * 24,
},
}
ctx = workflow.WithActivityOptions(ctx, ao)
// 执行链式调用
var agentResponse string
err := workflow.ExecuteActivity(ctx, AIClassifyOrder, orderID).Get(ctx, &agentResponse)
if err != nil {return "", fmt.Errorf("AI 分类失败: %v", err)
}
// 人工审核分支
if strings.Contains(agentResponse, "NEED_HUMAN") {
var humanInput string
workflow.Go(ctx, func(ctx workflow.Context) {
// 发送邮件通知人工审核
err := workflow.ExecuteActivity(ctx, SendReviewEmail, orderID).Get(ctx, nil)
// 等待人工响应信号（最长等待 3 天）workflow.SetQueryHandler(ctx, "get_status", func() (string, error) {return "等待人工审核中", nil})
sel := workflow.NewSelector(ctx)
sel.AddReceive(workflow.GetSignalChannel(ctx, "human_decision"),
func(c workflow.Channel, more bool) {c.Receive(ctx, &humanInput)
})
sel.Select(ctx)
})
}
return workflow.ExecuteActivity(ctx, AIFinalizeOrder, agentResponse).Get(ctx, nil)
}关键设计：状态同步
AI Agent 需要与工作流保持双向通信。我们采用信号机制实现：
# Python 信号处理示例
@workflow.defn
class AITicketWorkflow:
def __init__(self):
self._current_state = "INIT" # 持久化变量
self._llm_context = {}
@workflow.run
async def run(self, ticket_id: str):
# 注册信号处理器
workflow.set_signal_handler("update_context", self._handle_llm_update)
while True:
if self._current_state == "WAITING_FOR_LLM":
# 将上下文通过 Activity 传给 AI
llm_response = await workflow.execute_activity(
call_llm,
args=[self._llm_context],
start_to_close_timeout=timedelta(minutes=3),
retry_policy=RetryPolicy(initial_interval=timedelta(seconds=10),
maximum_attempts=3
)
)
self._process_ai_response(llm_response)
async def _handle_llm_update(self, new_context: dict):
# 来自外部的上下文更新
self._llm_context.update(new_context)
workflow.upsert_search_attributes({"LastContextUpdate": datetime.now()
}) # 可搜索字段性能调优实战
熔断配置黄金法则
- Activity 级别：
- 短任务（<5 分钟）：StartToCloseTimeout = 2x 预期时长
-
长任务：启用 HeartbeatTimeout（建议值 = 预期间隔的 2 倍）
-
工作流级别：
workflow.RegisterOptions{ WorkflowExecutionTimeout: time.Hour * 24 * 7, // 最长 1 周 WorkflowTaskTimeout: time.Minute * 10, // 单次决策超时 }
历史事件分片方案
当工作流执行步骤超过 1 万步时：
-
方案 1：启用
ContinueAsNew@workflow.defn class ChunkedWorkflow: async def run(self, input_data: list): chunk_size = 1000 for i in range(0, len(input_data), chunk_size): await self._process_chunk(input_data[i:i+chunk_size]) if i + chunk_size < len(input_data): workflow.continue_as_new(input_data[i+chunk_size:]) -
方案 2：配置 Cassandra 分片
# cadence-server 配置 persistence: numHistoryShards: 1024 # 生产环境建议值 advancedVisibilityStore: "cassandra"
生产环境避坑指南
信号丢失三大场景
- 工作流未运行：信号发送时工作流尚未启动或已结束
-
✅ 解决方案：先查询
DescribeWorkflowExecution -
信号队列积压：高频信号导致处理延迟
-
✅ 解决方案：合并信号 + 去重（如
workflow.Promise） -
版本不匹配：旧版本工作流代码无法处理新信号
- ✅ 解决方案：
// 信号处理器兼容写法 workflow.SetSignalHandler(ctx, "new_signal", func() {if workflow.GetVersion(ctx, "new_feature", 1, 1) == 1 {// 旧版处理逻辑} })
版本升级策略
采用 阶段式发布：
- 部署新版本代码（保持兼容旧工作流）
- 通过
GetVersion实现逻辑分支 - 对已完成的工作流执行
Reset：cadence --do <domain> workflow reset \ --wid <workflow_id> \ --reset_type LastDecisionTaskID \ --reason "version_upgrade"
开放挑战：LLM 的交互式修正
现有方案在遇到这种情况时会崩溃：
用户：” 刚才那个回答不对，我是想问退款流程 ”
可能的突破方向：
- 动态上下文注入：通过 Signal 实时更新工作流内存
- 子工作流隔离：每个对话分支作为独立子工作流
- 版本化状态快照：类似 git 的 branch/rebase 机制
期待看到大家更有创意的解决方案！
正文完
