1. 为什么我要给自家 LLM 应用“下毒”第一次听到“Chaos Engineering”这个词很多做 LLM 应用的朋友第一反应是这东西不是给微服务、K8s 集群用的吗跟大模型有什么关系我一开始也这么想直到我们线上一个 RAG 问答系统在某个周五下午集体“发疯”——用户问“退货政策”模型一本正经地编了一段根本不存在的条款还引用了伪造的文档编号。排查了三个小时才发现是上游 Provider 返回的 embedding 接口偶发超时检索层拿到的是一批空文档模型只能靠“脑补”填空。那次事故之后我彻底转变了思路LLM 应用的脆弱点跟传统后端完全不是一个维度。传统服务挂了就是 500用户一眼就知道出问题了LLM 应用挂了它可能还在“一本正经地胡说八道”你甚至要过很久才发现。所以主动给 AI 系统下毒、注入故障反而是唯一能提前知道它到底有多抗造的办法。这篇东西就是我这大半年在几个 LLM 项目里做 Chaos Engineering 的完整复盘。核心关键词就几个LLM、Chaos Engineering、故障注入、Python、Provider。我会讲清楚为什么 LLM 场景下的混沌工程跟传统做法不一样、故障注入点该怎么选、用 Python 怎么落地一套可复用的注入框架、以及我踩过的那些坑。适合已经在做 LLM 应用、并且开始被线上稳定性折磨的工程师也适合刚入门想了解“AI 系统怎么测”的朋友。看完你至少能搭出一套最小可用的故障注入流水线直接抄作业那种。2. LLM 应用的故障面到底长什么样2.1 传统混沌工程和 LLM 混沌工程的本质差异传统混沌工程的核心假设是系统是确定性的故障是可枚举的。你注入一个网络延迟服务要么超时要么降级行为可预测。Chaos Monkey 随机杀 Pod你也能通过重试、熔断、限流这套组合拳兜住。LLM 应用完全不是这个逻辑。它的故障面有三层而且层层叠加第一层是基础设施层跟传统服务一样——网络抖动、Provider 限流、超时、413 Payload Too Large、Provider 拒绝请求 schema。这些是硬故障好检测。第二层是模型行为层这才是要命的地方。模型不会“报错”它会“编”。你给它喂了脏数据它不会抛异常它会用流畅的自然语言把错误信息包装成看似合理的答案。这就是所谓的AgentPoison类攻击的核心——通过污染记忆或知识库让 Agent 在特定触发条件下输出攻击者想要的内容。第三层是语义层也是最隐蔽的。比如你问“我是谁、我在找什么、我能提供什么”这类涉及 token 语义的问题模型可能因为上下文窗口被截断、或者检索到的 chunk 顺序错乱给出一个逻辑自洽但事实错误的答案。这种故障你不主动注入永远发现不了。所以 LLM 混沌工程的目标不是“让系统不挂”而是“让系统在胡说八道之前先暴露出来”。2.2 五个必须覆盖的故障注入点我把 LLM 应用的故障注入点归纳成五类每一类都对应真实踩过的坑注入点典型故障检测难度影响面Provider 接口超时、限流、schema 拒绝、413低全链路检索层空结果、脏数据、顺序错乱中答案质量上下文组装截断、拼接错误、token 溢出中语义漂移模型输出幻觉、格式错误、拒答高用户体验记忆/知识库投毒、过期、冲突极高安全Provider 层是最容易注入也最容易检测的。我实测下来只要在 Provider 调用外面包一层代理就能模拟出llm request failed: provider rejected the request schema or tool payload这类错误还有unexpected status 413 payload too large这种。这些错误在真实环境里出现频率不低尤其是你用了多个 Provider 做路由的时候。检索层和上下文组装层是 LLM 特有的。传统服务没有“检索”这个概念但 RAG 系统里检索质量直接决定答案质量。我试过故意让检索返回空列表结果模型不但没报错还根据问题本身编了一段答案用户完全看不出来。模型输出层最难测因为你需要一个“裁判”。这里可以用LLM as judge的思路让另一个模型来判断输出是否偏离预期。但裁判本身也可能被污染所以裁判模型最好跟被测模型不同源。记忆/知识库层是安全重灾区。AgentPoison 那篇论文讲的就是通过污染 Agent 的长期记忆让它在特定 query 下触发恶意行为。这种注入在测试环境做能提前发现你的记忆写入有没有做校验。2.3 为什么必须用 Python 来做这件事有人问混沌工程不是有 Chaos Mesh、Litmus 这些现成工具吗为什么还要自己写原因很简单这些工具不懂 LLM 的语义。它们能注入网络延迟但没法注入“检索返回空结果”或者“上下文被截断到只剩前 100 个 token”。LLM 应用的故障注入80% 的工作是在应用层做的而不是基础设施层。Python 在这个场景下几乎是唯一选择因为LLM 生态的 SDK 基本都是 Python 优先OpenAI、Anthropic、各种开源框架都是做数据构造、prompt 变异、结果比对Python 的字符串处理和数据处理能力最顺手你要 hook 到应用内部Python 的 monkey patch、装饰器、context manager 用起来最自然做 LLM as judge 的时候调模型也是 Python 最方便我试过用 Go 写注入层最后发现光是构造测试数据就比 Python 多写一倍代码果断换回来了。3. 用 Python 搭一套最小可用的故障注入框架3.1 整体架构设计思路我的设计原则是注入层要能透明地插到现有代码里不改业务逻辑。理想情况下业务代码完全不知道自己在被测试。架构分四块注入点注册中心用装饰器标记哪些函数可以被注入故障策略库定义各种故障模式比如超时、返回空、返回脏数据、截断注入控制器决定什么时候注入、注入哪个、注入多久观测与断言层记录注入后的系统行为判断是否触发了预期降级这套东西的核心是装饰器 策略模式。业务代码只需要在关键函数上加一个injectable装饰器剩下的交给框架。为什么用装饰器而不是 AOP 或者中间件因为 LLM 应用的调用链往往很深从 API 入口到 Provider 调用中间可能隔了五六层。中间件只能拦最外层装饰器可以精确到每一个函数。而且装饰器对代码侵入性最小加一行就行删一行就恢复。3.2 注入点注册与装饰器实现先看核心的装饰器代码import functools import random from typing import Callable, Any # 全局注入点注册表 _INJECTION_REGISTRY {} def injectable(name: str): 标记一个函数为可注入点 def decorator(func: Callable) - Callable: _INJECTION_REGISTRY[name] { func: func, active_fault: None, inject_probability: 0.0, } functools.wraps(func) def wrapper(*args, **kwargs): entry _INJECTION_REGISTRY[name] fault entry[active_fault] # 按概率决定是否注入 if fault and random.random() entry[inject_probability]: return fault.execute(func, *args, **kwargs) return func(*args, **kwargs) wrapper._injection_name name return wrapper return decorator这个装饰器的关键设计点注册表用全局字典方便运行时动态开关。测试的时候可以一键把所有注入点打开概率控制不是每次都注入模拟真实环境的偶发性。我一般设 0.1 到 0.3太高了系统直接崩太低测不出问题fault.execute 接收原函数这样故障策略可以选择“完全替换”或者“先调用再破坏”注意装饰器一定要用functools.wraps否则被装饰函数的元信息会丢失调试的时候你会疯掉。我踩过这个坑日志里全是wrapper根本不知道是哪个函数出的问题。3.3 故障策略库的设计故障策略我抽象成一个基类每种故障实现自己的executefrom abc import ABC, abstractmethod class FaultStrategy(ABC): abstractmethod def execute(self, func, *args, **kwargs): pass class TimeoutFault(FaultStrategy): 模拟 Provider 超时 def __init__(self, delay: float 30.0): self.delay delay def execute(self, func, *args, **kwargs): import time time.sleep(self.delay) raise TimeoutError(Injected provider timeout) class EmptyRetrievalFault(FaultStrategy): 模拟检索返回空结果 def execute(self, func, *args, **kwargs): return [] class DirtyDataFault(FaultStrategy): 模拟检索返回脏数据 def __init__(self, dirty_ratio: float 0.5): self.dirty_ratio dirty_ratio def execute(self, func, *args, **kwargs): result func(*args, **kwargs) if not isinstance(result, list): return result # 随机替换一部分为无关内容 for i in range(len(result)): if random.random() self.dirty_ratio: result[i] {content: 这是一段无关的测试文本, score: 0.99} return result class TruncateContextFault(FaultStrategy): 模拟上下文被截断 def __init__(self, keep_ratio: float 0.3): self.keep_ratio keep_ratio def execute(self, func, *args, **kwargs): result func(*args, **kwargs) if isinstance(result, str): keep int(len(result) * self.keep_ratio) return result[:keep] return result class SchemaRejectFault(FaultStrategy): 模拟 Provider 拒绝 schema def execute(self, func, *args, **kwargs): raise ValueError( llm request failed: provider rejected the request schema or tool payload )这几种故障覆盖了我遇到的大部分真实场景。TimeoutFault对应 Provider 超时EmptyRetrievalFault对应检索失败DirtyDataFault对应知识库污染TruncateContextFault对应上下文窗口溢出SchemaRejectFault对应 Provider 拒绝请求。为什么DirtyDataFault要保留原结果再污染而不是直接返回假数据因为真实场景下检索往往能返回一些结果只是其中混了脏数据。直接返回全假数据太容易检测了混入脏数据才能测出你的过滤逻辑有没有用。3.4 注入控制器与运行时开关控制器负责在运行时决定注入什么class InjectionController: def __init__(self): self.enabled False self.scenario None def enable(self, scenario: str, probability: float 0.2): 启用某个故障场景 self.enabled True self.scenario scenario scenario_map { provider_timeout: (provider_call, TimeoutFault(30)), empty_retrieval: (retrieval, EmptyRetrievalFault()), dirty_data: (retrieval, DirtyDataFault(0.5)), truncate_context: (context_build, TruncateContextFault(0.3)), schema_reject: (provider_call, SchemaRejectFault()), } if scenario not in scenario_map: raise ValueError(fUnknown scenario: {scenario}) point_name, fault scenario_map[scenario] entry _INJECTION_REGISTRY.get(point_name) if not entry: raise ValueError(fInjection point not registered: {point_name}) entry[active_fault] fault entry[inject_probability] probability def disable_all(self): 关闭所有注入 self.enabled False for entry in _INJECTION_REGISTRY.values(): entry[active_fault] None entry[inject_probability] 0.0用起来是这样的controller InjectionController() controller.enable(empty_retrieval, probability0.3) # 跑你的测试用例 result run_qa_pipeline(退货政策是什么) # 检查系统是否正确降级 assert 无法确认 in result or 请稍后重试 in result, 系统没有正确降级 controller.disable_all()这套东西的好处是你可以在 CI 里跑也可以在预发环境跑甚至可以在生产环境小流量跑概率设低一点。我一般是在预发环境跑全量生产环境只在灰度实例上跑。4. 实操一次完整的故障注入演练4.1 演练场景设计我拿一个真实的 RAG 问答系统做例子。系统流程是用户提问 → 检索知识库 → 组装上下文 → 调用 LLM → 返回答案。演练目标验证当检索层返回空结果时系统是否会正确降级而不是让模型瞎编。预期行为系统应该返回“暂时无法找到相关信息请稍后重试”或者类似的兜底话术而不是编造答案。4.2 注入前的基线测试先跑一遍正常流程记录基线# 基线测试 questions [ 退货政策是什么, 发货需要多久, 支持哪些支付方式, ] baseline_results [] for q in questions: answer run_qa_pipeline(q) baseline_results.append({question: q, answer: answer}) print(fQ: {q}\nA: {answer}\n)基线跑完确认系统在正常情况下能给出正确答案。这一步很重要因为如果基线本身就有问题注入测试的结果没法解读。4.3 注入空检索故障现在打开空检索注入controller InjectionController() controller.enable(empty_retrieval, probability1.0) # 100% 注入确保触发 injected_results [] for q in questions: answer run_qa_pipeline(q) injected_results.append({question: q, answer: answer}) print(fQ: {q}\nA: {answer}\n) controller.disable_all()我实测下来第一次跑的时候系统直接翻车了。模型在检索为空的情况下根据问题本身编了一段“退货政策”还引用了不存在的文档编号。这就是典型的幻觉。4.4 结果比对与降级验证比对基线结果和注入结果def check_degradation(baseline, injected): 检查系统是否正确降级 issues [] for b, i in zip(baseline, injected): if b[question] ! i[question]: continue # 如果注入后的答案跟基线高度相似说明模型在瞎编 if similarity(b[answer], i[answer]) 0.8: issues.append({ question: i[question], issue: 检索为空但答案与基线相似疑似幻觉, answer: i[answer], }) return issues issues check_degradation(baseline_results, injected_results) for issue in issues: print(f[FAIL] {issue[question]}: {issue[issue]})similarity函数可以用简单的 Jaccard 相似度也可以用 embedding 相似度。我一般用 Jaccard 就够了因为幻觉答案往往跟正确答案用词高度重叠。跑完发现三个问题全部 FAIL说明系统的降级逻辑完全没生效。这就是混沌工程的价值——你不注入永远不知道系统这么脆弱。4.5 修复后的回归验证针对发现的问题我们在检索层加了空结果检测def retrieve_with_guard(query): docs vector_store.search(query, top_k5) if not docs: raise EmptyRetrievalError(检索结果为空) return docs然后在 pipeline 里捕获这个异常返回兜底话术。修复后再跑一遍注入测试三个问题全部 PASS。这个过程我走了大概两轮第一轮发现空检索问题第二轮发现脏数据问题。每轮修复后都要重新跑基线确保修复没有引入新问题。5. 踩过的坑和排查技巧5.1 注入概率设太高导致系统雪崩第一次做注入测试的时候我把概率设成了 1.0结果整个测试环境的所有请求全部失败连日志都刷不出来。后来改成 0.2才既能触发问题又不至于把系统打挂。经验注入概率从 0.1 开始试逐步往上加。生产环境永远不要超过 0.05。5.2 装饰器顺序导致的注入失效Python 装饰器是从下往上执行的。如果你有多个装饰器injectable必须放在最靠近函数的位置# 正确 log_call injectable(retrieval) def retrieve(query): ... # 错误注入会被 log_call 包住可能不生效 injectable(retrieval) log_call def retrieve(query): ...这个坑我踩了两天才发现因为日志看起来一切正常但注入就是不触发。5.3 Provider 错误信息被吞掉LLM 应用里经常有全局异常捕获把 Provider 的错误信息吞掉只返回一个通用的“服务异常”。做注入测试的时候这会导致你根本不知道注入有没有生效。解决办法是在注入层加一个标记比如在异常信息里带上[INJECTED]前缀然后在日志里 grep 这个标记。5.4 常见问题速查表问题现象可能原因排查方向注入不触发装饰器顺序错误检查injectable位置注入触发但无效果异常被上层吞掉加[INJECTED]标记系统直接崩溃注入概率过高降到 0.1 重试结果无法比对基线本身有问题先跑通基线再注入脏数据检测不到过滤逻辑太宽松提高脏数据比例5.5 关于 LLM as judge 的使用心得做输出层注入测试的时候你需要一个裁判来判断模型输出是否合理。我用过几种方案规则匹配简单关键词匹配快但覆盖不全LLM as judge让另一个模型打分准但慢且贵Embedding 相似度跟基线答案比相似度折中方案我最后选的是 Embedding 相似度 规则匹配的组合。先用相似度筛出可疑的再用规则确认。这样既快又准成本也可控。注意judge 模型不要跟被测模型用同一个否则可能一起犯同样的错误。我一般用不同厂商的模型做交叉验证。6. 从故障注入到持续混沌跑通一次注入测试只是开始。真正有价值的是把它变成持续的过程。我现在每个 LLM 项目的 CI 里都会跑一套注入测试覆盖五个注入点每个注入点至少三个用例。跑不过就不让合并。这套东西上线之后线上因为 Provider 问题导致的故障下降了大概七成。另外一个小技巧把注入测试的结果存下来做成趋势图。如果某个注入点的失败率突然上升说明最近的改动可能引入了新的脆弱点。这个比等用户投诉再排查要主动得多。最后分享一个我个人的体会做 LLM 混沌工程最难的不是技术是心态。你要接受“系统一定会出问题”这个前提然后主动去制造问题。很多团队不愿意做这件事觉得是在给自己找麻烦。但等你真的被线上幻觉坑过一次就会明白提前下毒比事后救火便宜太多了。
