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编程能力,也能跟着本文的步骤实现完整的可靠性管控层。

文档结构概述

本文先通过奶茶店的故事引入核心概念,再分别讲解四大策略的原理、数学模型、代码实现,最后通过完整的项目实战展示如何整合四大策略,同时给出实际应用场景、工具推荐、未来趋势与常见问题解答。

术语表

核心术语定义
  1. AI Agent Harness:AI Agent的执行管控层,类似操作系统的内核,负责封装所有外部调用、执行流程的调度、异常的兜底处理,是Agent可靠性的核心载体。
  2. 重试:当外部调用失败时,按照一定规则重复调用,直到成功或者达到重试次数上限。
  3. 超时:给每一次外部调用设置最长等待时间,超过时间直接终止调用,避免资源被长时间占用。
  4. 幂等:同一个请求执行一次和执行多次的效果完全一致,不会出现重复执行导致的资损、重复操作等问题。
  5. 降级:当外部依赖完全不可用时,放弃调用该依赖,返回兜底的可用结果,避免整个Agent崩溃。
缩略词列表
  • Harness:本文中均指代AI Agent Harness执行管控层
  • SRE:站点可靠性工程,本文的可靠性设计核心参考SRE的核心思想
  • P99:百分位统计,指99%的请求耗时都低于该值

核心概念与联系

故事引入

我们可以把AI Agent类比成一家网红奶茶店:

  • 用户的提问就是顾客的点单请求
  • 调用大模型就是你去原料供应商那里买奶茶原料
  • 调用搜索、计算器等工具就是找店里的师傅做奶茶
  • 返回结果就是把做好的奶茶送给顾客

但是开店过程中你会遇到各种意外:原料供应商临时关门、师傅做奶茶太慢、同一个顾客重复下了同一个订单、珍珠卖完了… 如果没有专门的人处理这些意外,你的店会经常被投诉、赔本、甚至倒闭。这个专门处理意外的人就是店长,也就是我们今天讲的Harness层,而店长处理意外的四个核心方法就是:

  1. 供应商关门了,隔10分钟再跑一趟(重试)
  2. 师傅做奶茶超过30分钟还没好,就别等了,换个师傅做(超时)
  3. 同一个顾客重复下同一个订单,只做一杯奶茶(幂等)
  4. 珍珠卖完了,问顾客能不能换椰果,或者退2块钱,别直接说我们做不了(降级)

核心概念解释(像给小学生讲故事一样)

核心概念一:超时控制

你去奶茶店点单,店员告诉你要等2个小时,你肯定会直接走对吧?超时控制就是给每一件事设置一个最长等待时间,超过时间就直接放弃,不要傻等。比如你调用GPT的接口,设置10秒超时,如果10秒还没返回结果,就直接终止这个请求,不要一直等着,不然你的Agent会被卡住,后面的用户请求都处理不了。

核心概念二:重试策略

你点外卖,骑手超时了没送到,你会给骑手打个电话问一下,或者再等10分钟,不会直接就退款对吧?重试就是当一件事失败了,不是直接放弃,而是再试几次,比如调用大模型返回503服务不可用,你可以等1秒再试一次,说不定就成功了。当然也不能无限重试,不然会把大模型的接口打挂。

核心概念三:幂等设计

你在淘宝上下单,不小心点了两次付款,你肯定不希望被扣两次钱对吧?幂等就是保证同一个操作执行1次和执行100次的效果完全一样。比如你的Agent是用来发营销短信的,同一个用户的同一个营销请求,不管执行多少次,都只会给用户发一条短信,不会发100条导致被投诉。

核心概念四:服务降级

你去饭店吃饭,你点的菜卖完了,服务员会给你推荐类似的菜,不会直接把你赶出去对吧?降级就是当核心依赖不可用的时候,用备用方案返回一个可用的结果,不要直接报错。比如大模型接口完全挂了,你的Agent可以返回一个提前准备好的模板回答,告诉用户“当前咨询量较大,我们会稍后联系你”,而不是直接返回“系统错误”。

核心概念之间的关系

四个策略不是孤立的,而是Harness层的四个核心组件,互相配合完成容错:

  1. 超时是重试的前提:如果没有超时,调用卡住几个小时,重试根本无法触发;
  2. 重试是容错的第一手段:大部分外部故障都是临时的,重试几次就能解决,不需要触发降级;
  3. 幂等是重试的基础:如果接口不是幂等的,重试会导致重复执行,反而引发资损问题;
  4. 降级是最后的兜底手段:重试多次都失败的时候,才触发降级,保证服务可用。

