本文为《AI Agent 企业级实战:从原理到落地,构建自主智能体》专栏第 12 篇,承接前11篇Agent核心能力、多智能体协作、工作流编排、垂直场景落地的内容,直击Agent从Demo走向生产环境的头号拦路虎:不可靠性
本文全程站在企业级生产落地视角,体系化拆解Agent全链路的可靠性风险,针对幻觉检测与抑制、工具调用失败重试、输出格式强校验、大模型不稳定降级四大核心问题,提供可落地、可复用、生产级的工程化解决方案,所有代码与前11篇完全兼容,看完就能搭建一套SLA 99.9%的高可靠Agent系统。

一、开篇:为什么Agent落地必须做可靠性工程?

前11篇我们已经完整实现了Agent的四大核心能力、多智能体团队协作、RAG增强、工作流编排、爬虫/自动化测试/客服工单等垂直场景落地,能跑通几乎所有业务Demo。但当我们把Agent部署到企业生产环境时,会立刻遇到致命问题:

  • 幻觉频发:大模型编造事实、逻辑错误、偏离上下文,导致客服回复错误信息、爬虫解析错误数据、工单分类错误,给企业带来业务风险;
  • 工具调用随机失败:参数错误、格式错误、API报错、调用逻辑混乱,导致Agent执行中断,流程卡死;
  • 输出格式不可控:明明要求输出JSON,却经常多了注释、少了括号、字段缺失,导致下游系统解析失败,直接崩溃;
  • 大模型服务不稳定:超时、限流、5xx报错、服务不可用,导致整个Agent系统直接瘫痪,无任何兜底能力。

这些问题的本质,是大模型的天生随机性,与企业生产环境要求的确定性、稳定性、可用性之间的核心矛盾。Demo场景里,我们可以接受60%的成功率,错了重新跑一次就行;但生产环境里,我们需要99.9%以上的成功率,绝对不允许出现错误信息、系统崩溃。

Agent可靠性工程,就是解决这个矛盾的体系化方案:它是一套覆盖「输入-推理-工具调用-输出-容灾」全链路的工程化方法论和实践,通过事前预防、事中检测、事后修复、兜底容灾的全闭环管控,把大模型的随机性锁在可控范围内,让Agent系统在生产环境稳定、可靠、持续运行。

Agent全链路可靠性风险拆解

我们把Agent的可靠性风险,对应到全链路的5个层级,正好对应本文要解决的四大核心问题:

链路层级 核心风险 本文对应解决方案
推理层 幻觉(事实编造、逻辑错误、上下文偏离) 幻觉检测与抑制体系
执行层 工具调用失败(参数/格式/API/逻辑错误) 分级重试与降级策略
输出层 格式不符合要求(JSON/XML结构错误、字段缺失) 输出格式强校验与自动修复
服务层 大模型服务不稳定(超时、限流、宕机) 多模型容灾与分级降级方案
全局层 系统崩溃、无兜底、不可观测 全链路可靠性闭环与监控告警

二、核心模块一:幻觉检测与抑制,从根源解决Agent的“胡说八道”

幻觉是Agent可靠性的头号敌人,也是最难彻底解决的问题。我们不追求100%消除幻觉(这是当前大模型技术无法实现的),但通过事前预防、事中检测、事后修正的三层闭环体系,可以把幻觉发生率从30%+降到1%以内,同时绝对禁止有害幻觉的输出。

1. 事前:幻觉预防,从根源减少幻觉发生

预防的核心是锁死大模型的自由发挥空间,强制它在我们给定的边界内输出,从根源上减少幻觉产生的可能。

(1)提示词硬约束,给Agent戴上“紧箍咒”

这是成本最低、效果最明显的预防手段,通过明确的禁止性指令和边界约束,大幅降低幻觉概率。

生产级幻觉预防Prompt模板
【核心规则-绝对遵守】
1.  事实性约束:你所有的回答必须完全基于提供的上下文/检索到的文档,禁止编造任何未在上下文中出现的事实、数据、案例、联系方式;
2.  边界约束:你只能回答与[产品客服]相关的问题,禁止回答任何超出职责边界的内容,禁止编造不存在的功能、政策、活动;
3.  未知约束:如果问题超出上下文范围、你无法确定答案,必须直接回答“非常抱歉,这个问题我暂时无法为您解答,请联系人工客服”,禁止编造答案;
4.  引用约束:所有事实性结论必须标注对应的上下文来源,格式为[来源文档X-段落X];
5.  逻辑约束:必须先输出思考过程,再输出最终答案,思考过程必须清晰说明答案的上下文依据,禁止逻辑跳跃。

