1. 问题背景与核心痛点拆解1.1 一个看似简单却频繁翻车的场景但凡做过 Agent 系统集成的人大概率都遇到过这种让人头皮发麻的情况Agent 调用外部工具去创建工单接口返回超时Agent 按照重试策略又发了一次请求结果系统里凭空多出两条一模一样的工单。用户投诉、运维排查、数据清洗一连串麻烦接踵而至。这个问题的本质其实不是超时本身而是超时之后 Agent 无法判断那次调用到底成功没成功。这是分布式系统里的经典难题——调用方和被调用方之间的状态出现了认知差。网络超时只代表我没在约定时间内收到响应它绝不等于对方没执行。这两件事之间的鸿沟就是重复创建工单的温床。我在实际项目里见过太多团队一开始不重视这个问题觉得超时重试嘛加个 try-catch 就行了结果上线后工单系统里重复数据堆积客服团队每天要手动合并几十条重复工单苦不堪言。所以这篇内容我想把这个问题从头到尾拆透给出几套能直接落地的方案。1.2 为什么 Agent 场景比普通服务调用更棘手普通微服务之间的调用重复请求问题相对好解因为调用链路是确定的、可控的。但 Agent 场景有几个额外的复杂性第一Agent 的决策是动态的。它可能因为超时触发重试也可能因为推理链条变化而重新规划动作甚至可能在多轮对话中忘记自己已经调用过一次。这种不确定性让幂等控制变得更难。第二工具调用的边界模糊。Agent 调用的工具可能是内部 API也可能是第三方服务有些第三方接口根本不支持幂等键你没法从服务端做去重。第三超时判定本身就不精确。Agent 框架里的超时时间设置、网络抖动、工具执行耗时波动都会让到底超没超时变成一个概率问题。理解了这些复杂性我们才能有针对性地设计方案而不是套用一个通用模板就完事。1.3 本文要解决的核心问题清单把问题聚焦一下我们需要回答的是这几个具体问题超时发生后Agent 如何判断工具是否真的执行了如果无法判断怎样保证重复调用不会产生重复副作用幂等键应该在哪里生成、怎么传递、由谁校验工单系统本身不支持幂等时Agent 侧能做什么补偿多轮对话、多 Agent 协作场景下状态怎么保持一致下面我会围绕这几个问题逐层展开方案设计和实操细节。2. 核心原理幂等性与状态确认的底层逻辑2.1 超时的本质是一次状态未知先把概念理清楚。一次工具调用从 Agent 发出请求到收到响应中间会经历请求发出、网络传输、服务端接收、服务端处理、响应返回、Agent 接收。超时可能发生在任何一个环节。如果超时发生在服务端处理之前那这次调用确实没生效重试是安全的。但如果超时发生在服务端处理之后、响应返回之前那服务端已经创建了工单只是 Agent 没收到确认。这时候重试就会产生重复。问题的关键在于Agent 无法从超时这个信号本身区分这两种情况。这就是为什么我们必须引入额外的机制来消除这种不确定性。2.2 幂等键让重复请求看起来像同一个幂等键Idempotency Key是解决这个问题的核心武器。它的思路很朴素给每一次逻辑上的创建操作分配一个唯一标识服务端收到请求时先检查这个标识是否处理过处理过就直接返回上次的结果不再重复执行。这里有个关键点容易被忽略幂等键必须在逻辑操作层面唯一而不是在请求层面唯一。什么意思如果 Agent 因为超时重试了三次这三次请求应该携带同一个幂等键因为它们代表的是同一个逻辑操作。如果每次重试都生成新的幂等键那幂等机制就完全失效了。所以幂等键的生成时机非常讲究——它应该在 Agent 决定要创建一个工单的那一刻生成然后在整个重试生命周期内保持不变。2.3 状态确认的三种技术路径除了幂等键还有几种确认状态的思路各有适用场景路径一查询确认Query-before-Retry。超时后不急着重试先调用一个查询接口看看目标资源是否已经存在。比如用业务唯一字段如用户ID问题类型时间窗口去查工单列表如果查到了就说明上次调用成功了。路径二服务端幂等Server-side Idempotency。由工单系统自己维护幂等键到结果的映射Agent 只管带着幂等键重试服务端负责去重。这是最干净的方案但需要工单系统配合改造。路径三本地状态机Local State Machine。Agent 侧维护一个调用状态记录把已发起已确认已失败等状态持久化重试前先查本地状态。这个方案不依赖服务端改造但需要处理本地状态与服务端实际状态不一致的情况。实际项目里这三种路径往往是组合使用的。下面我会详细讲每种方案的落地细节。3. 方案设计与选型从简单到完备的四层防御3.1 第一层幂等键的生成与传递规范幂等键怎么生成看似简单其实有很多坑。我见过有人用 UUID有人用时间戳有人用请求内容的哈希每种都有问题。UUID 的问题如果 Agent 在重试时重新生成 UUID幂等就失效了。所以 UUID 必须在逻辑操作开始时生成一次然后缓存起来供重试使用。时间戳的问题精度不够会碰撞精度太高又可能因为时钟回拨出问题。而且时间戳无法表达这是同一个逻辑操作。内容哈希的问题如果工单内容里有时间字段每次重试内容都不同哈希就变了。而且哈希计算本身有性能开销。我的建议是采用组合键业务前缀 会话ID 操作序号 随机因子。比如ticket:session_abc123:op_001:7f3a。这样既保证了唯一性又能从键本身看出它属于哪个会话的哪个操作排查问题时非常方便。传递方式上优先放在 HTTP Header 里如X-Idempotency-Key而不是放在 body 里。因为 Header 不参与业务逻辑解析不容易被误改而且很多网关和中间件能直接识别和处理。import uuid import hashlib class IdempotencyKeyGenerator: def __init__(self, session_id): self.session_id session_id self.op_counter 0 self._cache {} def get_key(self, operation_name): # 同一个逻辑操作复用同一个键 cache_key f{self.session_id}:{operation_name} if cache_key in self._cache: return self._cache[cache_key] self.op_counter 1 random_part uuid.uuid4().hex[:8] key fticket:{self.session_id}:op_{self.op_counter:03d}:{random_part} self._cache[cache_key] key return key这段代码的关键在于_cache——它保证了同一个逻辑操作在重试时拿到的是同一个键。这个缓存需要持久化不能只放在内存里否则 Agent 进程重启后缓存丢失幂等就断了。3.2 第二层服务端幂等表的落地实现如果工单系统能改造最彻底的方案是在服务端建一张幂等表。表结构大概是这样字段名类型说明idempotency_keyVARCHAR(128)幂等键唯一索引request_hashVARCHAR(64)请求内容哈希用于检测键复用但内容不同的情况statusTINYINT0处理中1成功2失败result_payloadTEXT成功时的返回结果created_atDATETIME创建时间expires_atDATETIME过期时间用于清理处理逻辑是收到请求后先尝试插入幂等记录。如果插入成功说明是首次请求继续执行业务逻辑如果插入冲突说明是重复请求查询已有记录的状态——如果状态是成功直接返回缓存的 result_payload如果状态是处理中返回一个处理中的响应让 Agent 稍后重试如果状态是失败根据策略决定是否允许重试。这里有个细节要注意request_hash 的校验。如果同一个幂等键配上了不同的请求内容这通常意味着 Agent 侧出了 bug比如键复用逻辑写错了。这时候应该拒绝请求并告警而不是默默执行否则会产生难以排查的数据问题。注意幂等表的过期时间设置很关键。设太短重试窗口内记录就被清理了幂等失效设太长表会无限膨胀。一般建议设置为业务最大重试窗口的 2-3 倍比如重试窗口是 5 分钟过期时间设 15 分钟。3.3 第三层Agent 侧的状态机与重试策略服务端改造往往推进缓慢所以 Agent 侧必须有一套自己的防御机制。核心是维护一个调用状态机PENDING - SENT - CONFIRMED \- TIMEOUT - RETRYING - CONFIRMED \- FAILED每次调用工具前先把状态记为 SENT 并持久化。收到成功响应后改为 CONFIRMED。超时后改为 TIMEOUT进入重试流程。重试前先检查状态如果已经是 CONFIRMED 就不再重试。重试策略上我强烈建议用指数退避 抖动而不是固定间隔。因为固定间隔重试容易造成重试风暴尤其在服务端已经压力很大的时候密集重试会让情况更糟。import random import time def retry_with_backoff(max_retries3, base_delay1.0, max_delay30.0): for attempt in range(max_retries): try: result call_tool() return result except TimeoutError: if attempt max_retries - 1: raise # 指数退避 抖动 delay min(base_delay * (2 ** attempt), max_delay) jitter random.uniform(0, delay * 0.5) time.sleep(delay jitter)抖动的作用是打散多个 Agent 实例的重试时间避免它们同时冲击服务端。这个细节在单机测试时看不出效果但生产环境多实例部署时能救命。3.4 第四层查询确认与补偿对账即使前面三层都做了仍然可能有漏网之鱼。比如幂等键在传输中丢失、服务端幂等表被误清理等。所以还需要一层兜底——定期对账。对账的逻辑是Agent 记录所有状态为 TIMEOUT 且最终未确认的调用定期比如每 5 分钟去工单系统查询这些调用对应的工单是否存在。查询依据是业务唯一字段比如会话ID 操作序号组合。如果查到了说明实际创建成功了把本地状态改为 CONFIRMED如果没查到说明确实没创建可以安全重试或标记为失败。这个对账机制是最后一道防线能兜住前面所有环节的意外情况。虽然增加了系统复杂度但对于工单这种重复创建代价很高的场景我认为是值得的。4. 实操落地从零搭建一套防重复创建机制4.1 环境准备与依赖说明假设我们用 Python 做一个 Agent 工具调用的封装层需要这些依赖pip install requests redis tenacityrequests发 HTTP 请求redis做幂等键缓存和状态存储tenacity重试库比手写重试逻辑更健壮Redis 在这里扮演两个角色一是缓存幂等键保证重试时键不变二是存储调用状态机支持多实例共享状态。如果你的部署环境没有 Redis也可以用本地 SQLite 替代但多实例场景下会有状态不一致问题。4.2 幂等键的持久化与复用先解决幂等键的持久化问题。前面代码里的内存缓存不够用我们改成 Redisimport redis import uuid class PersistentIdempotencyKey: def __init__(self, redis_client, session_id, ttl900): self.redis redis_client self.session_id session_id self.ttl ttl def get_or_create(self, operation_name): cache_key fidem:{self.session_id}:{operation_name} existing self.redis.get(cache_key) if existing: return existing.decode() new_key fticket:{self.session_id}:{uuid.uuid4().hex[:12]} # setnx 保证并发下只有一个能写入 if self.redis.setnx(cache_key, new_key): self.redis.expire(cache_key, self.ttl) return new_key else: # 并发情况下别人先写入了读回来 return self.redis.get(cache_key).decode()这里用setnx而不是set是为了处理并发场景。如果两个线程同时为同一个操作生成键setnx保证只有一个成功另一个读到已存在的键。这个细节在单线程测试时不会暴露但生产环境多线程调用时很关键。TTL 设置为 900 秒15 分钟比最大重试窗口长。这个值需要根据你的业务重试策略调整原则是覆盖整个重试生命周期 一定的安全余量。4.3 工具调用封装层的完整实现把幂等键、状态机、重试策略整合到一个封装层里import json import time import requests from enum import Enum class CallState(Enum): PENDING pending SENT sent CONFIRMED confirmed TIMEOUT timeout FAILED failed class TicketToolClient: def __init__(self, redis_client, base_url, session_id): self.redis redis_client self.base_url base_url self.session_id session_id self.key_gen PersistentIdempotencyKey(redis_client, session_id) def _state_key(self, op_name): return fstate:{self.session_id}:{op_name} def _set_state(self, op_name, state, extraNone): data {state: state.value, ts: time.time()} if extra: data.update(extra) self.redis.setex( self._state_key(op_name), 900, json.dumps(data) ) def _get_state(self, op_name): raw self.redis.get(self._state_key(op_name)) return json.loads(raw) if raw else None def create_ticket(self, op_name, ticket_data, max_retries3): # 检查是否已经确认成功 state self._get_state(op_name) if state and state[state] CallState.CONFIRMED.value: return state.get(result) idem_key self.key_gen.get_or_create(op_name) self._set_state(op_name, CallState.SENT) for attempt in range(max_retries): try: resp requests.post( f{self.base_url}/tickets, jsonticket_data, headers{X-Idempotency-Key: idem_key}, timeout10 ) if resp.status_code 200: result resp.json() self._set_state( op_name, CallState.CONFIRMED, {result: result} ) return result elif resp.status_code 409: # 服务端返回冲突说明幂等键已存在 # 查询已有结果 existing self._query_by_idem_key(idem_key) self._set_state( op_name, CallState.CONFIRMED, {result: existing} ) return existing except requests.Timeout: self._set_state(op_name, CallState.TIMEOUT) if attempt max_retries - 1: delay min(1.0 * (2 ** attempt), 30.0) time.sleep(delay random.uniform(0, delay * 0.5)) continue # 所有重试都失败标记为失败 self._set_state(op_name, CallState.FAILED) raise Exception(f创建工单失败操作: {op_name}) def _query_by_idem_key(self, idem_key): resp requests.get( f{self.base_url}/tickets/by-idem-key, params{key: idem_key}, timeout10 ) return resp.json()这段代码有几个设计要点值得说明状态检查前置。进入重试循环前先检查状态如果已经 CONFIRMED 就直接返回缓存结果。这避免了明明已经成功却因为状态没同步而重复调用的情况。409 冲突的处理。如果服务端实现了幂等重复请求会返回 409。这时候不能简单当作失败而应该去查询已有结果并返回。这是很多实现容易漏掉的一环。状态持久化的时机。每次状态变更都立即写入 Redis而不是等整个流程结束再写。这样即使进程崩溃重启后也能从 Redis 恢复状态。4.4 对账补偿任务的实现最后加上对账任务作为兜底def reconcile_pending_calls(redis_client, tool_client, interval300): 定期对账处理状态为 TIMEOUT 的调用 while True: # 扫描所有 TIMEOUT 状态的调用 pattern state:* for key in redis_client.scan_iter(matchpattern): raw redis_client.get(key) if not raw: continue state json.loads(raw) if state[state] ! CallState.TIMEOUT.value: continue # 解析出 session_id 和 op_name parts key.decode().split(:) session_id, op_name parts[1], parts[2] # 查询工单是否实际存在 exists query_ticket_exists(session_id, op_name) if exists: state[state] CallState.CONFIRMED.value state[result] exists redis_client.setex(key, 900, json.dumps(state)) time.sleep(interval)对账任务建议独立部署不要和 Agent 主进程耦合。因为对账是低频操作独立部署可以避免它影响主流程的性能。同时对账任务本身也要做幂等避免多个实例重复对账。提示对账的查询依据必须是业务唯一字段不能依赖幂等键。因为幂等键可能因为缓存过期而丢失但业务字段如会话ID操作序号是稳定的。5. 常见问题与排查技巧实录5.1 幂等键失效的几种典型场景实际排查中幂等键失效是最常见的问题。我整理了几种典型场景和排查方法现象可能原因排查方法重试后仍产生重复工单重试时生成了新幂等键检查键生成逻辑确认缓存是否生效幂等键冲突但内容不同键复用逻辑有 bug检查 request_hash 校验是否开启服务端不识别幂等键Header 名不一致或网关过滤抓包确认 Header 是否透传幂等记录被提前清理TTL 设置过短对比 TTL 和最大重试窗口排查时我习惯先看日志里每次请求携带的幂等键是否一致。如果不一致问题就在 Agent 侧如果一致但服务端还是重复创建问题就在服务端幂等实现。5.2 超时时间设置的取舍超时时间设多少合适这个问题没有标准答案但有几个原则不能设太短。如果工具正常执行需要 3 秒超时设 2 秒那大部分请求都会超时重试率飙升。一般建议超时时间设为 P99 执行时间的 2-3 倍。不能设太长。超时太长会导致 Agent 长时间阻塞影响整体响应速度。而且超时太长时用户可能已经等不及了。分层设置。连接超时和读取超时要分开设。连接超时一般设短一点如 3 秒因为连接建立通常很快读取超时设长一点如 10 秒给业务处理留足时间。# 分层超时设置示例 timeout (3.0, 10.0) # (连接超时, 读取超时) requests.post(url, jsondata, timeouttimeout)5.3 多 Agent 协作时的状态同步如果系统里有多个 Agent 协作状态同步会更复杂。比如 Agent A 创建工单超时了Agent B 不知道又去创建了一次。解决思路是把状态存储从 Agent 本地提升到共享层。前面用 Redis 存状态就是为了这个——所有 Agent 共享同一个 Redis状态变更对所有 Agent 可见。但共享状态也有新问题并发读写。两个 Agent 同时读到 TIMEOUT 状态同时决定重试还是会重复。这时候需要加分布式锁def retry_with_lock(redis_client, op_name, lock_timeout30): lock_key flock:{op_name} # 尝试获取锁 acquired redis_client.set(lock_key, 1, nxTrue, exlock_timeout) if not acquired: # 别人正在处理等待后查询结果 time.sleep(2) return query_result(op_name) try: # 持有锁执行重试 return do_retry(op_name) finally: redis_client.delete(lock_key)锁的粒度要控制好太粗会影响并发太细又起不到保护作用。一般以单个逻辑操作为粒度比较合适。5.4 几个我踩过的坑坑一以为超时就是失败。早期项目里我把超时直接当失败处理结果服务端实际创建成功了Agent 又重试产生重复。后来才明白超时是状态未知必须用幂等或查询来确认。坑二幂等键放在 body 里。有次对接第三方接口幂等键放在 JSON body 里结果对方的网关做了 body 重写把幂等键字段弄丢了。后来改成放 Header 才稳定。坑三忽略时钟问题。用时间戳做幂等键的一部分结果服务器时钟回拨生成了重复的键。后来改用 UUID 业务字段组合彻底避开时钟依赖。坑四对账任务没做幂等。对账任务部署了两个实例同时扫描到同一条 TIMEOUT 记录同时去查询和更新虽然不会产生重复工单但浪费了资源。后来给对账任务也加了分布式锁。5.5 性能与可靠性的平衡最后聊聊性能。加了幂等、状态机、对账这些机制后系统复杂度上去了性能也会受影响。怎么平衡我的经验是分级处理。对于重复创建代价低的工具比如查询类可以不做幂等直接重试。对于重复创建代价高的工具比如创建工单、扣款、发消息才上完整的幂等机制。这样既保证了关键路径的可靠性又避免了全链路过度设计。另外幂等键的生成和状态存储尽量用本地缓存 异步持久化减少同步 IO 的开销。但要注意异步持久化有丢失风险对于关键操作还是要同步写。这套机制我在几个项目里落地过实测下来能把重复创建率从千分之几降到几乎为零。当然没有银弹具体方案还是要根据你的工单系统能力、Agent 框架特性、业务容忍度来调整。核心思路就是那句话超时不是失败是状态未知而未知必须用幂等和确认来消除。 
长河君 · 2026/10/14 15:04:14 · 阅读 1.2k Agent工具调用超时重试导致重复创建工单的幂等解决方案
1. 问题背景与核心痛点拆解1.1 一个看似简单却频繁翻车的场景但凡做过 Agent 系统集成的人大概率都遇到过这种让人头皮发麻的情况Agent 调用外部工具去创建工单接口返回超时Agent 按照重试策略又发了一次请求结果系统里凭空多出两条一模一样的工单。用户投诉、运维排查、数据清洗一连串麻烦接踵而至。这个问题的本质其实不是超时本身而是超时之后 Agent 无法判断那次调用到底成功没成功。这是分布式系统里的经典难题——调用方和被调用方之间的状态出现了认知差。网络超时只代表我没在约定时间内收到响应它绝不等于对方没执行。这两件事之间的鸿沟就是重复创建工单的温床。我在实际项目里见过太多团队一开始不重视这个问题觉得超时重试嘛加个 try-catch 就行了结果上线后工单系统里重复数据堆积客服团队每天要手动合并几十条重复工单苦不堪言。所以这篇内容我想把这个问题从头到尾拆透给出几套能直接落地的方案。1.2 为什么 Agent 场景比普通服务调用更棘手普通微服务之间的调用重复请求问题相对好解因为调用链路是确定的、可控的。但 Agent 场景有几个额外的复杂性第一Agent 的决策是动态的。它可能因为超时触发重试也可能因为推理链条变化而重新规划动作甚至可能在多轮对话中忘记自己已经调用过一次。这种不确定性让幂等控制变得更难。第二工具调用的边界模糊。Agent 调用的工具可能是内部 API也可能是第三方服务有些第三方接口根本不支持幂等键你没法从服务端做去重。第三超时判定本身就不精确。Agent 框架里的超时时间设置、网络抖动、工具执行耗时波动都会让到底超没超时变成一个概率问题。理解了这些复杂性我们才能有针对性地设计方案而不是套用一个通用模板就完事。1.3 本文要解决的核心问题清单把问题聚焦一下我们需要回答的是这几个具体问题超时发生后Agent 如何判断工具是否真的执行了如果无法判断怎样保证重复调用不会产生重复副作用幂等键应该在哪里生成、怎么传递、由谁校验工单系统本身不支持幂等时Agent 侧能做什么补偿多轮对话、多 Agent 协作场景下状态怎么保持一致下面我会围绕这几个问题逐层展开方案设计和实操细节。2. 核心原理幂等性与状态确认的底层逻辑2.1 超时的本质是一次状态未知先把概念理清楚。一次工具调用从 Agent 发出请求到收到响应中间会经历请求发出、网络传输、服务端接收、服务端处理、响应返回、Agent 接收。超时可能发生在任何一个环节。如果超时发生在服务端处理之前那这次调用确实没生效重试是安全的。但如果超时发生在服务端处理之后、响应返回之前那服务端已经创建了工单只是 Agent 没收到确认。这时候重试就会产生重复。问题的关键在于Agent 无法从超时这个信号本身区分这两种情况。这就是为什么我们必须引入额外的机制来消除这种不确定性。2.2 幂等键让重复请求看起来像同一个幂等键Idempotency Key是解决这个问题的核心武器。它的思路很朴素给每一次逻辑上的创建操作分配一个唯一标识服务端收到请求时先检查这个标识是否处理过处理过就直接返回上次的结果不再重复执行。这里有个关键点容易被忽略幂等键必须在逻辑操作层面唯一而不是在请求层面唯一。什么意思如果 Agent 因为超时重试了三次这三次请求应该携带同一个幂等键因为它们代表的是同一个逻辑操作。如果每次重试都生成新的幂等键那幂等机制就完全失效了。所以幂等键的生成时机非常讲究——它应该在 Agent 决定要创建一个工单的那一刻生成然后在整个重试生命周期内保持不变。2.3 状态确认的三种技术路径除了幂等键还有几种确认状态的思路各有适用场景路径一查询确认Query-before-Retry。超时后不急着重试先调用一个查询接口看看目标资源是否已经存在。比如用业务唯一字段如用户ID问题类型时间窗口去查工单列表如果查到了就说明上次调用成功了。路径二服务端幂等Server-side Idempotency。由工单系统自己维护幂等键到结果的映射Agent 只管带着幂等键重试服务端负责去重。这是最干净的方案但需要工单系统配合改造。路径三本地状态机Local State Machine。Agent 侧维护一个调用状态记录把已发起已确认已失败等状态持久化重试前先查本地状态。这个方案不依赖服务端改造但需要处理本地状态与服务端实际状态不一致的情况。实际项目里这三种路径往往是组合使用的。下面我会详细讲每种方案的落地细节。3. 方案设计与选型从简单到完备的四层防御3.1 第一层幂等键的生成与传递规范幂等键怎么生成看似简单其实有很多坑。我见过有人用 UUID有人用时间戳有人用请求内容的哈希每种都有问题。UUID 的问题如果 Agent 在重试时重新生成 UUID幂等就失效了。所以 UUID 必须在逻辑操作开始时生成一次然后缓存起来供重试使用。时间戳的问题精度不够会碰撞精度太高又可能因为时钟回拨出问题。而且时间戳无法表达这是同一个逻辑操作。内容哈希的问题如果工单内容里有时间字段每次重试内容都不同哈希就变了。而且哈希计算本身有性能开销。我的建议是采用组合键业务前缀 会话ID 操作序号 随机因子。比如ticket:session_abc123:op_001:7f3a。这样既保证了唯一性又能从键本身看出它属于哪个会话的哪个操作排查问题时非常方便。传递方式上优先放在 HTTP Header 里如X-Idempotency-Key而不是放在 body 里。因为 Header 不参与业务逻辑解析不容易被误改而且很多网关和中间件能直接识别和处理。import uuid import hashlib class IdempotencyKeyGenerator: def __init__(self, session_id): self.session_id session_id self.op_counter 0 self._cache {} def get_key(self, operation_name): # 同一个逻辑操作复用同一个键 cache_key f{self.session_id}:{operation_name} if cache_key in self._cache: return self._cache[cache_key] self.op_counter 1 random_part uuid.uuid4().hex[:8] key fticket:{self.session_id}:op_{self.op_counter:03d}:{random_part} self._cache[cache_key] key return key这段代码的关键在于_cache——它保证了同一个逻辑操作在重试时拿到的是同一个键。这个缓存需要持久化不能只放在内存里否则 Agent 进程重启后缓存丢失幂等就断了。3.2 第二层服务端幂等表的落地实现如果工单系统能改造最彻底的方案是在服务端建一张幂等表。表结构大概是这样字段名类型说明idempotency_keyVARCHAR(128)幂等键唯一索引request_hashVARCHAR(64)请求内容哈希用于检测键复用但内容不同的情况statusTINYINT0处理中1成功2失败result_payloadTEXT成功时的返回结果created_atDATETIME创建时间expires_atDATETIME过期时间用于清理处理逻辑是收到请求后先尝试插入幂等记录。如果插入成功说明是首次请求继续执行业务逻辑如果插入冲突说明是重复请求查询已有记录的状态——如果状态是成功直接返回缓存的 result_payload如果状态是处理中返回一个处理中的响应让 Agent 稍后重试如果状态是失败根据策略决定是否允许重试。这里有个细节要注意request_hash 的校验。如果同一个幂等键配上了不同的请求内容这通常意味着 Agent 侧出了 bug比如键复用逻辑写错了。这时候应该拒绝请求并告警而不是默默执行否则会产生难以排查的数据问题。注意幂等表的过期时间设置很关键。设太短重试窗口内记录就被清理了幂等失效设太长表会无限膨胀。一般建议设置为业务最大重试窗口的 2-3 倍比如重试窗口是 5 分钟过期时间设 15 分钟。3.3 第三层Agent 侧的状态机与重试策略服务端改造往往推进缓慢所以 Agent 侧必须有一套自己的防御机制。核心是维护一个调用状态机PENDING - SENT - CONFIRMED \- TIMEOUT - RETRYING - CONFIRMED \- FAILED每次调用工具前先把状态记为 SENT 并持久化。收到成功响应后改为 CONFIRMED。超时后改为 TIMEOUT进入重试流程。重试前先检查状态如果已经是 CONFIRMED 就不再重试。重试策略上我强烈建议用指数退避 抖动而不是固定间隔。因为固定间隔重试容易造成重试风暴尤其在服务端已经压力很大的时候密集重试会让情况更糟。import random import time def retry_with_backoff(max_retries3, base_delay1.0, max_delay30.0): for attempt in range(max_retries): try: result call_tool() return result except TimeoutError: if attempt max_retries - 1: raise # 指数退避 抖动 delay min(base_delay * (2 ** attempt), max_delay) jitter random.uniform(0, delay * 0.5) time.sleep(delay jitter)抖动的作用是打散多个 Agent 实例的重试时间避免它们同时冲击服务端。这个细节在单机测试时看不出效果但生产环境多实例部署时能救命。3.4 第四层查询确认与补偿对账即使前面三层都做了仍然可能有漏网之鱼。比如幂等键在传输中丢失、服务端幂等表被误清理等。所以还需要一层兜底——定期对账。对账的逻辑是Agent 记录所有状态为 TIMEOUT 且最终未确认的调用定期比如每 5 分钟去工单系统查询这些调用对应的工单是否存在。查询依据是业务唯一字段比如会话ID 操作序号组合。如果查到了说明实际创建成功了把本地状态改为 CONFIRMED如果没查到说明确实没创建可以安全重试或标记为失败。这个对账机制是最后一道防线能兜住前面所有环节的意外情况。虽然增加了系统复杂度但对于工单这种重复创建代价很高的场景我认为是值得的。4. 实操落地从零搭建一套防重复创建机制4.1 环境准备与依赖说明假设我们用 Python 做一个 Agent 工具调用的封装层需要这些依赖pip install requests redis tenacityrequests发 HTTP 请求redis做幂等键缓存和状态存储tenacity重试库比手写重试逻辑更健壮Redis 在这里扮演两个角色一是缓存幂等键保证重试时键不变二是存储调用状态机支持多实例共享状态。如果你的部署环境没有 Redis也可以用本地 SQLite 替代但多实例场景下会有状态不一致问题。4.2 幂等键的持久化与复用先解决幂等键的持久化问题。前面代码里的内存缓存不够用我们改成 Redisimport redis import uuid class PersistentIdempotencyKey: def __init__(self, redis_client, session_id, ttl900): self.redis redis_client self.session_id session_id self.ttl ttl def get_or_create(self, operation_name): cache_key fidem:{self.session_id}:{operation_name} existing self.redis.get(cache_key) if existing: return existing.decode() new_key fticket:{self.session_id}:{uuid.uuid4().hex[:12]} # setnx 保证并发下只有一个能写入 if self.redis.setnx(cache_key, new_key): self.redis.expire(cache_key, self.ttl) return new_key else: # 并发情况下别人先写入了读回来 return self.redis.get(cache_key).decode()这里用setnx而不是set是为了处理并发场景。如果两个线程同时为同一个操作生成键setnx保证只有一个成功另一个读到已存在的键。这个细节在单线程测试时不会暴露但生产环境多线程调用时很关键。TTL 设置为 900 秒15 分钟比最大重试窗口长。这个值需要根据你的业务重试策略调整原则是覆盖整个重试生命周期 一定的安全余量。4.3 工具调用封装层的完整实现把幂等键、状态机、重试策略整合到一个封装层里import json import time import requests from enum import Enum class CallState(Enum): PENDING pending SENT sent CONFIRMED confirmed TIMEOUT timeout FAILED failed class TicketToolClient: def __init__(self, redis_client, base_url, session_id): self.redis redis_client self.base_url base_url self.session_id session_id self.key_gen PersistentIdempotencyKey(redis_client, session_id) def _state_key(self, op_name): return fstate:{self.session_id}:{op_name} def _set_state(self, op_name, state, extraNone): data {state: state.value, ts: time.time()} if extra: data.update(extra) self.redis.setex( self._state_key(op_name), 900, json.dumps(data) ) def _get_state(self, op_name): raw self.redis.get(self._state_key(op_name)) return json.loads(raw) if raw else None def create_ticket(self, op_name, ticket_data, max_retries3): # 检查是否已经确认成功 state self._get_state(op_name) if state and state[state] CallState.CONFIRMED.value: return state.get(result) idem_key self.key_gen.get_or_create(op_name) self._set_state(op_name, CallState.SENT) for attempt in range(max_retries): try: resp requests.post( f{self.base_url}/tickets, jsonticket_data, headers{X-Idempotency-Key: idem_key}, timeout10 ) if resp.status_code 200: result resp.json() self._set_state( op_name, CallState.CONFIRMED, {result: result} ) return result elif resp.status_code 409: # 服务端返回冲突说明幂等键已存在 # 查询已有结果 existing self._query_by_idem_key(idem_key) self._set_state( op_name, CallState.CONFIRMED, {result: existing} ) return existing except requests.Timeout: self._set_state(op_name, CallState.TIMEOUT) if attempt max_retries - 1: delay min(1.0 * (2 ** attempt), 30.0) time.sleep(delay random.uniform(0, delay * 0.5)) continue # 所有重试都失败标记为失败 self._set_state(op_name, CallState.FAILED) raise Exception(f创建工单失败操作: {op_name}) def _query_by_idem_key(self, idem_key): resp requests.get( f{self.base_url}/tickets/by-idem-key, params{key: idem_key}, timeout10 ) return resp.json()这段代码有几个设计要点值得说明状态检查前置。进入重试循环前先检查状态如果已经 CONFIRMED 就直接返回缓存结果。这避免了明明已经成功却因为状态没同步而重复调用的情况。409 冲突的处理。如果服务端实现了幂等重复请求会返回 409。这时候不能简单当作失败而应该去查询已有结果并返回。这是很多实现容易漏掉的一环。状态持久化的时机。每次状态变更都立即写入 Redis而不是等整个流程结束再写。这样即使进程崩溃重启后也能从 Redis 恢复状态。4.4 对账补偿任务的实现最后加上对账任务作为兜底def reconcile_pending_calls(redis_client, tool_client, interval300): 定期对账处理状态为 TIMEOUT 的调用 while True: # 扫描所有 TIMEOUT 状态的调用 pattern state:* for key in redis_client.scan_iter(matchpattern): raw redis_client.get(key) if not raw: continue state json.loads(raw) if state[state] ! CallState.TIMEOUT.value: continue # 解析出 session_id 和 op_name parts key.decode().split(:) session_id, op_name parts[1], parts[2] # 查询工单是否实际存在 exists query_ticket_exists(session_id, op_name) if exists: state[state] CallState.CONFIRMED.value state[result] exists redis_client.setex(key, 900, json.dumps(state)) time.sleep(interval)对账任务建议独立部署不要和 Agent 主进程耦合。因为对账是低频操作独立部署可以避免它影响主流程的性能。同时对账任务本身也要做幂等避免多个实例重复对账。提示对账的查询依据必须是业务唯一字段不能依赖幂等键。因为幂等键可能因为缓存过期而丢失但业务字段如会话ID操作序号是稳定的。5. 常见问题与排查技巧实录5.1 幂等键失效的几种典型场景实际排查中幂等键失效是最常见的问题。我整理了几种典型场景和排查方法现象可能原因排查方法重试后仍产生重复工单重试时生成了新幂等键检查键生成逻辑确认缓存是否生效幂等键冲突但内容不同键复用逻辑有 bug检查 request_hash 校验是否开启服务端不识别幂等键Header 名不一致或网关过滤抓包确认 Header 是否透传幂等记录被提前清理TTL 设置过短对比 TTL 和最大重试窗口排查时我习惯先看日志里每次请求携带的幂等键是否一致。如果不一致问题就在 Agent 侧如果一致但服务端还是重复创建问题就在服务端幂等实现。5.2 超时时间设置的取舍超时时间设多少合适这个问题没有标准答案但有几个原则不能设太短。如果工具正常执行需要 3 秒超时设 2 秒那大部分请求都会超时重试率飙升。一般建议超时时间设为 P99 执行时间的 2-3 倍。不能设太长。超时太长会导致 Agent 长时间阻塞影响整体响应速度。而且超时太长时用户可能已经等不及了。分层设置。连接超时和读取超时要分开设。连接超时一般设短一点如 3 秒因为连接建立通常很快读取超时设长一点如 10 秒给业务处理留足时间。# 分层超时设置示例 timeout (3.0, 10.0) # (连接超时, 读取超时) requests.post(url, jsondata, timeouttimeout)5.3 多 Agent 协作时的状态同步如果系统里有多个 Agent 协作状态同步会更复杂。比如 Agent A 创建工单超时了Agent B 不知道又去创建了一次。解决思路是把状态存储从 Agent 本地提升到共享层。前面用 Redis 存状态就是为了这个——所有 Agent 共享同一个 Redis状态变更对所有 Agent 可见。但共享状态也有新问题并发读写。两个 Agent 同时读到 TIMEOUT 状态同时决定重试还是会重复。这时候需要加分布式锁def retry_with_lock(redis_client, op_name, lock_timeout30): lock_key flock:{op_name} # 尝试获取锁 acquired redis_client.set(lock_key, 1, nxTrue, exlock_timeout) if not acquired: # 别人正在处理等待后查询结果 time.sleep(2) return query_result(op_name) try: # 持有锁执行重试 return do_retry(op_name) finally: redis_client.delete(lock_key)锁的粒度要控制好太粗会影响并发太细又起不到保护作用。一般以单个逻辑操作为粒度比较合适。5.4 几个我踩过的坑坑一以为超时就是失败。早期项目里我把超时直接当失败处理结果服务端实际创建成功了Agent 又重试产生重复。后来才明白超时是状态未知必须用幂等或查询来确认。坑二幂等键放在 body 里。有次对接第三方接口幂等键放在 JSON body 里结果对方的网关做了 body 重写把幂等键字段弄丢了。后来改成放 Header 才稳定。坑三忽略时钟问题。用时间戳做幂等键的一部分结果服务器时钟回拨生成了重复的键。后来改用 UUID 业务字段组合彻底避开时钟依赖。坑四对账任务没做幂等。对账任务部署了两个实例同时扫描到同一条 TIMEOUT 记录同时去查询和更新虽然不会产生重复工单但浪费了资源。后来给对账任务也加了分布式锁。5.5 性能与可靠性的平衡最后聊聊性能。加了幂等、状态机、对账这些机制后系统复杂度上去了性能也会受影响。怎么平衡我的经验是分级处理。对于重复创建代价低的工具比如查询类可以不做幂等直接重试。对于重复创建代价高的工具比如创建工单、扣款、发消息才上完整的幂等机制。这样既保证了关键路径的可靠性又避免了全链路过度设计。另外幂等键的生成和状态存储尽量用本地缓存 异步持久化减少同步 IO 的开销。但要注意异步持久化有丢失风险对于关键操作还是要同步写。这套机制我在几个项目里落地过实测下来能把重复创建率从千分之几降到几乎为零。当然没有银弹具体方案还是要根据你的工单系统能力、Agent 框架特性、业务容忍度来调整。核心思路就是那句话超时不是失败是状态未知而未知必须用幂等和确认来消除。