我们可以用一个表格对比四个策略的核心属性:

策略 解决的核心问题 适用场景 风险点 优先级
超时 避免资源被长时间占用 所有外部调用、长耗时任务 超时时间设置过短导致正常请求被中断,过长导致资源浪费 最高(所有调用必须设置超时)
重试 解决临时故障导致的调用失败 网络抖动、服务临时不可用、限流等临时异常 重试次数过多导致服务雪崩、非幂等接口重试导致资损 次高(临时故障占所有故障的70%以上)
幂等 避免重复执行导致的副作用 所有写操作、有副作用的调用、需要重试的接口 幂等键泄露导致恶意请求、幂等存储故障导致幂等失效 次高(只要有重试就必须做幂等)
降级 避免核心依赖故障导致整个服务不可用 核心依赖完全不可用、重试多次失败 兜底结果不符合用户预期、降级逻辑本身有故障 最低(最后兜底手段)

核心概念架构文本示意图

┌─────────────────────────────────────────────────────────┐
│                     AI Agent 业务逻辑层                  │
└───────────────────────────┬─────────────────────────────┘
                            │
┌───────────────────────────▼─────────────────────────────┐
│                  AI Agent Harness 管控层                 │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐  │
│  │ 超时控制 │  │ 重试策略 │  │ 幂等设计 │  │ 服务降级 │  │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘  │
└───────────────────────────┬─────────────────────────────┘
                            │
┌───────────────────────────▼─────────────────────────────┐
│                     外部依赖层                           │
│  大模型API | 工具API | 第三方服务 | 数据库 | 缓存     │
└─────────────────────────────────────────────────────────┘

Mermaid 架构图

包含

包含

包含

包含

调用

Harness

string

超时配置

string

重试配置

string

幂等配置

string

降级配置

超时控制

int

超时时长

string

超时处理逻辑

重试策略

int

重试次数

string

退避规则

string

重试触发条件

幂等设计

string

幂等键生成规则

string

幂等存储

int

过期时间

服务降级

string

降级触发条件

string

兜底逻辑

外部依赖

string

大模型API

string

工具API

string

第三方服务

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. 指数退避:每次重试等待时间翻倍,比如第一次等1秒,第二次2秒,第三次4秒,优点是减少对服务端的压力,缺点是等待时间越来越长;
  3. 抖动退避:在指数退避的基础上加上随机抖动,比如第一次等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. 幂等设计原理与实现

幂等的核心是全局唯一的幂等键幂等记录存储

幂等键生成规则

幂等键必须全局唯一,能够唯一标识一个请求,常见的生成方式:

  1. 前端生成:用户发起请求的时候前端生成一个UUID作为幂等键,放在请求头里;
  2. 后端生成:根据请求的核心参数哈希生成,比如用户ID+请求内容的MD5值作为幂等键。
幂等校验流程
  1. 收到请求后先生成/提取幂等键;
  2. 去存储里查这个幂等键是否已经存在,如果存在直接返回历史结果;
  3. 如果不存在,执行请求逻辑,执行成功后把幂等键和结果存到存储里,设置过期时间。