【错误示例-幻觉行为】
用户问:你们的产品支持无线充电吗?
上下文无相关内容,回答:我们的产品支持最高50W无线充电。(编造事实,幻觉)

【正确示例-合规行为】
用户问:你们的产品支持无线充电吗?
上下文无相关内容,回答:非常抱歉,这个问题我暂时无法为您解答,请联系人工客服。(合规,无幻觉)
(2)RAG检索增强,用事实替代自由生成

所有事实性回答,必须先检索、后生成,强制大模型只能基于检索到的权威文档生成内容,完全禁止大模型用自身参数知识回答事实性问题。

  • 强制要求:所有回答必须有对应的检索片段作为依据,无依据的内容必须删除;
  • 边界限制:检索不到相关内容时,直接触发兜底话术,禁止大模型自由发挥;
  • 多来源交叉验证:关键事实必须有2个以上的检索来源交叉验证,不一致的内容标记为高风险。
(3)参数与角色固化,减少随机性
  • 温度参数(temperature):事实性场景强制设置为0,创意类场景最高不超过0.3,大幅降低输出随机性;
  • 角色与职责固化:给Agent定死明确的职责边界,禁止跨职责回答,比如客服Agent不能回答技术开发问题,减少不必要的自由发挥;
  • 少样本示例:在提示词中加入正确的回答示例和错误的幻觉示例,让大模型明确知道什么能做、什么不能做。

2. 事中:幻觉实时检测,在输出前拦截风险

哪怕做了完善的预防,还是可能出现幻觉,我们需要在大模型生成内容后、输出给用户/下游系统前,完成实时检测,拦截幻觉内容。

(1)自检链(Self-Check):让大模型自己检查自己

这是最易落地、效果最显著的幻觉检测手段,核心逻辑是:让大模型扮演质检员,检查自己的回答是否存在幻觉。

生产级自检Prompt模板
你是专业的内容质检专家,需要检查下面的回答是否存在幻觉问题。
【用户原始问题】:{user_query}
【参考上下文】:{context}
【待检查回答】:{model_answer}

【幻觉判定标准】
1.  事实编造:回答中出现了参考上下文中没有的事实、数据、功能、政策;
2.  上下文偏离:回答的内容与用户问题、参考上下文无关,答非所问;
3.  逻辑错误:回答的逻辑链断裂、前提错误、结论与依据不符;
4.  过度承诺:回答中做出了参考上下文中没有的承诺、保证。

【输出要求】
必须严格输出JSON格式,禁止输出其他内容:
{
    "has_hallucination": true/false, // 是否存在幻觉
    "hallucination_type": "事实编造/上下文偏离/逻辑错误/无", // 幻觉类型
    "hallucination_content": "具体的幻觉内容片段", // 无幻觉则为空
    "risk_level": "高/中/低/无", // 风险等级
    "is_pass": true/false // 是否通过质检,无幻觉则为true
}
自检链工程化实现代码
def self_check_hallucination(user_query: str, context: str, model_answer: str) -> dict:
    """
    幻觉自检核心函数
    :param user_query: 用户原始问题
    :param context: 参考上下文/RAG检索内容
    :param model_answer: 大模型生成的待检查回答
    :return: 质检结果
    """
    check_prompt = f"""
    你是专业的内容质检专家,需要检查下面的回答是否存在幻觉问题。
    【用户原始问题】:{user_query}
    【参考上下文】:{context}
    【待检查回答】:{model_answer}

    【幻觉判定标准】
    1.  事实编造:回答中出现了参考上下文中没有的事实、数据、功能、政策;
    2.  上下文偏离:回答的内容与用户问题、参考上下文无关,答非所问;
    3.  逻辑错误:回答的逻辑链断裂、前提错误、结论与依据不符;
    4.  过度承诺:回答中做出了参考上下文中没有的承诺、保证。

    【输出要求】
    必须严格输出纯JSON格式,禁止输出任何其他内容、注释、markdown格式:
    {{
        "has_hallucination": true/false,
        "hallucination_type": "事实编造/上下文偏离/逻辑错误/无",
        "hallucination_content": "具体的幻觉内容片段",
        "risk_level": "高/中/低/无",
        "is_pass": true/false
    }}
    """
    # 调用大模型执行质检,温度设为0,保证输出稳定
    result = llm.chat([{"role": "user", "content": check_prompt}], temperature=0)
    # 格式校验与解析
    try:
        if result.startswith("```json"):
            result = result.replace("```json", "").replace("```", "").strip()
        return json.loads(result)
    except Exception as e:
        return {
            "has_hallucination": True,
            "hallucination_type": "格式错误",
            "hallucination_content": "质检结果解析失败",
            "risk_level": "高",
            "is_pass": False
        }
