第12篇:Agent 可靠性工程:幻觉、错误、崩溃如何系统性解决
用严格的JSON Schema定义工具的函数名、参数、必填项、数据类型、枚举值、格式要求,强制大模型遵循,这是减少参数/格式错误的核心手段。"description": "根据商品ID查询商品价格和库存信息","description": "商品唯一ID,必填,格式为GOODS+数字,比如GOODS123456"},"description": "仓库ID,可选,默认查询全国总仓",},"requ
本文为《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. 事后:幻觉修正与兜底,绝对禁止有害幻觉输出
检测到幻觉后,我们需要有完善的处理机制,而不是直接把错误内容输出给用户。
- 多轮迭代修正:把检测到的幻觉问题反馈给大模型,让它基于上下文重新生成回答,最多迭代3次;
- 风险分级处理:高风险幻觉(比如编造产品政策、虚假承诺)直接触发兜底,禁止重新生成;中低风险幻觉可以迭代修正;
- 兜底策略:多次迭代仍存在幻觉,直接输出合规的兜底话术,绝对禁止输出任何有幻觉风险的内容;
- 样本沉淀与持续优化:把检测到的幻觉样本沉淀下来,优化提示词、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)预校验钩子:调用前先校验,不执行非法请求
在工具调用执行前,增加预校验环节,不合法的请求直接打回,不执行实际调用,减少无效请求和失败:
- 格式校验:检查工具调用的JSON格式是否合法,函数名是否存在;
- Schema校验:用JSON Schema校验参数是否符合要求,必填字段是否缺失,类型是否正确;
- 权限校验:检查当前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流程中断。
- 工具级降级:单个工具调用失败,触发该工具的兜底逻辑,比如查询天气API失败,返回“暂时无法查询天气信息,请稍后再试”,而不是让整个Agent崩溃;
- 流程级降级:在工作流中,工具调用节点失败,自动走降级分支,比如自动回复节点失败,直接走人工升级分支;
- 全局兜底:任何工具调用失败,都有最终的兜底话术,绝对不能把报错信息、堆栈信息输出给用户。
四、核心模块三:输出格式强校验(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解析修复库)、jsonfixer、pyjson5。
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完全一致:
- 事前:用DTD/XML Schema做硬约束,提示词明确格式要求;
- 事中:流式增量校验标签闭合情况;
- 事后:用
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节点都配置降级分支,大模型不可用时,自动走规则化分支;
- 功能开关:配置全局的功能开关,大模型不可用时,自动关闭非核心的智能功能,保留核心的基础功能,保证系统可用。
第四层:全局兜底与熔断保护,极端场景的最终防线
- 统一兜底话术:任何大模型相关的功能失败,都输出友好的兜底话术,比如“系统暂时繁忙,请稍后再试,或联系人工客服”,绝对不能把报错信息、堆栈信息输出给用户;
- 熔断保护:用熔断器模式,当大模型连续失败达到阈值,触发熔断,一段时间内不再调用大模型,直接走兜底,避免把大模型服务打崩,也避免浪费配额;
# 熔断器实现(基于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) - 告警通知:触发模型切换、熔断、降级的时候,自动发送告警(邮件、企业微信、短信)给运维和业务人员,及时处理问题。
六、全链路可靠性闭环:从单次修复到持续优化
前面的四大模块解决了单点的可靠性问题,而企业级的可靠性工程,需要建立一个全链路、持续迭代的闭环体系:
- 输入层:参数校验、权限校验、内容安全校验,过滤非法输入;
- 推理层:RAG增强、提示词约束、幻觉检测与修正,保证输出内容准确;
- 执行层:工具调用预校验、分级重试、降级兜底,保证执行不中断;
- 输出层:格式强校验、自动修复、兜底格式,保证下游系统可解析;
- 服务层:多模型容灾、熔断降级、超时控制,保证服务不崩溃;
- 可观测层:全链路指标采集、日志审计、监控告警,及时发现问题;
- 优化层:失败样本沉淀、提示词优化、模型优化、流程迭代,持续提升可靠性。
核心可靠性指标(SLA)
我们需要用可量化的指标来衡量Agent系统的可靠性,核心指标包括:
- Agent执行成功率:≥99.9%
- 幻觉发生率:≤1%
- 工具调用成功率:≥99.5%
- 输出格式合规率:100%
- 系统可用性:≥99.95%
- 平均响应时间:≤3s
七、核心总结+下一篇预告
核心总结
本文我们体系化拆解了Agent可靠性工程的完整方案,解决了Agent从Demo走向生产环境的四大核心痛点:
- 建立了「事前预防-事中检测-事后修正」的幻觉管控体系,把幻觉发生率降到1%以内;
- 实现了「预校验-分级重试-降级兜底」的工具调用全流程管控,工具调用成功率提升到99.5%以上;
- 搭建了「生成前约束-生成中校验-生成后修复」的格式管控体系,实现100%的格式合规率;
- 设计了「请求重试-模型降级-功能降级-全局兜底」的四级容灾体系,保证大模型不稳定时系统也不会崩溃。
Agent可靠性工程的核心,不是追求100%消除大模型的随机性,而是通过工程化的手段,把随机性锁在可控范围内,用规则兜底、用智能增强,让Agent系统满足企业生产环境的确定性、稳定性、可用性要求。
下一篇预告
本文我们解决了Agent系统的可靠性问题,下一篇《第13篇:Agent企业级落地:权限管控、数据安全与合规审计》,我们会聚焦Agent企业级落地的最后一公里,详解多租户权限管控、数据脱敏与安全、合规审计、私有化部署、等保合规要求,让你的Agent系统真正满足企业级生产环境的安全合规要求。
更多推荐


所有评论(0)