单测全绿，上生产就翻车，说的就是LLM数据管线。

做了一段时间LLM应用后端，我最深的体会是：代码逻辑层面的bug真不难查，真正的坑全在“数据流动”上。尤其是在RAG、批量抽取、异步处理这类管线型工程里，单元测试跑得再漂亮，也只是证明了“单条数据在理想环境下能走通”，根本证明不了“一批真实数据在跑批、重试、并发时不会互相污染”。

这篇博文就记录两个我实际踩过的运行时问题。它们有一个共同特征：用单元测试都没法覆盖，因为它们不在“函数逻辑”里，而在“任务与任务之间、批次与批次之间”的隐式契约里。一个叫上下文污染，一个叫并发共享状态泄漏。如果你也在搭LLM数据管线，这两条经验应该能帮你少走不少弯路。

1. LLM数据管线的测试盲区为什么存在

1.1 单元测试的三大假设，LLM管线一条都不满足

单元测试的基本逻辑是：给定输入，断言输出。它隐含了三个假设——隔离性（每个用例独立运行）、确定性（相同输入必得相同输出）、无副作用（测试不依赖外部系统状态）。

可LLM数据管线是什么样的？它的核心处理对象不是纯函数，而是“带状态的文本流”。一条文本进来，可能要经过切分、向量化、缓存查询、模型调用、结果解析、入库，然后它的输出又会成为下一条文本检索时的上下文。这里每一步都不是孤立的：模型调用有吞吐限制和随机性，缓存有命中和淘汰，入库有重试和幂等性问题。更不用说，生产环境几乎都是并发跑批的，一批几十上百个任务同时推进，任务之间共享着大语言模型客户端、内存缓存、数据库连接池这些公共资源。

单元测试在这套体系里能覆盖什么？它只能覆盖“单次调用的函数逻辑”——比如prompt模板拼得对不对、JSON解析器能不能处理某个边界返回。至于“A任务的结果会不会污染B任务的检索上下文”“重试之后会不会写入重复数据”“并发场景下两个任务会不会拿到彼此的缓存结果”，这些全都是跨函数、跨任务、跨实例的运行时行为，单测根本无从下手。

1.2 LLM管线特有的三类“静默型缺陷”

我给这类问题起了个名叫“静默型缺陷”——它不报错，不红屏，但结果就是错的。单元测试之所以覆盖不了，是因为它设计出来就是查“崩溃型缺陷”的，查不了这种数据悄悄变质的问题。

• 状态泄漏：上游任务把信息写进了一个共享位置，下游任务读到了不该读的东西。
• 时序竞态：并发任务并行执行时，两个任务交错访问了同一个可变对象，最终结果取决于执行顺序。
• 非确定性放大：模型输出本身就有随机性，管线如果再用有状态的聚合逻辑，随机性就会被放大成系统性错误。

这三种缺陷有一个共同特点：只有在“多个任务同时在线”时才会显现。你写单测时，通常只跑一个任务、只喂一条数据，等于永远在“单车道空驶”状态下验证系统，当然测不出“多车并线时互相别车”的问题。

1.3 单测全绿带来的虚幻自信

我不反对单元测试，它是质量下限的兜底。问题在于很多人把“单测通过”当成“系统没问题”的充分条件，这在LLM工程里尤其危险。LLM管线里的真实复杂性——token截断怎么处理、缓存命中的内容是哪一批的、重试补偿是否幂等——这些恰恰是单元测试的结构性盲区。

换句话说，单测验证的是“计划中的路径”，而生产事故几乎都发生在“计划外的交错”——重试的交错、批次颗粒度的交错、共享缓存的交错。这些盲区本质上不是代码缺陷，而是数据契约缺陷，单测这个工具在设计上就不擅长发现它们。

下面进入正题，把两个典型案例拆开讲。

2. 缺陷实录一：上下文污染——父请求把“记忆”串给了子请求

2.1 事故现场：文档摘要里混入了无关内容

第一个项目是做一个RAG知识库的批量预处理管线。原始文档进来后，系统会先调用LLM生成一段摘要，再把摘要和正文一起向量化入库。这段逻辑看起来非常简单，核心函数大概是：

def generate_summary(doc_text: str, model_client) -> str: prompt = SUMMARY_TEMPLATE.format(doc_text=doc_text) response = model_client.chat(messages=[{"role": "user", "content": prompt}]) return response["content"]

当时每个单元测试都通过了：喂一段样本文档，断言返回的是一个字符串，长度大于某阈值。一切都很正常。

