企业官网建设流程全解析

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 层是最容易注入也最容易检测的。我实测下来，只要在 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 整体架构设计思路

我的设计原则是：注入层要能透明地插到现有代码里，不改业务逻辑。理想情况下，业务代码完全不知道自己在被测试。

架构分四块：

1. 注入点注册中心：用装饰器标记哪些函数可以被注入
2. 故障策略库：定义各种故障模式，比如超时、返回空、返回脏数据、截断
3. 注入控制器：决定什么时候注入、注入哪个、注入多久
4. 观测与断言层：记录注入后的系统行为，判断是否触发了预期降级

这套东西的核心是装饰器 + 策略模式。业务代码只需要在关键函数上加一个@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 故障策略库的设计

故障策略我抽象成一个基类，每种故障实现自己的execute：

from 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(f"Unknown scenario: {scenario}") point_name, fault = scenario_map[scenario] entry = _INJECTION_REGISTRY.get(point_name) if not entry: raise ValueError(f"Injection 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", probability=0.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(f"Q: {q}\nA: {answer}\n")

基线跑完，确认系统在正常情况下能给出正确答案。这一步很重要，因为如果基线本身就有问题，注入测试的结果没法解读。

4.3 注入空检索故障

现在打开空检索注入：

controller = InjectionController() controller.enable("empty_retrieval", probability=1.0) # 100% 注入，确保触发 injected_results = [] for q in questions: answer = run_qa_pipeline(q) injected_results.append({"question": q, "answer": answer}) print(f"Q: {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_k=5) 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 常见问题速查表

5.5 关于 LLM as judge 的使用心得

做输出层注入测试的时候，你需要一个裁判来判断模型输出是否合理。我用过几种方案：

• 规则匹配：简单关键词匹配，快但覆盖不全
• LLM as judge：让另一个模型打分，准但慢且贵
• Embedding 相似度：跟基线答案比相似度，折中方案

我最后选的是 Embedding 相似度 + 规则匹配的组合。先用相似度筛出可疑的，再用规则确认。这样既快又准，成本也可控。

注意：judge 模型不要跟被测模型用同一个，否则可能一起犯同样的错误。我一般用不同厂商的模型做交叉验证。

6. 从故障注入到持续混沌

跑通一次注入测试只是开始。真正有价值的是把它变成持续的过程。

我现在每个 LLM 项目的 CI 里都会跑一套注入测试，覆盖五个注入点，每个注入点至少三个用例。跑不过就不让合并。这套东西上线之后，线上因为 Provider 问题导致的故障下降了大概七成。

另外一个小技巧：把注入测试的结果存下来，做成趋势图。如果某个注入点的失败率突然上升，说明最近的改动可能引入了新的脆弱点。这个比等用户投诉再排查要主动得多。

最后分享一个我个人的体会：做 LLM 混沌工程，最难的不是技术，是心态。你要接受“系统一定会出问题”这个前提，然后主动去制造问题。很多团队不愿意做这件事，觉得是在给自己找麻烦。但等你真的被线上幻觉坑过一次，就会明白，提前下毒比事后救火便宜太多了。