(2)进阶检测方案
  • 交叉校验:用2个不同的大模型/不同的提示词,生成同一个问题的回答,交叉比对内容一致性,不一致的地方标记为高幻觉风险;
  • 事实一致性校验:用文本相似度模型,把生成的回答和参考上下文做片段级比对,检查是否有超出上下文的事实内容;
  • 逻辑链校验:强制大模型输出思考过程(Chain of Thought),一步步检查逻辑链的前提、推理、结论是否通顺,是否有逻辑跳跃、错误前提。

3. 事后:幻觉修正与兜底,绝对禁止有害幻觉输出

检测到幻觉后,我们需要有完善的处理机制,而不是直接把错误内容输出给用户。

  1. 多轮迭代修正:把检测到的幻觉问题反馈给大模型,让它基于上下文重新生成回答,最多迭代3次;
  2. 风险分级处理:高风险幻觉(比如编造产品政策、虚假承诺)直接触发兜底,禁止重新生成;中低风险幻觉可以迭代修正;
  3. 兜底策略:多次迭代仍存在幻觉,直接输出合规的兜底话术,绝对禁止输出任何有幻觉风险的内容;
  4. 样本沉淀与持续优化:把检测到的幻觉样本沉淀下来,优化提示词、RAG知识库、少样本示例,从根源减少后续幻觉的发生。

三、核心模块二:工具调用失败重试策略,解决Agent执行中断问题

工具调用是Agent与外部世界交互的核心能力,也是Agent执行中断、流程卡死的重灾区。我们需要建立一套**「事前预防-分级重试-降级兜底」的全流程管控体系**,把工具调用成功率从70%提升到99.5%以上。

1. 先拆解:工具调用失败的5大常见原因

我们先把工具调用失败的原因分类,不同的原因对应完全不同的处理策略,绝对不能无脑重试:

失败类型 常见场景 是否可重试 核心处理方案
格式错误 函数名错误、JSON格式错误、括号不闭合 不可重试,可修正 格式预校验+错误反馈修正
参数错误 必填字段缺失、参数类型错误、参数值不符合要求 不可重试,可修正 Schema强校验+参数修正
临时执行错误 网络超时、API限流429、服务临时5xx错误 可重试 指数退避重试
永久执行错误 权限不足、API密钥失效、接口下线 不可重试,不可修正 直接降级兜底
逻辑错误 调用了错误的工具、调用顺序错误 不可重试,需重新规划 重新规划工具调用链

2. 事前:工具调用失败预防,从根源减少错误

(1)工具定义强约束,用JSON Schema锁死规范

用严格的JSON Schema定义工具的函数名、参数、必填项、数据类型、枚举值、格式要求,强制大模型遵循,这是减少参数/格式错误的核心手段。

生产级工具定义示例(兼容OpenAI Function Call)
{
    "type": "function",
    "function": {
        "name": "query_goods_price",
        "description": "根据商品ID查询商品价格和库存信息",
        "parameters": {
            "type": "object",
            "properties": {
                "goods_id": {
                    "type": "string",
                    "description": "商品唯一ID,必填,格式为GOODS+数字,比如GOODS123456"
                },
                "warehouse_id": {
                    "type": "string",
                    "description": "仓库ID,可选,默认查询全国总仓",
                    "enum": ["WH001", "WH002", "WH003"]
                }
            },
            "required": ["goods_id"], // 强制必填字段
            "additionalProperties": false // 禁止传入未定义的参数
        }
    }
}
(2)预校验钩子:调用前先校验,不执行非法请求