结果一上生产，跑完一批100篇文档后，人工抽检发现了一个诡异现象：第37篇文档的摘要里，混入了一段第12篇文档中的原句。更蹊跷的是，第12篇文档本身并没有出现在第37篇的正文里，向量检索也不可能跨文档召回。这显然不是信息检索问题，而是摘要生成过程的输入被污染了。

2.2 根因定位：单次调用没问题，批量调用就出事

排查过程从最简单的地方开始：单独重跑第37篇文档，摘要完全正常；单独重跑第12篇文档，也正常。但只要按照生产顺序连续跑，第37篇就会抽风。这个“单独正常、连着跑就出错”的现象，是典型的跨任务状态残留。

顺着这个思路继续查，发现模型调用的封装层为了省token，加了一个“对话记忆”功能。它维护着一个全局的消息数组，每次调用前把历史消息塞进去当上下文：

# 简化后的污染代码 _conversation_history = [] def call_llm_with_memory(user_content: str): _conversation_history.append({"role": "user", "content": user_content}) response = model_client.chat(messages=_conversation_history) _conversation_history.append({"role": "assistant", "content": response["content"]}) return response["content"]

单测为什么会漏？因为在测试环境里，每个test case跑完就完事了，_conversation_history虽然在累积，但断言只看最后返回的字符串，不关心历史里有什么。而生产环境是连续跑几十上百篇文档，前一篇的user内容、assistant输出全都被当作上下文传给了下一次调用。第37篇被污染，本质上是因为第12篇处理完后，它的原文还躺在全局列表里，被第37篇的调用当成了“历史对话”一起发给了模型。

2.3 修复方案：让上下文生命周期显式化

修复说穿了不值钱：把全局变量改成局部变量，上下文要么彻底不用，要么严格按“一次任务一个实例”来创建。

def generate_summary(doc_text: str, model_client) -> str: # 不共享任何状态，每次调用都是独立的 return call_llm_single_turn(SUMMARY_TEMPLATE.format(doc_text=doc_text), model_client)

如果某个场景确实需要多轮对话记忆，那也绝不能用一个模块级list，而是要包在一个ConversationSession对象里，由上层任务决定何时创建、何时销毁。比如：

session = ConversationSession(model_client) for chunk in chunks: session.add_user_message(chunk) reply = session.complete() # 处理一个完整任务后，session自动释放，不会串到下一个任务

这里想强调一个经验：凡是出现“模块级可变对象”，就要立刻警觉它是否跨任务共享。单元测试天然串行执行，往往掩盖了这种共享状态的问题。要么改成无状态的写法，要么让状态的边界跟着业务任务走，而不是跟着模块走。全局可变状态是LLM管线的头号污染源，没有之一。

3. 缺陷实录二：并发共享状态泄漏——缓存把不同会话的数据串了

3.1 事故现场：结构化抽取结果张冠李戴

第二个项目是一条异步流水线，负责从用户上传的对话记录里抽取结构化信息，比如客户意向、商品偏好、沟通评分。每条记录进来后会先做一次缓存查询——如果这条记录的处理结果在缓存里，就直接返回缓存结果，省一次LLM调用。这个设计本身没问题，问题出在缓存的key上。

代码大概长这样：

@lru_cache(maxsize=512) def extract_structured_info(session_id: int, text: str) -> dict: response = llm_client.chat(messages=[...]) return parse_json(response["content"])

我把session_id和text作为缓存key了，表面上看起来很合理：不同会话、不同文本，缓存不会串。可生产环境是异步并发跑批，服务实例不止一个。有一次线上监控发现，A用户的上传记录里，抽取出的“偏好标签”竟然是B用户的内容。查了下日志，两个用户的文本内容非常相似——很多人上传的客服对话都是类似“咨询退货流程”这种话术。

问题出得更隐蔽：lru_cache是进程内的，但LLM服务实例在并发时会走共享的Redis缓存层。项目组在接入Redis缓存时，key只取了文本的哈希值，完全没拼session_id：

# 伪代码示意 cache_key = f"llm:extract:{hash(text)}"

于是，只要两个不同会话的文本内容一样，它们的LLM抽取结果就被认为是同一个东西，后到的请求直接命中了先到请求的缓存。而单测环境里，所有用例串行执行，缓存命中与否完全可控，你根本不会看到“两个并发任务同时写同一个key”的场景。

3.2 排查过程：从脏数据反推状态污染

这个问题的排查比上下文污染麻烦得多，因为它是随机性的：不是每次必现，只是间歇性出现，而且只出现在特定文本重复时。

我的排查思路是这样的：