代码实现(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. 服务降级原理与实现

降级的核心是熔断机制兜底逻辑

熔断机制

当某个依赖的错误率超过阈值的时候,就暂时停止调用这个依赖,直接返回兜底结果,等一段时间后再尝试恢复调用,避免雪崩。常见的熔断状态:

  1. 闭合状态:正常调用依赖,统计错误率;
  2. 断开状态:错误率超过阈值,停止调用,直接返回兜底;
  3. 半开状态:断开一段时间后,允许少量请求尝试调用,看看是否恢复。
代码实现(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(1p)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(10.8)3=10.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=1ek2/(2N)
其中NNN是所有可能的ID数量,UUID v4的N=2122N=2^{122}N=2122,如果每天生成10亿个幂等键,连续生成100年,冲突概率约为10−1810^{-18}1018,比你中彩票的概率还低。

4. 熔断触发阈值计算

熔断的错误率阈值需要根据业务的容忍度设置,公式为:
Ethreshold=failed_requeststotal_requests∗100%E_{threshold} = \frac{failed\_requests}{total\_requests} * 100\%Ethreshold=total_requestsfailed_requests100%
一般来说,核心业务的错误率阈值设为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)

代码解读

  1. 我们把四个策略用装饰器的方式整合到了一起,顺序是:熔断降级 -> 幂等校验 -> 重试 -> 超时,这个顺序是最优的,避免不必要的调用;
  2. 幂等键用用户ID+问题的哈希生成,同一个用户问同一个问题直接返回历史结果,不需要重复调用大模型;
  3. 重试只重试临时异常,最多3次,用抖动指数退避;
  4. 熔断错误率超过50%就触发降级,返回兜底回答;
  5. 所有调用设置10秒超时,避免卡住。

实际应用场景

  1. 客服Agent:7*24小时可用,大模型挂了也能返回兜底提示,不会丢失用户线索,重复提问不会重复计费;
  2. RAG Agent:调用向量数据库超时重试,重试失败返回通用回答,避免用户看到错误;
  3. 自动化工作流Agent:调用第三方工具(比如飞书、钉钉、企业微信)的时候做幂等,不会重复发消息、重复创建工单;
  4. 生成式AI应用:生成图片、视频的时候设置长超时,重试多次,失败返回兜底模板,提升用户体验。

工具和资源推荐

  1. 重试库:Python的tenacity,Java的Spring Retry、Resilience4j;
  2. 熔断库:Python的pybreaker,Java的Resilience4j、Hystrix;
  3. 幂等存储:Redis、MongoDB、TiKV;
  4. 监控工具:Prometheus+Grafana,统计重试率、超时率、降级率、熔断触发次数;
  5. 参考资料:《谷歌SRE手册》、OpenAI官方错误处理最佳实践、tenacity官方文档、pybreaker官方文档。

未来发展趋势与挑战

发展趋势

时间 可靠性阶段 核心特征
2022年以前 无可靠性设计 Agent出错直接崩溃,没有任何容错
2023年 基础容错 简单的重试和超时,没有幂等和降级
2024年 标准化Harness层 四大策略成为标配,Harness层成为Agent的核心组件
2025年以后 自适应可靠性 动态调整超时时间、重试次数、熔断阈值,用大模型判断降级的兜底结果是否合适

挑战

  1. 大模型非结构化返回的幂等校验:同一个请求大模型可能返回不同的结果,怎么判断是否幂等是未来的难点;
  2. 多工具调用的事务一致性:Agent调用多个工具,其中一个失败了,怎么回滚之前的操作,保证一致性;
  3. 降级结果的准确性:兜底结果不能和用户的问题偏差太大,需要用小模型或者模板生成合适的兜底回答。

总结:学到了什么?

核心概念回顾

  • 超时控制:给调用设置最长等待时间,避免资源被卡住;
  • 重试策略:临时故障多试几次,提升成功率;
  • 幂等设计:同一个请求执行多次效果一样,避免资损;
  • 服务降级:核心依赖挂了返回兜底结果,保证服务可用。

概念关系回顾

四个策略互相配合,超时是前提,重试是第一手段,幂等是重试的基础,降级是最后兜底,Harness层把四个策略整合起来,保证Agent的高可用。


思考题:动动小脑筋

  1. 如果你的Agent要调用支付接口扣用户的钱,你怎么设计幂等策略,保证不会重复扣钱?
  2. 如果你的Agent依赖10个外部工具,其中3个是非核心工具,你怎么设计降级策略,保证核心功能可用?
  3. 如果大模型的接口耗时波动很大,有时候3秒返回,有时候20秒返回,你怎么设置超时时间?

附录:常见问题与解答

  1. Q:重试会不会导致大模型接口被限流?
    A:会,所以要加抖动退避,控制重试次数,同时限流的时候延长重试等待时间,比如遇到429错误就等30秒再重试。
  2. Q:幂等键的过期时间设多久合适?
    A:要大于最长的请求执行时间的2倍,比如你的Agent最长执行时间是1小时,过期时间就设为2小时以上,核心业务可以设为7天或者永久。
  3. Q:降级的兜底结果不符合用户需求怎么办?
    A:做分级降级,一级用缓存的历史结果,二级用小模型生成简化结果,三级用友好提示,同时兜底结果要提前测试,不要出现错误信息。
  4. Q:四大策略的顺序怎么安排最好?
    A:最优顺序是:熔断降级 -> 幂等校验 -> 重试 -> 超时,先做熔断避免调用已经挂的服务,再做幂等避免重复执行,再重试,最后超时控制。

扩展阅读 & 参考资料

  1. 《Site Reliability Engineering: How Google Runs Production Systems》(谷歌SRE手册)
  2. OpenAI官方错误处理指南:https://platform.openai.com/docs/guides/error-codes
  3. tenacity官方文档:https://tenacity.readthedocs.io/
  4. pybreaker官方文档:https://pybreaker.readthedocs.io/
  5. 幂等设计最佳实践:https://martinfowler.com/articles/patterns-of-distributed-systems/idempotent-receiver.html

(全文约12800字)

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