在工具调用执行前,增加预校验环节,不合法的请求直接打回,不执行实际调用,减少无效请求和失败:

  1. 格式校验:检查工具调用的JSON格式是否合法,函数名是否存在;
  2. Schema校验:用JSON Schema校验参数是否符合要求,必填字段是否缺失,类型是否正确;
  3. 权限校验:检查当前Agent是否有调用该工具的权限。

3. 核心:分级重试策略,不同错误用不同的重试方案

重试的核心原则:只对临时、可恢复的错误进行重试,对不可恢复的错误绝对不重试,避免无效请求和资源浪费

(1)可重试错误:指数退避+抖动重试策略

针对网络超时、API限流429、服务临时5xx错误这类临时故障,我们采用带抖动的指数退避重试,这是业界公认的最优重试策略,既可以快速恢复,又能避免给服务端造成压力。

生产级重试实现(基于tenacity库)
import tenacity
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_result, retry_if_exception_type

# 定义可重试的异常类型
RETRY_EXCEPTIONS = (
    requests.exceptions.Timeout,  # 超时
    requests.exceptions.ConnectionError,  # 网络连接错误
    requests.exceptions.HTTPError,  # HTTP错误
)

# 定义可重试的响应状态码
def is_retryable_response(response: dict) -> bool:
    """判断是否是可重试的响应"""
    retryable_status_codes = [429, 500, 502, 503, 504]
    return response.get("status_code") in retryable_status_codes

# 带抖动的指数退避重试装饰器
@retry(
    stop=stop_after_attempt(3),  # 最多重试3次
    wait=wait_exponential(multiplier=1, min=1, max=10),  # 指数退避,1s/2s/4s
    retry=(
        retry_if_exception_type(RETRY_EXCEPTIONS) |  # 可重试异常
        retry_if_result(is_retryable_response)  # 可重试响应
    ),
    reraise=True,  # 重试失败后重新抛出异常
)
def tool_call_with_retry(tool_func, *args, **kwargs):
    """
    带重试策略的工具调用函数
    :param tool_func: 实际的工具执行函数
    :param args: 工具参数
    :param kwargs: 工具关键字参数
    :return: 工具执行结果
    """
    return tool_func(*args, **kwargs)
(2)可修正不可重试错误:错误反馈+修正重试

针对参数错误、格式错误这类问题,重试是没用的,我们需要把错误信息明确反馈给大模型,让它修正后重新生成工具调用,最多修正2次。

修正重试核心逻辑
def tool_call_with_correction(llm, tool_definitions: list, user_query: str, max_correction_times: int = 2):
    """
    带错误修正的工具调用函数
    :param llm: 大模型实例
    :param tool_definitions: 工具定义列表
    :param user_query: 用户原始问题
    :param max_correction_times: 最大修正次数
    :return: 工具调用结果
    """
    correction_times = 0
    last_error = ""
    while correction_times <= max_correction_times:
        # 1. 生成工具调用
        tool_call_result = llm.generate_tool_call(
            tool_definitions=tool_definitions,
            user_query=user_query,
            error_info=last_error
        )
        # 2. 预校验
        check_result, error_msg = pre_check_tool_call(tool_call_result, tool_definitions)
        if check_result:
            # 3. 校验通过,执行工具调用
            return execute_tool_call(tool_call_result)
        # 4. 校验不通过,记录错误,进入下一轮修正
        correction_times += 1
        last_error = error_msg
        print(f"工具调用校验失败,第{correction_times}次修正,错误信息:{error_msg}")
    # 5. 修正次数耗尽,抛出异常
    raise Exception(f"工具调用修正失败,超过最大修正次数,最后错误:{last_error}")

4. 事后:降级与兜底,保证流程不中断

重试和修正都失败后,我们需要有完善的降级兜底策略,绝对不能让整个Agent流程中断。

  1. 工具级降级:单个工具调用失败,触发该工具的兜底逻辑,比如查询天气API失败,返回“暂时无法查询天气信息,请稍后再试”,而不是让整个Agent崩溃;
  2. 流程级降级:在工作流中,工具调用节点失败,自动走降级分支,比如自动回复节点失败,直接走人工升级分支;
  3. 全局兜底:任何工具调用失败,都有最终的兜底话术,绝对不能把报错信息、堆栈信息输出给用户。

