写在前面
Workforce Hub 的 agent 模块是整个项目里最大的一个模块——九千多行代码，比考勤、假期这些业务模块都大。原因很简单：它承载了"AI 能力怎么安全地长在企业系统里"这个问题。
系列里之前写过权限模型，这篇聚焦一条最日常的链路：员工在 IM 里给 AI 助手发一条消息，到 AI 回复出现，中间发生了什么。 拆开看这条链路，会发现"调用大模型"反而是最简单的半步，剩下九步都在为它兜底。
一、为什么不能同步调模型
最直觉的实现：员工发消息 → 后端调 LLM → 拿到回复 → 写回会话。一个 HTTP 请求从头走到尾。
真写出来会立刻撞上三堵墙：
墙一：模型太慢。 LLM 一回复就是几秒起步，长任务几十秒。HTTP 同步等待要么超时，要么把 Web 容器的线程占满——十个人同时问 AI，十个线程全挂在"等模型"上，整个系统的其他接口跟着遭殃。
墙二：失败要重试，重试就重复。 调用超时了，算失败还是成功？如果模型其实已经生成回复但响应丢了，重试会让会话里出现两条一样的回答。
墙三：黑盒不可观测。 回复慢、回复错、根本没回——排查时你需要知道：任务什么时候创建的、什么时候开始跑、调的哪个版本、失败在哪一类错误上。同步调用链里这些只有日志，没有结构化记录。
所以我的答案是把"一次 AI 回复"建模成一个任务：落库、异步执行、状态机推进、结果回写。四段各管各的。
二、先落库，再执行：事务边界的讲究
任务化的第一步是把任务写进数据库。这里的事务设计有个反直觉的地方：
// IM 会话触发 AI 任务：先入库任务记录，事务提交后再异步执行
@Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
public AgentTaskResponse createConversationReplyTask(ImAgentTriggerCommand command) {
...
agentRepository.insertTask(taskId, agent.id(), agent.publishedVersionId(),
"CONVERSATION", "IM_CONVERSATION", conversationId,
"USER", senderId, "PENDING", null, senderId);
audit(taskId, "TASK_CREATED", "AGENT", agent.id(), "消息会话触发AI任务", false);
scheduleConversationExecution(taskId, command);
return toTaskResponse(agentRepository.getTaskOrThrow(taskId));
}
插入任务用了 REQUIRES_NEW——新开一个事务，而不是加入外层事务。为什么？因为消息写入和任务创建是两件事：消息发成功了任务就该存在，不应该因为消息外层事务后面回滚而跟着消失；反过来也一样。
真正的讲究在下一步：
// 事务提交后再异步执行，避免任务执行期间持有数据库连接
private void scheduleConversationExecution(Long taskId, ImAgentTriggerCommand command) {
Runnable execution = () -> agentExecutor.execute(() -> executeConversationReplyTask(taskId, command));
if (TransactionSynchronizationManager.isSynchronizationActive()) {
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
@Override
public void afterCommit() {
execution.run();
}
});
return;
}
execution.run();
}
异步执行被挂在 afterCommit 上——事务提交之后才投递到线程池，而不是方法里直接提交。这里有两个教训叠在一起：
教训一：提前投递会读到"不存在的任务"。 如果在事务提交前就把任务丢进线程池，异步线程可能比事务提交先跑，它去查任务时数据库里还没有这条记录——一个经典的并发时序坑，而且测试环境很难复现。
教训二：事务里等模型 = 数据库连接泄漏。 早期版本我把模型调用写在了事务方法里，LLM 转圈的几十秒里事务一直开着，数据库连接一直被占着。连接池十个连接，十个人同时问 AI，整个系统就没有连接可用了。事务的边界必须是"数据库操作"的边界，而不是"业务动作"的边界。
三、拾取任务：状态做护栏
异步线程拿到任务后，第一步不是干活，而是在一个新事务里"认领"它：
private ConversationExecutionContext startConversationExecution(Long taskId, ImAgentTriggerCommand command) {
return transactionTemplate.execute(status -> {
AgentTaskRow task = agentRepository.getTaskOrThrow(taskId);
if (!"PENDING".equals(task.status())) {
return null; // 只有 PENDING 状态的任务才能被拾取
}
agentRepository.markTaskRunning(taskId, command.triggerUserId());
// 工作流模式需要把配置快照写入任务，防止执行期间版本被修改
if (workflowParser.isWorkflowConfig(version.workflowConfigJson())) {
agentRepository.updateTaskWorkflowSnapshot(taskId, version.workflowConfigJson(), ...);
}
audit(taskId, "TASK_RUNNING", "AGENT_TASK", taskId, "AI会话任务开始执行", false);
return new ConversationExecutionContext(taskId, agent, version, ...);
});
}
两个护栏：
状态检查是幂等的前提。 为什么不直接信任"投递给我的任务就是我的"？因为同一条消息可能触发两次投递（重试、重复消费）。检查 PENDING 再改 RUNNING，意味着无论投递几次，只有第一次真正执行。这个模式和定时任务抢锁是同一个思想：用数据库状态做唯一的裁判。
配置快照是时间 Machine。 任务开始执行时，把 agent 版本的工作流配置原样写进任务记录。这样管理员中途改了配置、甚至改坏了、删了版本，正在跑的任务用的是自己快照里的那份——执行中的任务不关心现在，只关心它出生那一刻的世界。
四、两条执行路径：工作流优先，单轮兜底
真正的执行部分反而是最朴素的：
private void executeConversationReplyTask(Long taskId, ImAgentTriggerCommand command) {
ConversationExecutionContext context = startConversationExecution(taskId, command);
if (context == null) return;
try {
if (workflowParser.isWorkflowConfig(context.version().workflowConfigJson())) {
// 有工作流配置：走编排执行
WorkflowExecutionResult result = workflowExecutor.execute(...);
finishConversationSucceeded(command, context, result.replyText(), result.traceJson());
return;
}
// 否则：单轮对话模式
AgentModelResponse modelResponse = agentModelClient.generateReply(...);
finishConversationSucceeded(command, context, modelResponse.text(), null);
} catch (AgentModelException e) {
finishConversationFailed(command, context, e.failureType(), e.getMessage());
} catch (Exception e) {
finishConversationFailed(command, context, "UNKNOWN", e.getMessage());
}
}
助手可以配置成两种形态：简单的直接问模型（单轮），复杂的走工作流编排（多节点、工具调用、人工确认——那是另一篇文章的主题）。
值得看的是失败处理。失败不是一锅端的 “ERROR”，而是先按模型异常分型（failureType——超时、响应不合法、鉴权失败各不相同），未知的再归入 UNKNOWN。这个分型在管理端是有用的：失败摘要聚合页按类型统计，你能一眼看出这段时间是"供应商超时多了"还是"提示词写出了不合法响应"。
五、结果回写：幂等键兜底重复
任务跑完要把回复写回 IM 会话：
private void notifyImAgentSucceeded(ImAgentTriggerCommand command, AgentRow agent,
Long taskId, String replyText) {
...
callback.onAgentReply(new ImAgentReplyResult(
command.conversationId(),
command.sourceMessageId(),
agent.id(),
taskId,
replyContent.messageType(),
replyContent.contentJson(),
"SUCCEEDED",
null, null,
agentMessageIdempotencyKey(taskId) // "agent-task-{taskId}"
));
}
回复分两种形态：模型如果按约定返回了卡片 JSON（knowledge_answer、agent_result），就渲染成结构化卡片；普通文本就包成 TEXT 消息。无论哪种，消息都携带 agentTaskId 和幂等键。
幂等键是这条链路最后、也是最容易被省掉的一道闸。任务收尾逻辑如果因为网络抖动重放一次，回写逻辑靠 "agent-task-" + taskId 这个键识别出"这条回复我已经写过"，直接跳过。没有它，墙二里说的"重试产生重复回复"会原样复现——大模型贵就贵在每次生成都可能不一样，两条不一样的回复挂在同一条用户消息下面，用户的表情会很精彩。
六、卡住的任务：超时收口
状态机里有个特殊状态：WAITING_HUMAN_CONFIRM——工作流执行到人工确认节点时会暂停，等人点头。人要是永远不点头呢？
@Scheduled(fixedDelayString = "${workforce.agent.timeout-scan.fixed-delay-ms:60000}")
public void scheduledTimeoutScan() {
if (!scheduledEnabled) return;
try {
runTimeoutScan("SCHEDULED", scheduledBatchSize);
} catch (RuntimeException e) {
log.warn("Agent人工确认超时定时扫描失败", e);
}
}
每 60 秒扫一批"等待确认但过了截止时间"的任务：先抢分布式任务锁（抢不到说明别的实例在扫，直接 SKIPPED），然后把过期任务批量标记为 CANCELLED，写审计日志，最后如果任务源于审批预审，回调审批流通知超时。
三个参数都是防御性的：批次上限 500（防止一次扫出十万条把事务撑爆）、扫描开关（出问题可热关闭）、以及每次扫描都留一条 job_run 流水（开始时间、扫描量、取消量、失败原因）——定时任务最大的敌人是"它到底跑没跑"这个疑问，流水表就是答案。
七、每一步都留痕
整条链路的状态推进，每一步都写一条审计记录：
TASK_CREATED → TASK_RUNNING → TASK_SUCCEEDED / TASK_FAILED / TASK_TIMEOUT_CANCELLED
审计带 riskFlag（失败和超时标红）、带脱敏后的摘要。员工看到的是"AI 助手回复了"，管理员看到的是一条完整的任务流水线。这两个视角靠的就是这套结构化记录——出问题时，九千行的 agent 模块不会变成九千行猜谜。
八、小结
三点体会：
- 异步化不是性能优化，是正确性设计。 任务化解决的是重复回复、失败分类、可观测这三个同步调用根本无法回答的问题，速度只是附赠品。
- 事务边界跟着数据库操作走，不跟着业务动作走。
afterCommit投递、REQUIRES_NEW落库、快照隔离，三件事都在回答同一个问题：并发世界里，"开始执行"那一刻的数据得是确定的。 - 每一层兜底都要有一把钥匙。 状态护栏防重复执行，幂等键防重复回复，超时扫描防永久卡住，审计流水防黑盒。没有哪一层"应该用不到"——它们每一层都在某个凌晨被用到过。
员工按下发送，两秒钟后 AI 的回复出现在会话里。这两秒钟背后是一条完整的流水线：落库、认领、快照、执行、回写、留痕。AI 应用工程，工程的部分从来不比 AI 的部分少。
默认评论
Halo系统提供的评论