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（用于时效判断）
• 决策逻辑：
  1. 校验user_id是否在白名单（防恶意刷请求）
  2. 检查session_id是否匹配当前活跃会话（防会话劫持）
  3. 解析自然语言中的时间表达式（如“昨天”、“下周三”），转换为 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（变更标记位）
• 决策逻辑：
  1. 读取当前状态快照（Redis Hash 或 PostgreSQL 行）
  2. 比较version，若客户端传入的version小于服务端，拒绝更新（防脏写）
  3. 根据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（如用户常驻城市、历史偏好）
• 决策逻辑：
  1. 匹配 intent 到预定义的任务模板（如book_flight模板含 4 个原子任务：check_availability→validate_passenger_info→reserve_seat→send_confirmation）
  2. 检查每个任务的前置条件（如reserve_seat需check_availability返回status: "available"）
  3. 为每个任务生成唯一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设 5s（API 快），send_confirmation设 60s（含邮件发送延迟），这直接影响整个链路的超时熔断策略。

2.4 工具路由决策点：不是“调用”，而是“精准匹配”

工具不是越多越好，而是每个工具必须有明确的能力边界声明。路由决策点要解决：当前任务，该调用哪个工具？调用参数是否合规？工具是否处于可用状态？

• 触发条件：任务队列中取出下一个任务时
• 输入约束：任务对象必须含action（如check_availability）、params（如{flight_no: "CA123"}）、tool_requirements（如{"auth_scope": "flight_read", "region": "CN"}）
• 决策逻辑：
  1. 查询工具注册中心（Consul 或 Kubernetes Service），筛选出status: "healthy"且metadata.auth_scope包含flight_read的工具实例
  2. 校验params是否符合该工具的 OpenAPI Schema（用 jsonschema 库实时验证）
  3. 若有多个健康实例，按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}）
• 决策逻辑：
  1. 检查全局并发数（Redis INCR），若超限（如 > 100），进入等待队列
  2. 启动 goroutine（Go）或 asyncio task（Python），设置deadline超时（用 context.WithDeadline）
  3. 执行中监控资源消耗，若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（失败前的状态）
• 决策逻辑：
  1. 查找该任务所属的原子任务链（如book_flight链）
  2. 执行预定义的回滚脚本（如reserve_seat失败，则调用cancel_reservation工具）
  3. 若回滚也失败，启动“状态修复模式”：用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”）
• 决策逻辑：
  1. 用正则和规则引擎校验final_output（如价格字段price > 0 and price < 100000）
  2. 调用 PII 识别模型扫描敏感词（身份证号、手机号、银行卡号），发现则脱敏（138****1234）
  3. 根据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()解析时间，平均耗时 12ms；Rust 版本用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) -> Result<DateTime<Utc>, 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 状态建模决策点：用 Arc<Mutex<>> + PostgreSQL 实现强一致快照

Rust 的Arc<Mutex<>>是线程安全状态管理的基石，但必须配合数据库保证持久化：

// 状态快照结构体，#[derive(Serialize, Deserialize)] 支持 serde_json #[derive(Serialize, Deserialize, Clone, Debug)] pub struct AgentState { pub version: u64, pub last_modified: i64, pub dirty_flags: Vec<String>, pub data: StateData, } #[derive(Serialize, Deserialize, Clone, Debug)] pub struct StateData { pub user_profile: UserProfile, pub cart_items: Vec<CartItem>, pub pending_actions: Vec<PendingAction>, } // 状态管理器：封装读写逻辑 pub struct StateManager { db_pool: PgPool, // SQLx PostgreSQL 连接池 cache: Arc<Mutex<HashMap<String, AgentState>>>, // 内存缓存 } impl StateManager { // 读取状态：先查缓存，缓存未命中则查 DB，查完写回缓存 pub async fn get_state(&self, session_id: &str) -> Result<AgentState, StateError> { // 1. 尝试从内存缓存读取（Arc<Mutex<>> 确保线程安全） { 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(()) } }

关键点：Arc<Mutex<>>用于内存缓存的线程安全访问，但绝不依赖它做持久化。所有状态变更必须落库，缓存只是加速层。update_state中的乐观锁 SQL 是强一致保障，rows_affected() == 0即表示版本冲突，必须由上层逻辑处理重试。Python 的threading.Lock无法提供这种细粒度的、与 DB 事务联动的锁语义。

3.3 目标拆解决策点：用 YAML 模板 + schemars 生成强类型任务链

Python 版本用json.loads()解析任务模板，类型不安全；Rust 版本用serde_yaml+schemars生成 OpenAPI Schema，实现编译期校验：

// 定义任务模板结构（YAML 文件映射） #[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] pub struct TaskTemplate { pub intent: String, pub steps: Vec<TaskStep>, pub timeout_ms: u64, } #[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)] pub struct TaskStep { pub action: String, pub params: HashMap<String, String>, pub requires: Vec<String>, // 前置条件 pub deadline_ms: u64, } // 加载模板：编译期保证 YAML 结构合法 pub fn load_task_template(intent: &str) -> Result<TaskTemplate, 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() -> Result<String, 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 的reqwest+jsonschema-rs组合，比 Python 的requests+jsonschema更轻量、更可靠：

// 工具元数据结构 #[derive(Deserialize, Serialize, Clone, Debug)] pub struct ToolMetadata { pub name: String, pub url: String, pub method: String, pub auth_scope: Vec<String>, 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, ) -> Result<ToolInvocation, 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: Vec<ToolMetadata> = 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, ) -> Result<TaskResult, 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::Transaction+uuid保证回滚的原子性：

// 回滚函数：在一个 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)