第一步，对比出错记录的时区。把A用户和B用户的抽取结果并排摆在一起，发现A用户记录里写入的标签，在时间上正好对应于B用户记录的处理时刻。这说明不是“入库后被动串改”，而是“处理阶段就拿到了对方的结果”。

第二步，看缓存层。把Redis里的数据导出来，发现不同session_id对应的key完全一致，也就是说这个key根本没有区分业务实体。到了这一步，根因已经清楚了：缓存key设计缺少业务上下文维度。

第三步，模拟并发测试。我用线程池压了100条相同文本、不同session_id的请求，结果轻松复现。单测阶段之所以没发现，是因为——我根本没写过并发场景的测试用例。所有缓存相关单测都是串行的：先写缓存、再读缓存、断言一致，完全没有覆盖“多个任务同时读同一key”的竞态。

3.3 修复方案：缓存key带上业务上下文，配合幂等重试

修复方案有两层。第一层治标：缓存key强制包含全部业务标识，绝不只对文本做哈希。

# 修复后的缓存key设计 cache_key = f"llm:extract:{session_id}:{hash(text)}"

其实session_id在这里也不能完全避免串数据。更稳妥的做法是，给每条上游记录生成一个全局唯一的request_id，整个管线的所有中间产物都挂在这个request_id下面：

cache_key = f"llm:extract:{request_id}"

第二层治本：LLM管线的缓存和重试必须幂等。我们的流水线用了消息队列，消费端处理失败后会重试。原来的重试逻辑只判断“这条记录有没有处理过”，没判断“处理结果是否完整写入”。结果就是：第一次调用超时了但LLM其实已经产生了结果，重试时缓存不命中，LLM再次调用，返回了另一个结果，两条结果都写入了数据库。这个问题和缓存污染叠加在一起，数据想不脏都难。

修复后的逻辑是：每次处理一个消息前，先检查这个request_id是否已有“完成态”的结果记录；如果有，直接跳过，不管结果是何时产生的。这样的“状态机重试”才能保证幂等，普通的重试只会放大并发污染。

4. 从缺陷到防御：LLM数据管线运行时问题排查手册

4.1 排查思路四步走

经历了这两个事故之后，我给自己定了一套排查规范。遇到“单测通过但数据不对”的问题，按下面四个方向查，通常半小时内能定位。

• 查共享状态：先扫一遍代码里有没有模块级可变对象、全局缓存、类级别的可变属性。LLM管线的污染绝大多数发生在这里。
• 查缓存key维度：把缓存key打印出来，看它是否包含了完整业务上下文。如果发现只对文本内容做哈希，那基本已经锁定了问题。
• 查并发交错场景：写一个小的并发烟雾测试——几十个线程同时跑同一个任务，故意制造竞态。这个测试虽然不常跑，但每次改管线代码都应该跑一遍。
• 查重试幂等性：模拟“第一次调用超时但实际已成功”的场景，看系统会不会重复写入或产生双份结果。

4.2 常见问题速查表

4.3 预防性设计三条底线

踩过这两个坑之后，我给自己定了几条铁律，写进团队规范里了，每条都是拿事故换来的。

第一，LLM数据管线必须无状态化。任何函数，只要输入是确定的文本，输出就不该依赖函数外部任何可变的东西。需要上下文就显式传参，需要缓存就显式传对象，绝不使用全局变量和模块级缓存。

第二，缓存key必须包含完整业务实体标识。文本哈希只能作为辅助维度，不能作为唯一维度。只要业务上存在“两条记录文本相同但实体不同”的可能，key就必须把实体ID放进去。

第三，集成测试必须模拟多任务并发。单测负责逻辑正确性，集成测试负责管线契约正确性。我后来在管线里加了一个“并发冒烟测试”，每次部署前自动跑：同时处理10条语义相近的文本，断言每个输出只包含自己输入里的内容。这个测试的成本很低，但价值极高——它能把前两个事故都扑灭在发布之前。

写在最后

说实话，这两个问题修复之后，我一度产生了“单测没用”的错觉。后来想通了：单测不是没用，而是它的作用边界非常明确——它验证代码符合预期，但验证不了数据在流转过程中会不会被环境改变。LLM管线的特殊之处在于，它的每一步处理都引入了不确定性，而工程上的所有约定——缓存、重试、批量——又都在放大这种不确定性的副作用。所以，单测仍然是第一道门，但绝不能是最后一道门。真正的防线，是在管线层面加一层“数据血缘断言”，让每个中间产物都能追溯到它唯一的输入来源。这才是LLM数据管线工程里最值得投入的部分。