1. 什么是 AI Agent它不是“更聪明的聊天机器人”而是可执行闭环任务的工程实体你刷到过这样的标题“AI Agent 让小红书自动发消息”、“用 AI Agent 开发 Django”、“Hermes Agent 第三方工作台接入”。这些说法背后藏着一个被严重泛化、甚至误读的概念——很多人以为 Agent 就是“调用大模型 API 多几次”或者“加个记忆模块就叫智能体”。但真实情况是一个能落地的 AI Agent本质是一套具备明确输入-处理-输出边界的可编排、可观测、可容错的软件系统其核心不在于用了哪个 LLM而在于七个决策点如何被工程化地锚定与协同。我在带团队从零搭建三个生产级 Agent 系统含金融风控辅助决策、工业设备远程诊断、电商客服意图路由的过程中反复验证90% 的失败项目不是败在模型能力不足而是卡在对这七个决策点缺乏显式设计——它们被模糊地塞进提示词里靠运气运行一出异常就全线崩盘。所谓“七要素”其实是学术界对 Agent 理想形态的抽象描述感知Perception、记忆Memory、规划Planning、行动Action、工具调用Tool Use、反思Reflection、学习Learning。但问题来了这些词放在论文里很美在工程现场却毫无操作性。比如“记忆”——你是用 Redis 缓存对话历史还是用向量库做长期知识检索抑或用结构化数据库存用户偏好每种选择对应完全不同的数据一致性模型、延迟容忍度和运维成本。“反思”更典型是每次调用后让 LLM 自评回答质量还是用规则引擎判断响应是否含敏感词或是引入独立的 reward model 打分没有决策依据就只能拍脑袋选。真正决定一个 Agent 能不能上线、能不能扛住流量、能不能被业务方信任的是它背后那七个必须被代码固化、被配置驱动、被监控覆盖的决策点。它们分别是输入解析决策点、状态建模决策点、目标拆解决策点、工具路由决策点、执行调度决策点、容错回滚决策点、结果校验决策点。注意这七个点不是流程步骤而是系统在每个关键岔路口必须做出的、有明确输入输出和判定逻辑的“开关”。比如“工具路由决策点”当用户说“帮我查下昨天北京的天气”系统必须在毫秒级内完成三件事——识别出这是查询类意图、确认当前无可用本地缓存、判断应调用气象 API 而非搜索网页。这个决策过程不能依赖 LLM 的自由发挥必须由预设规则轻量模型上下文特征共同驱动。我见过太多团队把 Agent 当成“LLM Prompt 工程”的升级版结果上线后发现用户问“把上周五的会议纪要发给张三”系统要么死循环调用邮件工具要么把纪要内容直接塞进邮件正文导致超长报错要么根本识别不出“上周五”这个时间表达式。根源就在于他们没把“目标拆解决策点”工程化——没有定义时间解析器、没有建立收件人实体识别模块、没有设置邮件长度截断策略。所有这些都得写成代码而不是指望大模型“自己懂”。所以这篇文章不讲概念不画架构图不罗列 LangChain/Dify/CrewAI 的对比参数。我要带你逐个击穿这七个决策点告诉你每个点在 Rust/Python/Go 里怎么写、为什么这么写、踩过哪些坑、监控指标怎么设。因为真正的 Agent 工程实现从来不是堆砌框架而是把每一个模糊的“智能”要求翻译成一行行可测试、可回滚、可审计的代码。2. 七要素 vs 七个决策点为什么必须用工程视角重定义 Agent把“七要素”直接当开发清单是新手最容易掉进的陷阱。我带过的实习生里有位同学花两周用 LangChain 搭了个“带记忆的客服 Agent”演示时流畅无比但一上测试环境就崩溃——用户连续问三个问题后系统开始把前两轮的客户投诉记录当成当前问题的背景给出完全错误的补偿方案。复盘发现他的“记忆”要素实现只是简单把全部对话 history 塞进 prompt没做任何状态隔离、时效过滤或敏感信息脱敏。这暴露了根本问题要素是功能描述决策点是控制开关要素告诉你“要什么”决策点告诉你“在哪儿切、怎么切、切坏了怎么办”。我们来对照拆解。先看传统七要素的典型实现误区感知Perception常被简化为“接收用户文本输入”。但真实场景中输入可能是语音转文字后的带时间戳片段、多模态图像OCR 文本、IoT 设备上报的 JSON 数据流。如果只做字符串接收连输入源合法性校验都做不到。记忆Memory多数人用 ChromaDB 存向量化对话但没考虑用户 A 的会话向量和用户 B 的混在一起检索会不会泄露隐私向量相似度阈值设为 0.7 还是 0.85低于阈值时是返回空结果还是触发 fallback 流程这些全是决策点不是要素。规划Planning常被等同于“让 LLM 生成 step-by-step 思路”。但生产环境要求规划结果必须可序列化、可中断、可重入。如果 LLM 返回“先查库存再比价格最后下单”系统得能精确识别出这三个原子动作并在第二步失败时自动跳过第三步并通知用户“价格查询失败已为您保留库存”。现在我们用工程语言重定义这七个决策点每个点都包含触发条件、输入约束、决策逻辑、输出契约、失败兜底五个维度2.1 输入解析决策点不是“接收”而是“理解边界”这是 Agent 的第一道闸门。它的任务不是把用户输入原样传给 LLM而是判断这条输入是否属于本 Agent 的责任域是否满足最低语义完整性是否携带可验证的上下文标识触发条件HTTP 请求到达 / WebSocket 消息抵达 / Kafka Topic 消费输入约束必须含user_id字段用于后续状态隔离、session_id用于会话追踪、timestamp用于时效判断决策逻辑校验user_id是否在白名单防恶意刷请求检查session_id是否匹配当前活跃会话防会话劫持解析自然语言中的时间表达式如“昨天”、“下周三”转换为 UTC 时间戳若解析失败则标记为time_parse_error输出契约返回结构化对象{intent: query_weather, params: {city: Beijing, date: 2024-06-15}, context: {user_tz: Asia/Shanghai}}失败兜底若user_id不合法直接返回 HTTP 401若时间解析失败返回{error: invalid_time_format, suggestion: 请使用YYYY-MM-DD格式}提示很多团队在这里犯的错是把输入解析和 LLM 调用耦合。正确做法是——输入解析必须 100% 由确定性代码完成LLM 只负责处理已清洗、已标注、已结构化的输入。我见过某电商 Agent 因未校验user_id被恶意构造的超长 ID 导致 Redis 内存溢出整个服务雪崩。2.2 状态建模决策点不是“记住”而是“管理快照”状态是 Agent 的灵魂但状态管理绝不是“把所有东西存起来”。它必须回答当前会话的权威状态是什么哪些数据必须强一致哪些可以最终一致状态变更的原子性如何保证触发条件输入解析通过后、每次工具调用返回后、定时心跳检测时输入约束状态快照必须含version乐观锁版本号、last_modified时间戳、dirty_flags变更标记位决策逻辑读取当前状态快照Redis Hash 或 PostgreSQL 行比较version若客户端传入的version小于服务端拒绝更新防脏写根据dirty_flags判断需更新的字段如仅inventory_status变更则只更新该字段输出契约返回新状态快照含version: 123,last_modified: 1718432100,data: {user_profile: {...}, cart_items: [...], pending_actions: [...]}}失败兜底若乐观锁冲突触发重试机制最多 3 次超时则降级为全量状态覆盖牺牲一致性保可用性注意状态建模最易被忽视的是“状态生命周期”。比如客服 Agent 中用户投诉记录需永久保存但临时生成的推荐商品列表 2 小时后自动过期。这必须在状态 schema 中明确定义 TTL 字段而非靠应用层清理。2.3 目标拆解决策点不是“思考”而是“分解契约”LLM 的“规划能力”不可信必须由工程逻辑兜底。目标拆解的核心是把模糊的用户意图转化为一组有明确前置条件、执行顺序、成功标准的原子任务链。触发条件状态建模完成后、且pending_actions为空时输入约束必须含intent如book_flight、params如{from: PEK, to: SHA, date: 2024-07-01}、context如用户常驻城市、历史偏好决策逻辑匹配 intent 到预定义的任务模板如book_flight模板含 4 个原子任务check_availability→validate_passenger_info→reserve_seat→send_confirmation检查每个任务的前置条件如reserve_seat需check_availability返回status: available为每个任务生成唯一task_id和deadline如reserve_seat的 deadline 是now 30s输出契约返回任务队列[{task_id: t_abc123, action: check_availability, params: {...}, deadline: 1718432130}, ...]失败兜底若无匹配模板返回{error: unsupported_intent, supported_intents: [query_weather, book_flight, ...]}实操心得任务模板不能硬编码在代码里。我们用 YAML 文件定义模板支持热加载。当业务方说“新增火车票预订”只需提交新 YAML无需发版。模板中deadline的设定有讲究——check_availability设 5sAPI 快send_confirmation设 60s含邮件发送延迟这直接影响整个链路的超时熔断策略。2.4 工具路由决策点不是“调用”而是“精准匹配”工具不是越多越好而是每个工具必须有明确的能力边界声明。路由决策点要解决当前任务该调用哪个工具调用参数是否合规工具是否处于可用状态触发条件任务队列中取出下一个任务时输入约束任务对象必须含action如check_availability、params如{flight_no: CA123}、tool_requirements如{auth_scope: flight_read, region: CN}决策逻辑查询工具注册中心Consul 或 Kubernetes Service筛选出status: healthy且metadata.auth_scope包含flight_read的工具实例校验params是否符合该工具的 OpenAPI Schema用 jsonschema 库实时验证若有多个健康实例按region标签路由优先调用同区域实例降低延迟输出契约返回工具调用指令{url: https://flight-api.internal/v1/check, method: GET, headers: {X-Auth: token_abc}, body: {...}}失败兜底若无健康实例返回{error: tool_unavailable, fallback: query_cache}若参数校验失败返回{error: invalid_params, schema: {...}}关键细节工具注册中心必须记录每个实例的last_heartbeat和error_rate_5m。我们曾因未监控error_rate_5m导致一个故障率 40% 的气象 API 实例持续被路由用户投诉暴增。现在规则是error_rate_5m 10%的实例自动下线。2.5 执行调度决策点不是“运行”而是“可控并发”Agent 不是单线程玩具。调度决策点决定任务以什么顺序执行允许多少并发失败后如何重试资源如何隔离触发条件工具路由返回有效指令后输入约束必须含task_id、deadline、retry_policy如{max_attempts: 3, backoff: exponential}、resource_limits如{cpu_ms: 100, memory_mb: 512}决策逻辑检查全局并发数Redis INCR若超限如 100进入等待队列启动 goroutineGo或 asyncio taskPython设置deadline超时用 context.WithDeadline执行中监控资源消耗若cpu_ms超限强制 kill 并标记resource_exhausted输出契约返回执行状态{task_id: t_abc123, status: running, started_at: 1718432100, timeout_at: 1718432130}失败兜底若超时触发retry_policy若重试后仍失败标记task_failed并推送告警Slack Prometheus Alert实测经验并发数不能拍脑袋定。我们用压测数据反推——在 95% P99 延迟 2s 的前提下最大并发为 87。超过此数Redis 连接池耗尽整个系统抖动。这个数字写死在配置中心随流量自动伸缩。2.6 容错回滚决策点不是“重试”而是“状态修复”这是区分玩具和生产系统的分水岭。回滚决策点要回答当某个任务失败如何最小化影响如何恢复一致状态用户看到什么触发条件任务执行返回status ! success或超时输入约束必须含task_id、failed_step如reserve_seat、current_state_snapshot失败前的状态决策逻辑查找该任务所属的原子任务链如book_flight链执行预定义的回滚脚本如reserve_seat失败则调用cancel_reservation工具若回滚也失败启动“状态修复模式”用current_state_snapshot覆盖当前状态并标记state_recovered输出契约返回修复结果{task_id: t_abc123, rollback_status: success, recovered_state_version: 122, user_message: 航班预订失败已为您取消预留座位}失败兜底若状态修复失败触发人工介入流程Jira ticket 自动生成指派给 SRE重要提醒回滚脚本必须幂等我们吃过亏——某支付回滚脚本未加幂等 key用户一次失败操作触发三次退款。现在所有回滚工具都强制要求idempotency_key参数且服务端校验。2.7 结果校验决策点不是“返回”而是“可信交付”最后一步不是把 LLM 输出扔给用户而是确保结果符合业务规则不含敏感信息格式可被下游系统消费用户能理解触发条件任务链全部完成或主动终止时输入约束必须含final_output原始 LLM 输出、execution_log完整执行轨迹、business_rules如“机票价格必须大于 0”、“邮件主题不能含 emoji”决策逻辑用正则和规则引擎校验final_output如价格字段price 0 and price 100000调用 PII 识别模型扫描敏感词身份证号、手机号、银行卡号发现则脱敏138****1234根据content_type如text/plain或application/json格式化输出输出契约返回交付包{payload: 您的航班已预订成功价格¥1280, format: text, trace_id: tr_789, confidence_score: 0.92}失败兜底若校验失败返回{error: output_violates_business_rule, rule: price_must_be_positive, raw_output: ...}并触发人工审核队列总结一下这七个决策点不是理论模型而是你写代码时必须声明的七个switch语句、七个配置项、七个监控埋点。它们共同构成 Agent 的“工程骨架”。脱离这个骨架谈“AI Agent”就像脱离钢筋谈摩天大楼——看着高一震就塌。3. 七个决策点的 Rust 实现为什么生产级 Agent 必须用 Rust 重写核心链路当团队第一次用 Python LangChain 上线客服 Agent 时P99 延迟是 3.2 秒错误率 1.8%。老板问“能不能做到 500ms 内响应错误率 0.1%” 我们没敢答应因为知道 Python 的 GIL 和 GC 在高并发下就是瓶颈。三个月后我们用 Rust 重写了从输入解析到结果校验的全部七个决策点P99 降到 187ms错误率 0.03%。这不是玄学是 Rust 的内存安全、零成本抽象和并发模型带来的确定性收益。下面我带你逐个看 Rust 如何实现这七个决策点重点讲清为什么必须用 Rust以及每个点的关键 unsafe 代码规避点。3.1 输入解析决策点用 Serde Regex 实现毫秒级结构化解析Python 版本用re.match()解析时间平均耗时 12msRust 版本用regexcrate chrono平均 0.3ms。关键在零拷贝解析// 定义输入结构体#[derive(Deserialize)] 自动绑定 JSON 字段 #[derive(Deserialize, Debug)] pub struct UserInput { pub user_id: String, pub session_id: String, pub timestamp: i64, #[serde(rename user_input)] pub text: String, } // 时间解析器预编译正则避免 runtime 编译开销 lazy_static::lazy_static! { static ref TIME_REGEX: Regex Regex::new(r(昨天|今天|明天)|(\d{4}年\d{1,2}月\d{1,2}日)).unwrap(); } impl UserInput { pub fn parse_time(self) - ResultDateTimeUtc, ParseError { // 零拷贝直接在 str 上匹配不 clone if let Some(caps) TIME_REGEX.captures(self.text) { // 根据捕获组选择解析逻辑 if caps.get(1).is_some() { // 处理“昨天”等相对时间 let offset_days match caps[1] { 昨天 -1, 今天 0, 明天 1, _ 0, }; Ok(Utc::now().date_naive().and_hms_opt(0,0,0).unwrap() Duration::days(offset_days)) } else { // 处理绝对日期 let date_str caps[2]; // 使用 chrono::parse_from_str避免字符串拼接 chrono::parse_from_str(date_str, %Y年%m月%d日) } } else { Err(ParseError::NoTimeFound) } } }注意这里没用String::from()创建新字符串所有操作都在str引用上完成。lazy_static确保正则只编译一次避免每次请求都 recompile。Python 版本做不到这点因为re.compile()在多线程下有锁竞争。3.2 状态建模决策点用 ArcMutex PostgreSQL 实现强一致快照Rust 的ArcMutex是线程安全状态管理的基石但必须配合数据库保证持久化// 状态快照结构体#[derive(Serialize, Deserialize)] 支持 serde_json #[derive(Serialize, Deserialize, Clone, Debug)] pub struct AgentState { pub version: u64, pub last_modified: i64, pub dirty_flags: VecString, pub data: StateData, } #[derive(Serialize, Deserialize, Clone, Debug)] pub struct StateData { pub user_profile: UserProfile, pub cart_items: VecCartItem, pub pending_actions: VecPendingAction, } // 状态管理器封装读写逻辑 pub struct StateManager { db_pool: PgPool, // SQLx PostgreSQL 连接池 cache: ArcMutexHashMapString, AgentState, // 内存缓存 } impl StateManager { // 读取状态先查缓存缓存未命中则查 DB查完写回缓存 pub async fn get_state(self, session_id: str) - ResultAgentState, StateError { // 1. 尝试从内存缓存读取ArcMutex 确保线程安全 { let cache self.cache.lock().await; if let Some(state) cache.get(session_id) { return Ok(state.clone()); } } // 2. 缓存未命中查 PostgreSQL let row sqlx::query( SELECT version, last_modified, data FROM agent_states WHERE session_id $1 AND status active ) .bind(session_id) .fetch_one(self.db_pool) .await?; let state: AgentState serde_json::from_value(row.get(data))?; // 3. 写回缓存注意这里要重新 lock因为上面的 lock 已释放 { let mut cache self.cache.lock().await; cache.insert(session_id.to_string(), state.clone()); } Ok(state) } // 更新状态乐观锁实现 pub async fn update_state(self, session_id: str, new_state: AgentState) - Result(), StateError { // 先检查版本号 let result sqlx::query( UPDATE agent_states SET version $1, last_modified $2, data $3 WHERE session_id $4 AND version $5 ) .bind(new_state.version) .bind(new_state.last_modified) .bind(serde_json::to_value(new_state.data)?) .bind(session_id) .bind(new_state.version - 1) // 期望旧版本 .execute(self.db_pool) .await?; if result.rows_affected() 0 { return Err(StateError::VersionConflict); } // 更新缓存 { let mut cache self.cache.lock().await; cache.insert(session_id.to_string(), new_state.clone()); } Ok(()) } }关键点ArcMutex用于内存缓存的线程安全访问但绝不依赖它做持久化。所有状态变更必须落库缓存只是加速层。update_state中的乐观锁 SQL 是强一致保障rows_affected() 0即表示版本冲突必须由上层逻辑处理重试。Python 的threading.Lock无法提供这种细粒度的、与 DB 事务联动的锁语义。3.3 目标拆解决策点用 YAML 模板 schemars 生成强类型任务链Python 版本用json.loads()解析任务模板类型不安全Rust 版本用serde_yamlschemars生成 OpenAPI Schema实现编译期校验// 定义任务模板结构YAML 文件映射 #[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] pub struct TaskTemplate { pub intent: String, pub steps: VecTaskStep, pub timeout_ms: u64, } #[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] pub struct TaskStep { pub action: String, pub params: HashMapString, String, pub requires: VecString, // 前置条件 pub deadline_ms: u64, } // 加载模板编译期保证 YAML 结构合法 pub fn load_task_template(intent: str) - ResultTaskTemplate, TemplateError { let yaml_path format!(templates/{}.yaml, intent); let yaml_content fs::read_to_string(yaml_path)?; // serde_yaml::from_str 在编译期检查字段是否存在、类型是否匹配 let template: TaskTemplate serde_yaml::from_str(yaml_content)?; // 运行时校验检查所有 requires 的 action 是否在 steps 中存在 for step in template.steps { for req in step.requires { if !template.steps.iter().any(|s| s.action *req) { return Err(TemplateError::MissingDependency(req.clone())); } } } Ok(template) } // 生成 OpenAPI Schema 供前端调用方校验 pub fn generate_openapi_schema() - ResultString, SchemaError { let schema schemars::schema_for!(TaskTemplate); Ok(serde_json::to_string_pretty(schema)?) }优势serde_yaml::from_str在解析时就做类型检查非法 YAML如timeout_ms: abc直接 panic不会让错误流入运行时。schemars生成的 Schema 可直接集成到 Swagger UI前端调用方能提前知道每个 intent 需要什么参数。Python 的pydantic也能做但 Rust 的编译期检查更彻底且无运行时反射开销。3.4 工具路由决策点用 Consul API jsonschema-rs 实现动态服务发现Rust 的reqwestjsonschema-rs组合比 Python 的requestsjsonschema更轻量、更可靠// 工具元数据结构 #[derive(Deserialize, Serialize, Clone, Debug)] pub struct ToolMetadata { pub name: String, pub url: String, pub method: String, pub auth_scope: VecString, pub region: String, pub health_status: HealthStatus, pub error_rate_5m: f64, } #[derive(Deserialize, Serialize, Clone, Debug)] pub enum HealthStatus { Healthy, Unhealthy, Degraded, } // 工具路由逻辑 pub async fn route_tool( task: TaskStep, consul_client: ConsulClient, ) - ResultToolInvocation, RoutingError { // 1. 查询 Consul 获取健康工具列表 let services consul_client .get_services_with_tag(tool) .await?; // 2. 过滤auth_scope 包含 task.action 所需权限且 error_rate_5m 0.1 let candidates: VecToolMetadata services .into_iter() .filter(|s| { s.auth_scope.contains(task.action) s.error_rate_5m 0.1 s.health_status HealthStatus::Healthy }) .collect(); if candidates.is_empty() { return Err(RoutingError::NoHealthyTool); } // 3. 按 region 路由优先同区域 let target_tool candidates .into_iter() .find(|s| s.region cn-east) .unwrap_or_else(|| candidates[0].clone()); // 4. 参数校验用 jsonschema-rs 验证 params 符合工具 OpenAPI let schema fetch_openapi_schema(target_tool.url).await?; let params_json serde_json::to_value(task.params)?; let compiled_schema jsonschema::JSONSchema::compile(schema)?; if !compiled_schema.is_valid(params_json) { return Err(RoutingError::InvalidParams); } Ok(ToolInvocation { url: target_tool.url, method: target_tool.method, headers: build_auth_headers(target_tool), body: params_json, }) }关键jsonschema-rs的is_valid()是零分配的比 Python 的jsonschema.validate()快 3 倍。Consul 查询用异步reqwest避免阻塞线程。整个路由过程在 5ms 内完成而 Python 版本平均 42ms主要耗在 JSON 解析和 schema 编译。3.5 执行调度决策点用 tokio::sync::Semaphore 实现精确并发控制Rust 的tokio::sync::Semaphore是真正的无锁并发控制Python 的asyncio.Semaphore底层仍是锁// 全局并发信号量初始化为 100 lazy_static::lazy_static! { static ref CONCURRENCY_SEM: Semaphore Semaphore::new(100); } // 执行调度函数 pub async fn schedule_task( task: TaskStep, tool_invocation: ToolInvocation, ) - ResultTaskResult, ExecutionError { // 1. 获取信号量带超时防饿死 let permit CONCURRENCY_SEM .acquire_owned() .await .map_err(|_| ExecutionError::SemaphoreTimeout)?; // 2. 设置 deadline context let deadline Instant::now() Duration::from_millis(task.deadline_ms); let (tx, mut rx) mpsc::channel::TaskResult(1); // 3. 启动异步任务 tokio::spawn(async move { // 用 reqwest 调用工具 let client reqwest::Client::new(); let response tokio::time::timeout( Duration::from_millis(task.deadline_ms), client.request( tool_invocation.method.parse().unwrap(), tool_invocation.url, ) .json(tool_invocation.body) .send(), ) .await; let result match response { Ok(Ok(res)) { let status res.status(); let body res.text().await.unwrap_or_default(); TaskResult { status, body, task_id: task.task_id } } Ok(Err(e)) TaskResult { status: 0, body: format!(HTTP error: {}, e), task_id: task.task_id }, Err(_) TaskResult { status: 0, body: timeout.to_string(), task_id: task.task_id }, }; let _ tx.send(result).await; // 忽略 send 失败receiver 可能已 drop }); // 4. 等待结果或超时 match tokio::time::timeout( Duration::from_millis(task.deadline_ms), rx.recv(), ) .await { Ok(Some(result)) Ok(result), Ok(None) Err(ExecutionError::ChannelClosed), Err(_) Err(ExecutionError::Timeout), } }优势Semaphore::acquire_owned()返回OwnedSemaphorePermit它实现了Droptrait离开作用域自动释放许可不可能出现忘记 release 的 bug。Python 的asyncio.Semaphore.acquire()必须手动release()漏掉就死锁。tokio::time::timeout是真正的异步超时不阻塞线程。3.6 容错回滚决策点用事务 幂等 Key 实现原子回滚Rust 的sqlx::Transactionuuid保证回滚的原子性// 回滚函数在一个 DB 事务中完成 pub async fn rollback_task( task_id: str, db_pool: PgPool, ) - Result(), RollbackError { let mut tx db_pool.begin().await?; // 1. 查询失败任务详情 let failed_task sqlx::query( SELECT intent, action, params FROM tasks WHERE task_id $1 ) .bind(task_id) .fetch_one(mut *tx) .await?; // 2. 构建幂等 key防止重复回滚 let idempotency_key format!(rollback_{}_{}, task_id, Uuid::new_v4()); // 3. 执行回滚工具如 cancel_reservation let rollback_result call_rollback_tool(failed_task, idempotency_key).await?; // 4. 记录回滚日志在同一个事务中 sqlx::query( INSERT INTO rollback_logs (task_id, idempotency_key, status, timestamp) VALUES ($1, $2, $3, NOW()) ) .bind(task_id)