Agent 与外部 API 的可靠集成超时、幂等与最终一致性一、Agent 调了外部 API 发邮件API 返回超时但邮件实际发出去了。Agent 重试 → 用户收到 5 封一样的邮件Agent 和外部 API 集成时三个经典问题超时不等于失败API 返回 504 超时但后端可能已经执行完了重试不安全没有幂等机制重试就是重复执行状态不一致API 返回成功但 Agent 没收到响应Agent 认为失败了Agent 调用外部 API 不是你调一个/api/predict那么简单。Agent 是在编排一个业务流程涉及多个外部系统一个 API 调用失败可能让整个 Agent 任务状态错乱。二、可靠集成的网络故障模型故障模式分析故障Agent 看到什么API 实际状态正确做法请求丢失连接错误/超时未收到安全重试执行中超时超时不确定可能已执行幂等重试响应丢失超时已执行成功幂等重试拿缓存响应延迟超时可能在阈值边缘正在执行等待或幂等重试API 崩溃连接重置执行中断幂等重试API 返回错误错误码未执行根据错误类型决定重试三、生产级可靠集成实现Agent 侧带幂等的 API 调用客户端# reliable_api_client.py import asyncio import hashlib import json import time import uuid from dataclasses import dataclass, field from enum import Enum from typing import Any, Dict, Optional import httpx class CallStatus(Enum): PENDING pending SUCCESS success FAILED_PERMANENT failed_permanent FAILED_RETRYABLE failed_retryable TIMEOUT timeout dataclass class ApiCallRecord: 每次 API 调用的本地记录 idempotency_key: str url: str method: str status: CallStatus attempt: int 0 response: Optional[Dict] None error: Optional[str] None started_at: float field(default_factorytime.time) completed_at: Optional[float] None class ReliableApiClient: Agent 可靠 API 调用客户端 def __init__( self, max_retries: int 3, base_timeout: float 30.0, max_backoff: float 60.0, ): self.max_retries max_retries self.base_timeout base_timeout self.max_backoff max_backoff self.client httpx.AsyncClient( timeouthttpx.Timeout(base_timeout), limitshttpx.Limits(max_keepalive_connections20), ) # 本地记录生产用 Redis self._records: Dict[str, ApiCallRecord] {} def _generate_idempotency_key( self, method: str, url: str, body: Optional[Dict] None, idempotency_source: Optional[str] None ) - str: 生成幂等 Key if idempotency_source: # 业务侧提供的幂等标识如订单号 raw f{method}:{url}:{idempotency_source} else: # 无业务标识时用请求体 hash body_str json.dumps(body or {}, sort_keysTrue) raw f{method}:{url}:{body_str} return hashlib.sha256(raw.encode()).hexdigest()[:32] async def call( self, method: str, url: str, body: Optional[Dict] None, headers: Optional[Dict] None, idempotency_source: Optional[str] None, timeout: Optional[float] None, ) - Dict[str, Any]: 执行可靠的 API 调用 # 1. 生成幂等 Key idempotency_key self._generate_idempotency_key( method, url, body, idempotency_source ) # 2. 检查是否有已完成的结果幂等 if idempotency_key in self._records: record self._records[idempotency_key] if record.status CallStatus.SUCCESS: return record.response if record.status CallStatus.FAILED_PERMANENT: raise Exception(fPermanent failure: {record.error}) # 3. 记录 record ApiCallRecord( idempotency_keyidempotency_key, urlurl, methodmethod, statusCallStatus.PENDING, ) self._records[idempotency_key] record # 4. 重试循环 last_error None for attempt in range(1, self.max_retries 1): record.attempt attempt try: resp await self._make_request( method, url, body, headers, idempotency_key, timeout or self.base_timeout, ) # 成功 if resp.status_code 500: record.status CallStatus.SUCCESS record.response resp.json() record.completed_at time.time() return record.response # 服务端错误5xx→ 可重试 if resp.status_code 500: last_error fServer error: {resp.status_code} record.status CallStatus.FAILED_RETRYABLE # 客户端错误429 限流→ 可重试 elif resp.status_code 429: last_error Rate limited record.status CallStatus.FAILED_RETRYABLE # 使用 API 返回的 Retry-After retry_after int( resp.headers.get(Retry-After, 5) ) await asyncio.sleep(retry_after) continue # 其它 4xx → 不可重试 else: record.status CallStatus.FAILED_PERMANENT record.error fClient error: {resp.status_code} raise Exception(record.error) except httpx.TimeoutException as e: last_error fTimeout after {timeout}s record.status CallStatus.TIMEOUT except httpx.ConnectError as e: last_error fConnection failed: {e} record.status CallStatus.FAILED_RETRYABLE except httpx.NetworkError as e: last_error fNetwork error: {e} record.status CallStatus.FAILED_RETRYABLE # 退避等待 if attempt self.max_retries: backoff min( self.base_timeout * (2 ** (attempt - 1)), self.max_backoff, ) # 加 Jitter 避免惊群 jitter backoff * 0.1 * (0.5 - __import__(random).random()) wait_time backoff jitter await asyncio.sleep(wait_time) # 所有重试用完 record.status CallStatus.FAILED_RETRYABLE record.error last_error record.completed_at time.time() raise Exception( fFailed after {self.max_retries} attempts: {last_error} ) async def _make_request( self, method: str, url: str, body: Optional[Dict], headers: Optional[Dict], idempotency_key: str, timeout: float, ) - httpx.Response: 发送带幂等 Key 的 HTTP 请求 req_headers { Content-Type: application/json, Idempotency-Key: idempotency_key, X-Request-ID: str(uuid.uuid4()), **(headers or {}), } return await self.client.request( methodmethod, urlurl, jsonbody, headersreq_headers, timeouttimeout, ) async def close(self): await self.client.aclose()API 侧幂等性服务端实现# server_idempotency.py import hashlib import json from datetime import datetime, timedelta from typing import Optional, Dict import redis.asyncio as redis class IdempotencyMiddleware: API 服务端幂等性中间件 def __init__(self, redis_client: redis.Redis, ttl_hours: int 24): self.redis redis_client self.ttl timedelta(hoursttl_hours) async def check_or_create( self, idempotency_key: str ) - Optional[Dict]: 检查幂等 Key返回缓存结果或 None首次执行 使用 Redis SET NX 保证原子性 lock_key fidempotency:{idempotency_key}:lock result_key fidempotency:{idempotency_key}:result # 1. 先检查是否已有结果 cached await self.redis.get(result_key) if cached: return json.loads(cached) # 2. 尝试获取处理锁原子操作 lock_acquired await self.redis.set( lock_key, processing, nxTrue, # Only set if not exists ex60, # 60 秒过期防止死锁 ) if lock_acquired: # 首次到达执行业务 return None # 3. 锁已被占用 → 另一个请求正在处理 # 轮询等待结果 for _ in range(30): # 最多等 30 秒 await asyncio.sleep(1) cached await self.redis.get(result_key) if cached: return json.loads(cached) # 超时 → 可能处理失败客户端应重试 raise Exception(Idempotency lock timeout, please retry) async def store_result( self, idempotency_key: str, response: Dict, status_code: int ): 存储处理结果 result { status_code: status_code, response: response, stored_at: datetime.utcnow().isoformat(), } result_key fidempotency:{idempotency_key}:result lock_key fidempotency:{idempotency_key}:lock await self.redis.setex( result_key, int(self.ttl.total_seconds()), json.dumps(result), ) await self.redis.delete(lock_key) # FastAPI 中间件集成 from fastapi import FastAPI, Request, HTTPException app FastAPI() redis_client redis.Redis.from_url(redis://localhost:6379) idempotency IdempotencyMiddleware(redis_client) app.post(/api/send-email) async def send_email(request: Request): idem_key request.headers.get(Idempotency-Key) if not idem_key: raise HTTPException(400, Idempotency-Key required) # 幂等检查 cached await idempotency.check_or_create(idem_key) if cached: return cached[response] # 执行业务逻辑 body await request.json() result await actually_send_email(body) # 存储结果 await idempotency.store_result( idem_key, result, 200 ) return resultAgent 编排超时 补偿事务# agent_orchestrator.py from enum import Enum from dataclasses import dataclass from typing import List import asyncio class StepStatus(Enum): SUCCESS success FAILED failed TIMEOUT timeout COMPENSATED compensated dataclass class WorkflowStep: name: str execute: callable # 正向操作 compensate: callable # 补偿操作回滚 timeout: float 30.0 retry_count: int 3 class AgentWorkflow: Agent 工作流编排支持超时和补偿事务 def __init__(self, api_client: ReliableApiClient): self.api api_client async def execute_workflow(self, steps: List[WorkflowStep]) - dict: 执行多步骤工作流失败时自动补偿 executed_steps [] results {} try: for step in steps: try: # 执行步骤带超时 result await asyncio.wait_for( step.execute(self.api), timeoutstep.timeout, ) results[step.name] { status: StepStatus.SUCCESS, result: result, } executed_steps.append(step) except asyncio.TimeoutError: results[step.name] { status: StepStatus.TIMEOUT, } raise # 触发补偿 except Exception as e: results[step.name] { status: StepStatus.FAILED, error: str(e), } raise # 触发补偿 return { success: True, steps: results, } except Exception as workflow_error: # 补偿事务逆序执行已成功步骤的补偿 for step in reversed(executed_steps): try: await asyncio.wait_for( step.compensate(self.api), timeout15.0, ) results[step.name][status] StepStatus.COMPENSATED except Exception as comp_error: # 补偿失败 → 需要人工介入 results[step.name][compensation_error] str( comp_error ) return { success: False, error: str(workflow_error), steps: results, } # 使用示例Agent 创建订单流程 async def create_order_workflow(): client ReliableApiClient() workflow AgentWorkflow(client) steps [ WorkflowStep( namereserve_inventory, executelambda c: c.call(POST, https://api.example.com/inventory/reserve, {sku: PROD-123, qty: 1}, idempotency_sourceORD-456-inventory), compensatelambda c: c.call(POST, https://api.example.com/inventory/release, {sku: PROD-123, qty: 1}, idempotency_sourceORD-456-inventory-release), timeout10.0, ), WorkflowStep( namecreate_payment, executelambda c: c.call(POST, https://api.example.com/payment/charge, {amount: 99.00}, idempotency_sourceORD-456-payment), compensatelambda c: c.call(POST, https://api.example.com/payment/refund, {order_id: ORD-456}, idempotency_sourceORD-456-refund), timeout15.0, ), WorkflowStep( namesend_confirmation, executelambda c: c.call(POST, https://api.example.com/email/send, {template: order_confirmation, order_id: ORD-456}, idempotency_sourceORD-456-email), compensatelambda c: c.call(POST, https://api.example.com/email/send, {template: order_cancelled, order_id: ORD-456}, idempotency_sourceORD-456-email-cancel), timeout5.0, ), ] result await workflow.execute_workflow(steps) return result四、边界分析与架构权衡边界场景风险缓解措施Idempotency-Key 碰撞两个不同请求生成相同 Key使用足够长的 HashSHA-256 前 32 位幂等缓存过期缓存清理后重复请求会重复执行TTL 要 业务窗口24h 起步API 不支持 Idempotency-Key服务端无法去重客户端自己维护去重Redis 请求指纹补偿事务也失败数据不一致需要人工介入失败告警 自动重试 运维面板部分成功步骤 1/3 成功补偿 1 失败Saga 模式记录中间状态幂等 Key 丢失Agent 崩溃后重启不知道处理到哪持久化幂等 Key 到数据库权衡重试次数 vs 延迟多次重试成功率更高但总延迟更长。少重试响应快失败率可能高。推荐公式重试次数 log(可接受失败率) / log(单次失败率) # 例单次成功率 95%期望 99.9%需要 log(0.001)/log(0.05) ≈ 2.3 → 3 次权衡最终一致性 vs 强一致性Agent 调用外部 API 的场景最终一致性通常够用发邮件晚 10 秒收到也无所谓更新 CRM1 分钟后再反映也可以创建工单创建了就行不需要毫秒级需要强一致性的场景才用分布式事务支付扣款/库存扣减但大多数 Agent API 调用不需要。五、总结Agent 集成外部 API 的三个核心原则每个 API 调用都带 Idempotency-Key服务端做幂等性去重超时 ≠ 失败不要因为超时就不重试也不要乱重试多步骤用补偿事务Saga 模式正向操作配反向补偿代码自检清单所有非 GET 请求都带 Idempotency-Key重试策略有指数退避 Jitter超时时间基于 API SLA 设置不是盲目 30s有补偿事务处理部分成功场景幂等 Key 支持业务语义如订单号而不是只靠请求体 Hash幂等缓存 TTL 业务可接受的重发窗口一句话Agent 调 API 不是调一次就完了而是调一次 备好重试策略 保证幂等 失败了能回滚。把这四步做成框架别每次手写。
Agent 与外部 API 的可靠集成：超时、幂等与最终一致性
Agent 与外部 API 的可靠集成超时、幂等与最终一致性一、Agent 调了外部 API 发邮件API 返回超时但邮件实际发出去了。Agent 重试 → 用户收到 5 封一样的邮件Agent 和外部 API 集成时三个经典问题超时不等于失败API 返回 504 超时但后端可能已经执行完了重试不安全没有幂等机制重试就是重复执行状态不一致API 返回成功但 Agent 没收到响应Agent 认为失败了Agent 调用外部 API 不是你调一个/api/predict那么简单。Agent 是在编排一个业务流程涉及多个外部系统一个 API 调用失败可能让整个 Agent 任务状态错乱。二、可靠集成的网络故障模型故障模式分析故障Agent 看到什么API 实际状态正确做法请求丢失连接错误/超时未收到安全重试执行中超时超时不确定可能已执行幂等重试响应丢失超时已执行成功幂等重试拿缓存响应延迟超时可能在阈值边缘正在执行等待或幂等重试API 崩溃连接重置执行中断幂等重试API 返回错误错误码未执行根据错误类型决定重试三、生产级可靠集成实现Agent 侧带幂等的 API 调用客户端# reliable_api_client.py import asyncio import hashlib import json import time import uuid from dataclasses import dataclass, field from enum import Enum from typing import Any, Dict, Optional import httpx class CallStatus(Enum): PENDING pending SUCCESS success FAILED_PERMANENT failed_permanent FAILED_RETRYABLE failed_retryable TIMEOUT timeout dataclass class ApiCallRecord: 每次 API 调用的本地记录 idempotency_key: str url: str method: str status: CallStatus attempt: int 0 response: Optional[Dict] None error: Optional[str] None started_at: float field(default_factorytime.time) completed_at: Optional[float] None class ReliableApiClient: Agent 可靠 API 调用客户端 def __init__( self, max_retries: int 3, base_timeout: float 30.0, max_backoff: float 60.0, ): self.max_retries max_retries self.base_timeout base_timeout self.max_backoff max_backoff self.client httpx.AsyncClient( timeouthttpx.Timeout(base_timeout), limitshttpx.Limits(max_keepalive_connections20), ) # 本地记录生产用 Redis self._records: Dict[str, ApiCallRecord] {} def _generate_idempotency_key( self, method: str, url: str, body: Optional[Dict] None, idempotency_source: Optional[str] None ) - str: 生成幂等 Key if idempotency_source: # 业务侧提供的幂等标识如订单号 raw f{method}:{url}:{idempotency_source} else: # 无业务标识时用请求体 hash body_str json.dumps(body or {}, sort_keysTrue) raw f{method}:{url}:{body_str} return hashlib.sha256(raw.encode()).hexdigest()[:32] async def call( self, method: str, url: str, body: Optional[Dict] None, headers: Optional[Dict] None, idempotency_source: Optional[str] None, timeout: Optional[float] None, ) - Dict[str, Any]: 执行可靠的 API 调用 # 1. 生成幂等 Key idempotency_key self._generate_idempotency_key( method, url, body, idempotency_source ) # 2. 检查是否有已完成的结果幂等 if idempotency_key in self._records: record self._records[idempotency_key] if record.status CallStatus.SUCCESS: return record.response if record.status CallStatus.FAILED_PERMANENT: raise Exception(fPermanent failure: {record.error}) # 3. 记录 record ApiCallRecord( idempotency_keyidempotency_key, urlurl, methodmethod, statusCallStatus.PENDING, ) self._records[idempotency_key] record # 4. 重试循环 last_error None for attempt in range(1, self.max_retries 1): record.attempt attempt try: resp await self._make_request( method, url, body, headers, idempotency_key, timeout or self.base_timeout, ) # 成功 if resp.status_code 500: record.status CallStatus.SUCCESS record.response resp.json() record.completed_at time.time() return record.response # 服务端错误5xx→ 可重试 if resp.status_code 500: last_error fServer error: {resp.status_code} record.status CallStatus.FAILED_RETRYABLE # 客户端错误429 限流→ 可重试 elif resp.status_code 429: last_error Rate limited record.status CallStatus.FAILED_RETRYABLE # 使用 API 返回的 Retry-After retry_after int( resp.headers.get(Retry-After, 5) ) await asyncio.sleep(retry_after) continue # 其它 4xx → 不可重试 else: record.status CallStatus.FAILED_PERMANENT record.error fClient error: {resp.status_code} raise Exception(record.error) except httpx.TimeoutException as e: last_error fTimeout after {timeout}s record.status CallStatus.TIMEOUT except httpx.ConnectError as e: last_error fConnection failed: {e} record.status CallStatus.FAILED_RETRYABLE except httpx.NetworkError as e: last_error fNetwork error: {e} record.status CallStatus.FAILED_RETRYABLE # 退避等待 if attempt self.max_retries: backoff min( self.base_timeout * (2 ** (attempt - 1)), self.max_backoff, ) # 加 Jitter 避免惊群 jitter backoff * 0.1 * (0.5 - __import__(random).random()) wait_time backoff jitter await asyncio.sleep(wait_time) # 所有重试用完 record.status CallStatus.FAILED_RETRYABLE record.error last_error record.completed_at time.time() raise Exception( fFailed after {self.max_retries} attempts: {last_error} ) async def _make_request( self, method: str, url: str, body: Optional[Dict], headers: Optional[Dict], idempotency_key: str, timeout: float, ) - httpx.Response: 发送带幂等 Key 的 HTTP 请求 req_headers { Content-Type: application/json, Idempotency-Key: idempotency_key, X-Request-ID: str(uuid.uuid4()), **(headers or {}), } return await self.client.request( methodmethod, urlurl, jsonbody, headersreq_headers, timeouttimeout, ) async def close(self): await self.client.aclose()API 侧幂等性服务端实现# server_idempotency.py import hashlib import json from datetime import datetime, timedelta from typing import Optional, Dict import redis.asyncio as redis class IdempotencyMiddleware: API 服务端幂等性中间件 def __init__(self, redis_client: redis.Redis, ttl_hours: int 24): self.redis redis_client self.ttl timedelta(hoursttl_hours) async def check_or_create( self, idempotency_key: str ) - Optional[Dict]: 检查幂等 Key返回缓存结果或 None首次执行 使用 Redis SET NX 保证原子性 lock_key fidempotency:{idempotency_key}:lock result_key fidempotency:{idempotency_key}:result # 1. 先检查是否已有结果 cached await self.redis.get(result_key) if cached: return json.loads(cached) # 2. 尝试获取处理锁原子操作 lock_acquired await self.redis.set( lock_key, processing, nxTrue, # Only set if not exists ex60, # 60 秒过期防止死锁 ) if lock_acquired: # 首次到达执行业务 return None # 3. 锁已被占用 → 另一个请求正在处理 # 轮询等待结果 for _ in range(30): # 最多等 30 秒 await asyncio.sleep(1) cached await self.redis.get(result_key) if cached: return json.loads(cached) # 超时 → 可能处理失败客户端应重试 raise Exception(Idempotency lock timeout, please retry) async def store_result( self, idempotency_key: str, response: Dict, status_code: int ): 存储处理结果 result { status_code: status_code, response: response, stored_at: datetime.utcnow().isoformat(), } result_key fidempotency:{idempotency_key}:result lock_key fidempotency:{idempotency_key}:lock await self.redis.setex( result_key, int(self.ttl.total_seconds()), json.dumps(result), ) await self.redis.delete(lock_key) # FastAPI 中间件集成 from fastapi import FastAPI, Request, HTTPException app FastAPI() redis_client redis.Redis.from_url(redis://localhost:6379) idempotency IdempotencyMiddleware(redis_client) app.post(/api/send-email) async def send_email(request: Request): idem_key request.headers.get(Idempotency-Key) if not idem_key: raise HTTPException(400, Idempotency-Key required) # 幂等检查 cached await idempotency.check_or_create(idem_key) if cached: return cached[response] # 执行业务逻辑 body await request.json() result await actually_send_email(body) # 存储结果 await idempotency.store_result( idem_key, result, 200 ) return resultAgent 编排超时 补偿事务# agent_orchestrator.py from enum import Enum from dataclasses import dataclass from typing import List import asyncio class StepStatus(Enum): SUCCESS success FAILED failed TIMEOUT timeout COMPENSATED compensated dataclass class WorkflowStep: name: str execute: callable # 正向操作 compensate: callable # 补偿操作回滚 timeout: float 30.0 retry_count: int 3 class AgentWorkflow: Agent 工作流编排支持超时和补偿事务 def __init__(self, api_client: ReliableApiClient): self.api api_client async def execute_workflow(self, steps: List[WorkflowStep]) - dict: 执行多步骤工作流失败时自动补偿 executed_steps [] results {} try: for step in steps: try: # 执行步骤带超时 result await asyncio.wait_for( step.execute(self.api), timeoutstep.timeout, ) results[step.name] { status: StepStatus.SUCCESS, result: result, } executed_steps.append(step) except asyncio.TimeoutError: results[step.name] { status: StepStatus.TIMEOUT, } raise # 触发补偿 except Exception as e: results[step.name] { status: StepStatus.FAILED, error: str(e), } raise # 触发补偿 return { success: True, steps: results, } except Exception as workflow_error: # 补偿事务逆序执行已成功步骤的补偿 for step in reversed(executed_steps): try: await asyncio.wait_for( step.compensate(self.api), timeout15.0, ) results[step.name][status] StepStatus.COMPENSATED except Exception as comp_error: # 补偿失败 → 需要人工介入 results[step.name][compensation_error] str( comp_error ) return { success: False, error: str(workflow_error), steps: results, } # 使用示例Agent 创建订单流程 async def create_order_workflow(): client ReliableApiClient() workflow AgentWorkflow(client) steps [ WorkflowStep( namereserve_inventory, executelambda c: c.call(POST, https://api.example.com/inventory/reserve, {sku: PROD-123, qty: 1}, idempotency_sourceORD-456-inventory), compensatelambda c: c.call(POST, https://api.example.com/inventory/release, {sku: PROD-123, qty: 1}, idempotency_sourceORD-456-inventory-release), timeout10.0, ), WorkflowStep( namecreate_payment, executelambda c: c.call(POST, https://api.example.com/payment/charge, {amount: 99.00}, idempotency_sourceORD-456-payment), compensatelambda c: c.call(POST, https://api.example.com/payment/refund, {order_id: ORD-456}, idempotency_sourceORD-456-refund), timeout15.0, ), WorkflowStep( namesend_confirmation, executelambda c: c.call(POST, https://api.example.com/email/send, {template: order_confirmation, order_id: ORD-456}, idempotency_sourceORD-456-email), compensatelambda c: c.call(POST, https://api.example.com/email/send, {template: order_cancelled, order_id: ORD-456}, idempotency_sourceORD-456-email-cancel), timeout5.0, ), ] result await workflow.execute_workflow(steps) return result四、边界分析与架构权衡边界场景风险缓解措施Idempotency-Key 碰撞两个不同请求生成相同 Key使用足够长的 HashSHA-256 前 32 位幂等缓存过期缓存清理后重复请求会重复执行TTL 要 业务窗口24h 起步API 不支持 Idempotency-Key服务端无法去重客户端自己维护去重Redis 请求指纹补偿事务也失败数据不一致需要人工介入失败告警 自动重试 运维面板部分成功步骤 1/3 成功补偿 1 失败Saga 模式记录中间状态幂等 Key 丢失Agent 崩溃后重启不知道处理到哪持久化幂等 Key 到数据库权衡重试次数 vs 延迟多次重试成功率更高但总延迟更长。少重试响应快失败率可能高。推荐公式重试次数 log(可接受失败率) / log(单次失败率) # 例单次成功率 95%期望 99.9%需要 log(0.001)/log(0.05) ≈ 2.3 → 3 次权衡最终一致性 vs 强一致性Agent 调用外部 API 的场景最终一致性通常够用发邮件晚 10 秒收到也无所谓更新 CRM1 分钟后再反映也可以创建工单创建了就行不需要毫秒级需要强一致性的场景才用分布式事务支付扣款/库存扣减但大多数 Agent API 调用不需要。五、总结Agent 集成外部 API 的三个核心原则每个 API 调用都带 Idempotency-Key服务端做幂等性去重超时 ≠ 失败不要因为超时就不重试也不要乱重试多步骤用补偿事务Saga 模式正向操作配反向补偿代码自检清单所有非 GET 请求都带 Idempotency-Key重试策略有指数退避 Jitter超时时间基于 API SLA 设置不是盲目 30s有补偿事务处理部分成功场景幂等 Key 支持业务语义如订单号而不是只靠请求体 Hash幂等缓存 TTL 业务可接受的重发窗口一句话Agent 调 API 不是调一次就完了而是调一次 备好重试策略 保证幂等 失败了能回滚。把这四步做成框架别每次手写。