四、核心模块三:输出格式强校验(JSON/XML),解决下游系统解析崩溃问题

这是生产环境中最常见、最容易踩坑的问题:明明在提示词里反复要求输出JSON,大模型还是经常输出多余的注释、少了括号、字段缺失、用markdown代码块包裹,导致下游系统解析失败,直接崩溃。

我们需要建立一套**「生成前约束-生成中校验-生成后修复」的三层格式管控体系**,实现100%的格式合规率,保证下游系统永远能解析到合法的格式,绝对不会因为格式问题崩溃。

1. 事前:格式生成硬约束,让大模型尽量输出正确格式

(1)提示词强约束,锁死输出格式
生产级JSON输出Prompt模板
【绝对遵守的输出规则】
1.  你必须严格输出纯JSON格式,禁止输出任何其他内容,包括但不限于解释、说明、注释、问候语、markdown格式、代码块;
2.  禁止输出```json、```这类markdown代码块包裹符号,直接输出JSON对象本身;
3.  JSON必须严格符合语法规范,禁止出现多余的逗号、不闭合的引号、不匹配的括号、换行符;
4.  JSON必须包含以下所有字段,禁止缺失字段、新增未定义的字段,字段类型必须严格符合要求:
    - code: 数字类型,状态码,200表示成功,500表示失败
    - message: 字符串类型,状态描述
    - data: 对象类型,业务数据,必须包含user_id、user_name两个字符串字段
5.  如果生成失败,也必须输出符合格式的JSON,比如{"code": 500, "message": "生成失败", "data": {"user_id": "", "user_name": ""}}

【正确输出示例】
{"code": 200, "message": "查询成功", "data": {"user_id": "123456", "user_name": "张三"}}

【错误输出示例】
```json
{
    "code": 200,
    "message": "查询成功",
    "data": {
        "user_id": "123456",
        "user_name": "张三"
    }
}

以上是错误示例,禁止用代码块包裹。


#### (2)利用大模型原生JSON模式
现在主流大模型都支持原生JSON输出模式(比如GPT-4 JSON Mode、豆包JSON模式、通义千问JSON模式),可以强制大模型输出合法的JSON格式,这是成本最低、效果最好的手段。
##### 豆包大模型JSON模式调用示例
```python
def chat_with_json_mode(prompt: str) -> str:
    """
    大模型JSON模式调用,强制输出合法JSON
    """
    url = "https://aquasearch.ai/api/v1/chat/completions"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {os.getenv('DOUBAO_API_KEY')}"
    }
    data = {
        "model": "doubao-pro",
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0,
        "response_format": {"type": "json_object"}  # 强制JSON格式
    }
    response = requests.post(url, headers=headers, json=data)
    response.raise_for_status()
    return response.json()["choices"][0]["message"]["content"]

2. 事中:生成时增量校验,提前拦截格式错误

针对流式输出场景,我们可以在大模型生成内容的同时,做增量格式校验,发现格式错误立即中断生成,重新生成,避免生成完一整段错误内容再修正,浪费时间和资源。

  • 增量语法校验:流式输出时,实时解析JSON结构,发现引号不闭合、括号不匹配、多余逗号等问题,立即中断;
  • 结构预校验:生成完成后,先做快速语法校验,无法解析的直接打回修正,不进入下游流程。

3. 事后:三层校验与自动修复,哪怕生成错了也能救回来

这是格式管控的最后一道防线,哪怕大模型输出的格式有问题,我们也能通过自动修复,生成合法的格式,保证下游系统不会崩溃。

第一层:语法校验与自动修复

针对JSON语法错误,我们用专业的工具做自动修复,解决90%的常见语法问题:

  • 常见可修复问题:多余的逗号、引号不闭合、括号不匹配、换行符、多余的注释、markdown代码块包裹、多余的文本内容;
  • 修复工具:demjson3(兼容性最强的JSON解析修复库)、jsonfixerpyjson5
JSON语法自动修复代码
import demjson3

