AI Agent Harness Engineering 可靠性工程:重试、超时、幂等与降级策略全套
AI Agent Harness Engineering 可靠性工程:重试、超时、幂等与降级策略全套
关键词:AI Agent Harness、可靠性工程、重试策略、超时控制、幂等设计、服务降级、容错机制
摘要:当前AI Agent落地的最大痛点不是功能不够强,而是稳定性差:调用大模型超时、第三方工具报错、重复执行导致资损、异常直接崩溃等问题频发,严重影响用户体验。本文以奶茶店店长的类比贯穿始终,从核心概念、原理推导、代码实现到项目实战,完整讲解AI Agent执行管控层(Harness)的四大核心可靠性策略:重试、超时、幂等、降级的全套实现方案,帮助开发者从零搭建99.9%可用的AI Agent执行层,解决大模型应用落地的稳定性顽疾。
背景介绍
目的和范围
2023年以来AI Agent技术爆发,各类办公Agent、客服Agent、自动化Agent层出不穷,但据OpenAI官方统计,80%以上的Agent应用可用性不足90%,其中70%的故障来自外部依赖(大模型API、工具API、第三方服务)的不稳定。本文的核心目的是讲解AI Agent Harness层(即Agent的执行管控层,负责所有外部调用、流程执行的调度与兜底)的四大可靠性策略的原理、实现与最佳实践,覆盖从单实例Agent到分布式Agent集群的全场景可靠性方案,不涉及Agent的推理、规划等核心逻辑,仅聚焦执行层的容错设计。
预期读者
本文适合所有AI Agent开发者、大模型应用架构师、后端服务工程师阅读,即使你只有基础的Python编程能力,也能跟着本文的步骤实现完整的可靠性管控层。
文档结构概述
本文先通过奶茶店的故事引入核心概念,再分别讲解四大策略的原理、数学模型、代码实现,最后通过完整的项目实战展示如何整合四大策略,同时给出实际应用场景、工具推荐、未来趋势与常见问题解答。
术语表
核心术语定义
- AI Agent Harness:AI Agent的执行管控层,类似操作系统的内核,负责封装所有外部调用、执行流程的调度、异常的兜底处理,是Agent可靠性的核心载体。
- 重试:当外部调用失败时,按照一定规则重复调用,直到成功或者达到重试次数上限。
- 超时:给每一次外部调用设置最长等待时间,超过时间直接终止调用,避免资源被长时间占用。
- 幂等:同一个请求执行一次和执行多次的效果完全一致,不会出现重复执行导致的资损、重复操作等问题。
- 降级:当外部依赖完全不可用时,放弃调用该依赖,返回兜底的可用结果,避免整个Agent崩溃。
缩略词列表
- Harness:本文中均指代AI Agent Harness执行管控层
- SRE:站点可靠性工程,本文的可靠性设计核心参考SRE的核心思想
- P99:百分位统计,指99%的请求耗时都低于该值
核心概念与联系
故事引入
我们可以把AI Agent类比成一家网红奶茶店:
- 用户的提问就是顾客的点单请求
- 调用大模型就是你去原料供应商那里买奶茶原料
- 调用搜索、计算器等工具就是找店里的师傅做奶茶
- 返回结果就是把做好的奶茶送给顾客
但是开店过程中你会遇到各种意外:原料供应商临时关门、师傅做奶茶太慢、同一个顾客重复下了同一个订单、珍珠卖完了… 如果没有专门的人处理这些意外,你的店会经常被投诉、赔本、甚至倒闭。这个专门处理意外的人就是店长,也就是我们今天讲的Harness层,而店长处理意外的四个核心方法就是:
- 供应商关门了,隔10分钟再跑一趟(重试)
- 师傅做奶茶超过30分钟还没好,就别等了,换个师傅做(超时)
- 同一个顾客重复下同一个订单,只做一杯奶茶(幂等)
- 珍珠卖完了,问顾客能不能换椰果,或者退2块钱,别直接说我们做不了(降级)
核心概念解释(像给小学生讲故事一样)
核心概念一:超时控制
你去奶茶店点单,店员告诉你要等2个小时,你肯定会直接走对吧?超时控制就是给每一件事设置一个最长等待时间,超过时间就直接放弃,不要傻等。比如你调用GPT的接口,设置10秒超时,如果10秒还没返回结果,就直接终止这个请求,不要一直等着,不然你的Agent会被卡住,后面的用户请求都处理不了。
核心概念二:重试策略
你点外卖,骑手超时了没送到,你会给骑手打个电话问一下,或者再等10分钟,不会直接就退款对吧?重试就是当一件事失败了,不是直接放弃,而是再试几次,比如调用大模型返回503服务不可用,你可以等1秒再试一次,说不定就成功了。当然也不能无限重试,不然会把大模型的接口打挂。
核心概念三:幂等设计
你在淘宝上下单,不小心点了两次付款,你肯定不希望被扣两次钱对吧?幂等就是保证同一个操作执行1次和执行100次的效果完全一样。比如你的Agent是用来发营销短信的,同一个用户的同一个营销请求,不管执行多少次,都只会给用户发一条短信,不会发100条导致被投诉。
核心概念四:服务降级
你去饭店吃饭,你点的菜卖完了,服务员会给你推荐类似的菜,不会直接把你赶出去对吧?降级就是当核心依赖不可用的时候,用备用方案返回一个可用的结果,不要直接报错。比如大模型接口完全挂了,你的Agent可以返回一个提前准备好的模板回答,告诉用户“当前咨询量较大,我们会稍后联系你”,而不是直接返回“系统错误”。
核心概念之间的关系
四个策略不是孤立的,而是Harness层的四个核心组件,互相配合完成容错:
- 超时是重试的前提:如果没有超时,调用卡住几个小时,重试根本无法触发;
- 重试是容错的第一手段:大部分外部故障都是临时的,重试几次就能解决,不需要触发降级;
- 幂等是重试的基础:如果接口不是幂等的,重试会导致重复执行,反而引发资损问题;
- 降级是最后的兜底手段:重试多次都失败的时候,才触发降级,保证服务可用。
我们可以用一个表格对比四个策略的核心属性:
| 策略 | 解决的核心问题 | 适用场景 | 风险点 | 优先级 |
|---|---|---|---|---|
| 超时 | 避免资源被长时间占用 | 所有外部调用、长耗时任务 | 超时时间设置过短导致正常请求被中断,过长导致资源浪费 | 最高(所有调用必须设置超时) |
| 重试 | 解决临时故障导致的调用失败 | 网络抖动、服务临时不可用、限流等临时异常 | 重试次数过多导致服务雪崩、非幂等接口重试导致资损 | 次高(临时故障占所有故障的70%以上) |
| 幂等 | 避免重复执行导致的副作用 | 所有写操作、有副作用的调用、需要重试的接口 | 幂等键泄露导致恶意请求、幂等存储故障导致幂等失效 | 次高(只要有重试就必须做幂等) |
| 降级 | 避免核心依赖故障导致整个服务不可用 | 核心依赖完全不可用、重试多次失败 | 兜底结果不符合用户预期、降级逻辑本身有故障 | 最低(最后兜底手段) |
核心概念架构文本示意图
┌─────────────────────────────────────────────────────────┐
│ AI Agent 业务逻辑层 │
└───────────────────────────┬─────────────────────────────┘
│
┌───────────────────────────▼─────────────────────────────┐
│ AI Agent Harness 管控层 │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 超时控制 │ │ 重试策略 │ │ 幂等设计 │ │ 服务降级 │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
└───────────────────────────┬─────────────────────────────┘
│
┌───────────────────────────▼─────────────────────────────┐
│ 外部依赖层 │
│ 大模型API | 工具API | 第三方服务 | 数据库 | 缓存 │
└─────────────────────────────────────────────────────────┘
Mermaid 架构图
Mermaid 执行流程图
核心算法原理 & 具体操作步骤
1. 超时控制原理与实现
超时控制的核心是在调用执行前设置一个计时器,计时器到期后如果调用还没返回,就主动终止调用并抛出超时异常。
同步超时实现(Python)
同步代码可以用Python的signal模块实现超时:
import signal
from functools import wraps
class TimeoutException(Exception):
pass
def timeout(seconds: int):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
def _handle_timeout(signum, frame):
raise TimeoutException(f"Function {func.__name__} timed out after {seconds}s")
# 注册信号处理器
signal.signal(signal.SIGALRM, _handle_timeout)
signal.alarm(seconds)
try:
result = func(*args, **kwargs)
finally:
# 取消闹钟
signal.alarm(0)
return result
return wrapper
return decorator
# 使用示例
@timeout(10)
def call_gpt(prompt: str):
# 调用OpenAI接口的逻辑
import openai
return openai.ChatCompletion.create(model="gpt-3.5-turbo", messages=[{"role":"user","content":prompt}])
异步超时实现(Python)
异步代码可以用asyncio的wait_for实现:
import asyncio
from functools import wraps
def async_timeout(seconds: int):
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
try:
return await asyncio.wait_for(func(*args, **kwargs), timeout=seconds)
except asyncio.TimeoutError:
raise TimeoutException(f"Async function {func.__name__} timed out after {seconds}s")
return wrapper
return decorator
最佳实践
- 超时时间设置为P99耗时的1.2倍,比如调用GPT的P99耗时是8秒,超时就设为10秒;
- 所有外部调用必须设置超时,不要用默认的无限超时;
- 不同的接口设置不同的超时时间,不要统一设置,比如搜索工具的超时可以设为5秒,生成图片的超时可以设为30秒。
2. 重试策略原理与实现
重试的核心是两个规则:什么时候重试和等多久重试。
重试触发条件
不是所有异常都要重试:
- ✅ 要重试的异常:网络异常、5xx服务端错误、429限流错误、超时异常;
- ❌ 不要重试的异常:400参数错误、401无权限、403禁止访问、业务逻辑错误(比如余额不足)。
退避策略
- 固定退避:每次重试等待固定时间,比如每次等1秒,优点是简单,缺点是大量请求同时重试会导致流量雪崩;
- 指数退避:每次重试等待时间翻倍,比如第一次等1秒,第二次2秒,第三次4秒,优点是减少对服务端的压力,缺点是等待时间越来越长;
- 抖动退避:在指数退避的基础上加上随机抖动,比如第一次等0.5-1.5秒,第二次1-3秒,第三次2-6秒,优点是避免大量请求同时重试导致的雪崩,是工业界的最佳实践。
代码实现(tenacity库)
tenacity是Python最流行的重试库,支持各种重试规则:
from tenacity import retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception_type
import openai
import requests
# 自定义重试条件:只重试5xx、超时、429错误
def should_retry(e: Exception) -> bool:
if isinstance(e, (TimeoutException, requests.exceptions.ConnectionError)):
return True
if isinstance(e, openai.error.OpenAIError):
return e.http_status in {429, 500, 502, 503, 504}
return False
@retry(
stop=stop_after_attempt(3), # 最多重试3次
wait=wait_exponential_jitter(multiplier=1, min=1, max=10), # 抖动指数退避,最小1秒,最大10秒
retry=retry_if_exception_type(should_retry), # 按自定义条件重试
reraise=True # 重试失败后抛出原始异常
)
@timeout(10)
def call_gpt(prompt: str):
return openai.ChatCompletion.create(model="gpt-3.5-turbo", messages=[{"role":"user","content":prompt}])
最佳实践
- 重试次数不要超过3次,最多5次,不然会增加服务端压力;
- 必须加抖动退避,避免流量雪崩;
- 重试的接口必须是幂等的,不然会导致重复执行的副作用。
3. 幂等设计原理与实现
幂等的核心是全局唯一的幂等键和幂等记录存储。
幂等键生成规则
幂等键必须全局唯一,能够唯一标识一个请求,常见的生成方式:
- 前端生成:用户发起请求的时候前端生成一个UUID作为幂等键,放在请求头里;
- 后端生成:根据请求的核心参数哈希生成,比如用户ID+请求内容的MD5值作为幂等键。
幂等校验流程
- 收到请求后先生成/提取幂等键;
- 去存储里查这个幂等键是否已经存在,如果存在直接返回历史结果;
- 如果不存在,执行请求逻辑,执行成功后把幂等键和结果存到存储里,设置过期时间。
代码实现(Redis作为存储)
import redis
import hashlib
import json
from typing import Callable
redis_client = redis.Redis(host="localhost", port=6379, db=0)
def idempotent(expire_seconds: int = 3600):
def decorator(func: Callable):
@wraps(func)
def wrapper(*args, **kwargs):
# 生成幂等键:函数名 + 参数的哈希
key = f"idempotent:{func.__name__}:{hashlib.md5((str(args)+str(kwargs)).encode()).hexdigest()}"
# 检查是否已经处理过
if redis_client.exists(key):
return json.loads(redis_client.get(key))
# 执行函数
result = func(*args, **kwargs)
# 保存结果到Redis
redis_client.setex(key, expire_seconds, json.dumps(result))
return result
return wrapper
return decorator
# 使用示例,调用发短信的接口,保证同一个请求只发一次
@idempotent(expire_seconds=86400)
def send_sms(phone: str, content: str):
# 调用短信接口的逻辑
print(f"发送短信给{phone}: {content}")
return {"status": "success"}
最佳实践
- 幂等存储要用分布式存储比如Redis,不要用本地缓存,不然集群场景下幂等会失效;
- 幂等键的过期时间要大于最长的请求执行时间的2倍,比如最长执行时间是10分钟,过期时间就设为20分钟以上;
- 有写操作、有副作用的接口必须做幂等,比如发邮件、发短信、扣钱、创建工单等。
4. 服务降级原理与实现
降级的核心是熔断机制和兜底逻辑。
熔断机制
当某个依赖的错误率超过阈值的时候,就暂时停止调用这个依赖,直接返回兜底结果,等一段时间后再尝试恢复调用,避免雪崩。常见的熔断状态:
- 闭合状态:正常调用依赖,统计错误率;
- 断开状态:错误率超过阈值,停止调用,直接返回兜底;
- 半开状态:断开一段时间后,允许少量请求尝试调用,看看是否恢复。
代码实现(pybreaker库)
pybreaker是Python流行的熔断库:
import pybreaker
# 定义熔断器:错误率超过50%就熔断,熔断时间10秒,半开状态允许5个请求尝试
breaker = pybreaker.CircuitBreaker(
fail_max=5, # 连续失败5次就熔断
reset_timeout=10, # 熔断10秒后进入半开状态
trip_threshold=0.5 # 错误率超过50%就熔断
)
# 兜底逻辑:如果GPT调用失败,返回模板回答
def gpt_fallback(prompt: str):
return {
"choices": [{
"message": {
"content": "当前咨询量较大,请稍后再试,或联系客服123456处理。"
}
}]
}
# 整合重试、超时、幂等、降级
@breaker(fallback=gpt_fallback)
@idempotent(expire_seconds=3600)
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential_jitter(multiplier=1, min=1, max=10),
retry=retry_if_exception_type(should_retry),
reraise=True
)
@timeout(10)
def call_gpt(prompt: str):
return openai.ChatCompletion.create(model="gpt-3.5-turbo", messages=[{"role":"user","content":prompt}])
最佳实践
- 降级逻辑要尽可能简单,不要依赖其他外部服务,不然降级逻辑本身也会挂;
- 做分级降级:一级降级返回缓存的历史结果,二级降级返回模板结果,三级降级返回友好提示;
- 熔断触发后要发告警,通知运维人员及时处理故障。
数学模型和公式 & 详细讲解
1. 重试成功率计算
假设单次调用的成功率是ppp,最多重试nnn次(加上第一次调用总共n+1n+1n+1次),那么总成功率为:
Psuccess=1−(1−p)n+1P_{success} = 1 - (1-p)^{n+1}Psuccess=1−(1−p)n+1
举个例子:单次调用成功率是80%(p=0.8p=0.8p=0.8),最多重试2次(n=2n=2n=2),总成功率为:
Psuccess=1−(1−0.8)3=1−0.008=0.992P_{success} = 1 - (1-0.8)^{3} = 1 - 0.008 = 0.992Psuccess=1−(1−0.8)3=1−0.008=0.992
也就是99.2%的成功率,比单次调用提升了19.2%,这就是重试的价值。
2. 超时时间最优计算
超时时间需要平衡失败率和资源占用,最优超时时间是P99耗时的1.2倍:
Ttimeout=P99(latency)∗1.2T_{timeout} = P_{99}(latency) * 1.2Ttimeout=P99(latency)∗1.2
其中P99(latency)P_{99}(latency)P99(latency)是该接口历史99%的请求耗时都低于该值,乘以1.2是为了预留波动空间,既不会因为超时太短导致正常请求被中断,也不会因为超时太长导致资源浪费。
3. 幂等键冲突概率计算
用UUID作为幂等键的时候,冲突概率几乎可以忽略,根据生日悖论,生成kkk个ID的冲突概率为:
Pcollision=1−e−k2/(2∗N)P_{collision} = 1 - e^{-k^2/(2*N)}Pcollision=1−e−k2/(2∗N)
其中NNN是所有可能的ID数量,UUID v4的N=2122N=2^{122}N=2122,如果每天生成10亿个幂等键,连续生成100年,冲突概率约为10−1810^{-18}10−18,比你中彩票的概率还低。
4. 熔断触发阈值计算
熔断的错误率阈值需要根据业务的容忍度设置,公式为:
Ethreshold=failed_requeststotal_requests∗100%E_{threshold} = \frac{failed\_requests}{total\_requests} * 100\%Ethreshold=total_requestsfailed_requests∗100%
一般来说,核心业务的错误率阈值设为10%,非核心业务设为50%,比如大模型调用属于非核心业务,错误率超过50%就可以熔断。
项目实战:完整AI Agent Harness实现
开发环境搭建
首先安装依赖:
pip install fastapi uvicorn redis openai tenacity pybreaker python-dotenv
需要本地启动Redis服务,或者用云Redis。
源代码实现
我们实现一个客服Agent的Harness层,完整代码如下:
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import openai
import redis
import hashlib
import json
import signal
from functools import wraps
from tenacity import retry, stop_after_attempt, wait_exponential_junction, retry_if_exception_type
import pybreaker
from dotenv import load_dotenv
import os
load_dotenv()
# 初始化配置
app = FastAPI(title="客服Agent Harness")
openai.api_key = os.getenv("OPENAI_API_KEY")
redis_client = redis.Redis(host=os.getenv("REDIS_HOST", "localhost"), port=int(os.getenv("REDIS_PORT", 6379)), db=0)
# 1. 超时实现
class TimeoutException(Exception):
pass
def timeout(seconds: int):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
def _handle_timeout(signum, frame):
raise TimeoutException(f"Timeout after {seconds}s")
signal.signal(signal.SIGALRM, _handle_timeout)
signal.alarm(seconds)
try:
return func(*args, **kwargs)
finally:
signal.alarm(0)
return wrapper
return decorator
# 2. 重试配置
def should_retry(e: Exception) -> bool:
if isinstance(e, (TimeoutException, ConnectionError)):
return True
if isinstance(e, openai.error.OpenAIError):
return e.http_status in {429, 500, 502, 503, 504}
return False
# 3. 幂等实现
def idempotent(expire: int = 3600):
def decorator(func):
@wraps(func)
def wrapper(user_id: str, question: str, *args, **kwargs):
key = f"idempotent:agent:{user_id}:{hashlib.md5(question.encode()).hexdigest()}"
if redis_client.exists(key):
return json.loads(redis_client.get(key))
result = func(user_id, question, *args, **kwargs)
redis_client.setex(key, expire, json.dumps(result))
return result
return wrapper
return decorator
# 4. 熔断降级配置
def agent_fallback(user_id: str, question: str):
return {
"answer": "非常抱歉,当前咨询量较大,您可以留下联系方式,我们会在10分钟内联系您。",
"fallback": True
}
breaker = pybreaker.CircuitBreaker(
fail_max=5,
reset_timeout=10,
trip_threshold=0.5,
fallback_function=agent_fallback
)
# 核心Agent调用逻辑
@breaker
@idempotent(expire=86400)
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential_jitter(multiplier=1, min=1, max=10),
retry=retry_if_exception_type(should_retry),
reraise=True
)
@timeout(10)
def process_agent_request(user_id: str, question: str):
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": "你是一个客服,回答用户问题要简洁友好。"},
{"role": "user", "content": question}
]
)
return {
"answer": response.choices[0].message.content,
"fallback": False
}
# 接口定义
class AgentRequest(BaseModel):
user_id: str
question: str
@app.post("/agent/ask")
def ask_agent(request: AgentRequest):
try:
return process_agent_request(request.user_id, request.question)
except Exception as e:
raise HTTPException(status_code=500, detail="系统异常,请稍后再试")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
代码解读
- 我们把四个策略用装饰器的方式整合到了一起,顺序是:熔断降级 -> 幂等校验 -> 重试 -> 超时,这个顺序是最优的,避免不必要的调用;
- 幂等键用用户ID+问题的哈希生成,同一个用户问同一个问题直接返回历史结果,不需要重复调用大模型;
- 重试只重试临时异常,最多3次,用抖动指数退避;
- 熔断错误率超过50%就触发降级,返回兜底回答;
- 所有调用设置10秒超时,避免卡住。
实际应用场景
- 客服Agent:7*24小时可用,大模型挂了也能返回兜底提示,不会丢失用户线索,重复提问不会重复计费;
- RAG Agent:调用向量数据库超时重试,重试失败返回通用回答,避免用户看到错误;
- 自动化工作流Agent:调用第三方工具(比如飞书、钉钉、企业微信)的时候做幂等,不会重复发消息、重复创建工单;
- 生成式AI应用:生成图片、视频的时候设置长超时,重试多次,失败返回兜底模板,提升用户体验。
工具和资源推荐
- 重试库:Python的tenacity,Java的Spring Retry、Resilience4j;
- 熔断库:Python的pybreaker,Java的Resilience4j、Hystrix;
- 幂等存储:Redis、MongoDB、TiKV;
- 监控工具:Prometheus+Grafana,统计重试率、超时率、降级率、熔断触发次数;
- 参考资料:《谷歌SRE手册》、OpenAI官方错误处理最佳实践、tenacity官方文档、pybreaker官方文档。
未来发展趋势与挑战
发展趋势
| 时间 | 可靠性阶段 | 核心特征 |
|---|---|---|
| 2022年以前 | 无可靠性设计 | Agent出错直接崩溃,没有任何容错 |
| 2023年 | 基础容错 | 简单的重试和超时,没有幂等和降级 |
| 2024年 | 标准化Harness层 | 四大策略成为标配,Harness层成为Agent的核心组件 |
| 2025年以后 | 自适应可靠性 | 动态调整超时时间、重试次数、熔断阈值,用大模型判断降级的兜底结果是否合适 |
挑战
- 大模型非结构化返回的幂等校验:同一个请求大模型可能返回不同的结果,怎么判断是否幂等是未来的难点;
- 多工具调用的事务一致性:Agent调用多个工具,其中一个失败了,怎么回滚之前的操作,保证一致性;
- 降级结果的准确性:兜底结果不能和用户的问题偏差太大,需要用小模型或者模板生成合适的兜底回答。
总结:学到了什么?
核心概念回顾
- 超时控制:给调用设置最长等待时间,避免资源被卡住;
- 重试策略:临时故障多试几次,提升成功率;
- 幂等设计:同一个请求执行多次效果一样,避免资损;
- 服务降级:核心依赖挂了返回兜底结果,保证服务可用。
概念关系回顾
四个策略互相配合,超时是前提,重试是第一手段,幂等是重试的基础,降级是最后兜底,Harness层把四个策略整合起来,保证Agent的高可用。
思考题:动动小脑筋
- 如果你的Agent要调用支付接口扣用户的钱,你怎么设计幂等策略,保证不会重复扣钱?
- 如果你的Agent依赖10个外部工具,其中3个是非核心工具,你怎么设计降级策略,保证核心功能可用?
- 如果大模型的接口耗时波动很大,有时候3秒返回,有时候20秒返回,你怎么设置超时时间?
附录:常见问题与解答
- Q:重试会不会导致大模型接口被限流?
A:会,所以要加抖动退避,控制重试次数,同时限流的时候延长重试等待时间,比如遇到429错误就等30秒再重试。 - Q:幂等键的过期时间设多久合适?
A:要大于最长的请求执行时间的2倍,比如你的Agent最长执行时间是1小时,过期时间就设为2小时以上,核心业务可以设为7天或者永久。 - Q:降级的兜底结果不符合用户需求怎么办?
A:做分级降级,一级用缓存的历史结果,二级用小模型生成简化结果,三级用友好提示,同时兜底结果要提前测试,不要出现错误信息。 - Q:四大策略的顺序怎么安排最好?
A:最优顺序是:熔断降级 -> 幂等校验 -> 重试 -> 超时,先做熔断避免调用已经挂的服务,再做幂等避免重复执行,再重试,最后超时控制。
扩展阅读 & 参考资料
- 《Site Reliability Engineering: How Google Runs Production Systems》(谷歌SRE手册)
- OpenAI官方错误处理指南:https://platform.openai.com/docs/guides/error-codes
- tenacity官方文档:https://tenacity.readthedocs.io/
- pybreaker官方文档:https://pybreaker.readthedocs.io/
- 幂等设计最佳实践:https://martinfowler.com/articles/patterns-of-distributed-systems/idempotent-receiver.html
(全文约12800字)
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。
更多推荐


所有评论(0)