def fix_and_parse_json(raw_content: str) -> Tuple[bool, dict, str]:
    """
    自动修复并解析JSON
    :param raw_content: 大模型输出的原始内容
    :return: (是否成功, 解析后的JSON对象, 错误信息)
    """
    # 第一步:预处理,清理多余内容
    # 移除markdown代码块包裹
    if raw_content.startswith("```json"):
        raw_content = raw_content.replace("```json", "").replace("```", "").strip()
    elif raw_content.startswith("```"):
        raw_content = raw_content.replace("```", "").strip()
    # 移除代码块外的多余文本,只保留第一个{到最后一个}的内容
    if "{" in raw_content and "}" in raw_content:
        start_idx = raw_content.find("{")
        end_idx = raw_content.rfind("}") + 1
        raw_content = raw_content[start_idx:end_idx]
    
    # 第二步:用demjson3解析并修复
    try:
        json_data = demjson3.decode(raw_content, strict=False)
        return True, json_data, ""
    except Exception as e:
        return False, {}, f"JSON解析修复失败:{str(e)}"
第二层:JSON Schema合规性校验

语法正确不代表内容正确,我们需要用JSON Schema校验字段是否齐全、类型是否正确、枚举值是否符合要求,保证输出的内容完全符合下游系统的预期。

Schema合规性校验代码
from jsonschema import validate, ValidationError

def validate_json_schema(json_data: dict, schema: dict) -> Tuple[bool, str]:
    """
    JSON Schema合规性校验
    :param json_data: 解析后的JSON对象
    :param schema: 预设的JSON Schema
    :return: (是否通过, 错误信息)
    """
    try:
        validate(instance=json_data, schema=schema)
        return True, ""
    except ValidationError as e:
        # 提取清晰的错误信息
        error_path = "->".join([str(p) for p in e.path])
        error_msg = f"字段[{error_path}]校验失败:{e.message}"
        return False, error_msg

# 预设的JSON Schema示例
USER_INFO_SCHEMA = {
    "type": "object",
    "properties": {
        "code": {"type": "number", "enum": [200, 500]},
        "message": {"type": "string"},
        "data": {
            "type": "object",
            "properties": {
                "user_id": {"type": "string"},
                "user_name": {"type": "string"}
            },
            "required": ["user_id", "user_name"],
            "additionalProperties": False
        }
    },
    "required": ["code", "message", "data"],
    "additionalProperties": False
}
第三层:兜底格式生成,绝对保证下游系统能解析

哪怕语法修复、Schema校验都失败了,我们也必须输出一个符合格式的默认JSON对象,绝对不能让下游系统解析失败。

# 兜底默认JSON对象
DEFAULT_USER_INFO_JSON = {
    "code": 500,
    "message": "生成失败",
    "data": {
        "user_id": "",
        "user_name": ""
    }
}

# 完整的格式处理流程
def process_json_output(raw_content: str, schema: dict, default_json: dict) -> dict:
    """
    完整的JSON输出处理流程
    :param raw_content: 大模型原始输出
    :param schema: JSON Schema
    :param default_json: 兜底默认JSON
    :return: 最终合法的JSON对象
    """
    # 第一步:修复并解析JSON
    parse_success, json_data, parse_error = fix_and_parse_json(raw_content)
    if not parse_success:
        print(f"JSON解析失败:{parse_error},使用兜底JSON")
        return default_json
    # 第二步:Schema合规性校验
    validate_success, validate_error = validate_json_schema(json_data, schema)
    if not validate_success:
        print(f"JSON Schema校验失败:{validate_error},使用兜底JSON")
        return default_json
    # 第三步:校验通过,返回合法JSON
    return json_data
XML格式校验与修复

XML格式的处理逻辑和JSON完全一致:

  1. 事前:用DTD/XML Schema做硬约束,提示词明确格式要求;
  2. 事中:流式增量校验标签闭合情况;
  3. 事后:用lxml库做XML语法修复、Schema校验,失败则输出兜底XML格式。

五、核心模块四:大模型不稳定时的降级方案,解决系统崩溃问题

大模型服务不是100%稳定的,超时、限流、5xx报错、服务宕机是生产环境一定会遇到的问题。我们需要建立一套**「请求重试-模型降级-功能降级-全局兜底」的四级容灾体系**,哪怕大模型服务完全不可用,也能保证系统不会崩溃,核心业务可用。

1. 先拆解:大模型不稳定的6大常见场景

不稳定场景 常见错误码 可恢复性 核心处理方案
请求超时 无错误码,请求超时 临时 超时重试、增加超时时间
频率限制/配额耗尽 429 临时/永久 限流重试、切换备用模型
服务端临时错误 500/502/503/504 临时 指数退避重试
鉴权失败 401/403 永久 直接降级、告警
内容安全拦截 400/内容拒绝 永久 内容兜底、告警
服务完全不可用 连接失败、域名无法解析 永久 全功能降级、熔断、告警

2. 四级容灾降级体系

第一层:请求级重试,解决临时故障

针对超时、429限流、5xx临时错误,我们用指数退避重试策略,快速恢复,这是最轻度的故障处理。

  • 超时重试:第一次超时时间30s,重试时增加到60s,最多重试2次;
  • 429限流重试:根据响应头的Retry-After设置重试间隔,没有的话默认10s/30s/60s递增,最多重试3次;
  • 5xx错误重试:指数退避重试,1s/2s/4s,最多重试3次。
第二层:模型级降级,主模型不可用自动切换备用模型

这是解决大模型服务不稳定的核心方案,核心是多模型容灾池,不要把所有鸡蛋放在一个篮子里。

(1)多模型容灾池设计
模型层级 模型示例 适用场景 降级触发条件
主模型 豆包Pro 日常业务,效果最好 连续3次调用失败、服务不可用
备用模型1 豆包Lite 降级场景,效果接近主模型,成本更低 主模型连续失败3次,自动切换
备用模型2 通义千问/文心一言 跨厂商容灾,主厂商服务完全不可用 主模型+备用模型1连续失败5次,自动切换
兜底模型 本地开源模型(Qwen/Llama) 极端场景,所有公有云服务不可用 所有公有云模型都不可用,自动切换
(2)自动切换与恢复工程化实现
class ModelFailoverPool:
    """多模型容灾池,自动切换与恢复"""
    def __init__(self):
        # 模型优先级列表,从主模型到兜底模型
        self.model_list = [
            {"model_id": "doubao_pro", "name": "豆包Pro", "status": "active", "fail_count": 0},
            {"model_id": "doubao_lite", "name": "豆包Lite", "status": "standby", "fail_count": 0},
            {"model_id": "qwen_turbo", "name": "通义千问", "status": "standby", "fail_count": 0},
            {"model_id": "local_qwen", "name": "本地Qwen", "status": "standby", "fail_count": 0},
        ]
        self.current_model_index = 0
        self.max_fail_count = 3  # 最大失败次数,超过则切换
        self.fail_lock = threading.Lock()

    def get_current_model(self):
        """获取当前激活的模型"""
        with self.fail_lock:
            return self.model_list[self.current_model_index]

    def report_fail(self):
        """上报模型调用失败,触发切换逻辑"""
        with self.fail_lock:
            current_model = self.model_list[self.current_model_index]
            current_model["fail_count"] += 1
            print(f"模型{current_model['name']}调用失败,累计失败{current_model['fail_count']}次")
            # 超过最大失败次数,切换到下一个模型
            if current_model["fail_count"] >= self.max_fail_count:
                current_model["status"] = "failed"
                if self.current_model_index < len(self.model_list) - 1:
                    self.current_model_index += 1
                    new_model = self.model_list[self.current_model_index]
                    new_model["status"] = "active"
                    print(f"切换到备用模型:{new_model['name']}")
                    # 触发告警
                    send_alert(f"主模型{current_model['name']}连续失败,已切换到备用模型{new_model['name']}")
                else:
                    print("所有模型都已失败,触发全局兜底")
                    send_alert("所有大模型都调用失败,系统已触发全局兜底")

    def report_success(self):
        """上报模型调用成功,重置失败计数"""
        with self.fail_lock:
            current_model = self.model_list[self.current_model_index]
            current_model["fail_count"] = 0

    def chat(self, messages: list, temperature: float = 0.3, **kwargs):
        """带容灾的大模型调用"""
        current_model = self.get_current_model()
        try:
            # 调用当前模型
            result = call_llm_model(current_model["model_id"], messages, temperature, **kwargs)
            self.report_success()
            return result
        except Exception as e:
            # 上报失败,触发切换
            self.report_fail()
            # 递归调用,使用新的模型
            return self.chat(messages, temperature, **kwargs)
第三层:功能级降级,大模型完全不可用时,关闭智能功能,保留核心业务

当所有大模型都不可用时,我们需要关闭非核心的智能功能,走规则化兜底,保证系统核心业务可用,而不是完全瘫痪。

  • 场景化规则兜底:比如客服Agent,大模型不可用时,自动回复固定的常见问题话术,引导用户联系人工客服;工单分类Agent,大模型不可用时,所有工单默认分到通用客服组;
  • 工作流分支降级:在工作流中,所有Agent节点都配置降级分支,大模型不可用时,自动走规则化分支;
  • 功能开关:配置全局的功能开关,大模型不可用时,自动关闭非核心的智能功能,保留核心的基础功能,保证系统可用。
第四层:全局兜底与熔断保护,极端场景的最终防线
  1. 统一兜底话术:任何大模型相关的功能失败,都输出友好的兜底话术,比如“系统暂时繁忙,请稍后再试,或联系人工客服”,绝对不能把报错信息、堆栈信息输出给用户;
  2. 熔断保护:用熔断器模式,当大模型连续失败达到阈值,触发熔断,一段时间内不再调用大模型,直接走兜底,避免把大模型服务打崩,也避免浪费配额;
    # 熔断器实现(基于pybreaker)
    import pybreaker
    # 定义熔断器:连续5次失败触发熔断,熔断时间30秒
    circuit_breaker = pybreaker.CircuitBreaker(fail_max=5, reset_timeout=30)
    
    @circuit_breaker
    def llm_call_with_circuit_breaker(messages: list):
        return llm.chat(messages)
    
  3. 告警通知:触发模型切换、熔断、降级的时候,自动发送告警(邮件、企业微信、短信)给运维和业务人员,及时处理问题。

六、全链路可靠性闭环:从单次修复到持续优化

前面的四大模块解决了单点的可靠性问题,而企业级的可靠性工程,需要建立一个全链路、持续迭代的闭环体系

  1. 输入层:参数校验、权限校验、内容安全校验,过滤非法输入;
  2. 推理层:RAG增强、提示词约束、幻觉检测与修正,保证输出内容准确;
  3. 执行层:工具调用预校验、分级重试、降级兜底,保证执行不中断;
  4. 输出层:格式强校验、自动修复、兜底格式,保证下游系统可解析;
  5. 服务层:多模型容灾、熔断降级、超时控制,保证服务不崩溃;
  6. 可观测层:全链路指标采集、日志审计、监控告警,及时发现问题;
  7. 优化层:失败样本沉淀、提示词优化、模型优化、流程迭代,持续提升可靠性。

核心可靠性指标(SLA)

我们需要用可量化的指标来衡量Agent系统的可靠性,核心指标包括:

  • Agent执行成功率:≥99.9%
  • 幻觉发生率:≤1%
  • 工具调用成功率:≥99.5%
  • 输出格式合规率:100%
  • 系统可用性:≥99.95%
  • 平均响应时间:≤3s

七、核心总结+下一篇预告

核心总结

本文我们体系化拆解了Agent可靠性工程的完整方案,解决了Agent从Demo走向生产环境的四大核心痛点:

  1. 建立了「事前预防-事中检测-事后修正」的幻觉管控体系,把幻觉发生率降到1%以内;
  2. 实现了「预校验-分级重试-降级兜底」的工具调用全流程管控,工具调用成功率提升到99.5%以上;
  3. 搭建了「生成前约束-生成中校验-生成后修复」的格式管控体系,实现100%的格式合规率;
  4. 设计了「请求重试-模型降级-功能降级-全局兜底」的四级容灾体系,保证大模型不稳定时系统也不会崩溃。

Agent可靠性工程的核心,不是追求100%消除大模型的随机性,而是通过工程化的手段,把随机性锁在可控范围内,用规则兜底、用智能增强,让Agent系统满足企业生产环境的确定性、稳定性、可用性要求。

下一篇预告

本文我们解决了Agent系统的可靠性问题,下一篇《第13篇:Agent企业级落地:权限管控、数据安全与合规审计》,我们会聚焦Agent企业级落地的最后一公里,详解多租户权限管控、数据脱敏与安全、合规审计、私有化部署、等保合规要求,让你的Agent系统真正满足企业级生产环境的安全合规要求。

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