企业官网建设流程全解析

AI 服务链路的全链路压测——从流量录制到瓶颈定位的完整工程实践

一、背景与问题

AI 推理服务的性能评估与传统的 Web 服务压测有本质区别。一个 AI 应用链路通常涉及多个组件：API 网关、Prompt 模板引擎、RAG 检索、模型推理引擎、后处理与格式化。任何一个环节成为瓶颈，都会导致端到端延迟放大。更麻烦的是，推理服务的延时分布呈现明显的重尾特征——P50 延迟 2 秒，P99 延迟可能高达 15 秒。

我们在上线"客服意图识别 + 知识库问答"链路时，首次全链路压测发现端到端 P99 延迟是预期的 3.2 倍。通过流量录制、逐环节拆解与 bottleneck 定位，最终将 P99 从 18.6 秒优化到 5.8 秒。本文将完整记录这一过程。

二、方案设计

全链路压测的核心思路是：先录制线上真实流量（包括请求参数与上下文），再在隔离环境回放，通过分布式追踪逐环节分析耗时分布，最终定位瓶颈点。

flowchart TD subgraph "第一阶段: 流量录制" A1["线上网关流量采样"] --> A2["GoReplay / tcpdump 抓包"] A2 --> A3["流量脱敏处理<br/>（手机号/身份证/会话信息）"] A3 --> A4["存储为压测数据集"] end subgraph "第二阶段: 压测执行" B1["JMeter / wrk2 发起压测"] --> B2["压测环境<br/>（独立 K8s 集群）"] B2 --> B3["全链路追踪<br/>（OpenTelemetry）"] B3 --> B4["指标采集<br/>（Prometheus + Grafana）"] end subgraph "第三阶段: 瓶颈分析" C1["端到端延迟分布"] --> C2["Spans 耗时拆解"] C2 --> C3["CPU / GPU 利用率"] C3 --> C4["连接池 / 线程池状态"] C4 --> C5["定位瓶颈环节"] end A4 --> B1 B2 --> C1

关键工具链

三、实战演示

3.1 流量录制与脱敏

# 使用 GoReplay 从线上网关抓取流量 # -input-raw: 监听网卡端口 # -output-file: 输出到文件用于离线回放 gor --input-raw :8080 \ --http-allow-url "/api/v1/chat" \ --http-allow-method POST \ --output-file "requests_%Y%m%d_%H%M.gor" \ --output-file-max-size-limit "1GB" # 脱敏处理：替换手机号、身份证、Token 等敏感字段 python scripts/desensitize.py \ --input requests_20260715_1400.gor \ --output requests_20260715_1400_clean.gor \ --rules pii_replace.yaml

""" 流量脱敏脚本：移除请求中的个人身份信息（PII）。 支持正则匹配、字段白名单两种模式。 """ import re import json import sys from typing import Dict, List class TrafficDesensitizer: """HTTP 流量脱敏处理器""" # PII 匹配规则 PHONE_PATTERN = re.compile(r'1[3-9]\d{9}') ID_CARD_PATTERN = re.compile(r'\d{17}[\dXx]') def __init__(self, rules_file: str): """加载脱敏规则配置""" with open(rules_file, 'r', encoding='utf-8') as f: self.rules = json.load(f) def desensitize(self, line: str) -> str: """ 对单行请求数据进行脱敏。 Args: line: 原始请求行（可能的格式：Gor 协议或 JSON） Returns: 脱敏后的请求行 """ try: # 替换手机号 line = self.PHONE_PATTERN.sub('138****0000', line) # 替换身份证号 line = self.ID_CARD_PATTERN.sub('110101****1234', line) # 如果请求体是 JSON，还需要按字段白名单脱敏 # 此处简化处理，实际需解析 Gor 协议格式 return line except Exception as e: print(f"脱敏处理异常: {e}", file=sys.stderr) return line def main(): """主入口""" import argparse parser = argparse.ArgumentParser(description='流量脱敏工具') parser.add_argument('--input', required=True, help='输入文件路径') parser.add_argument('--output', required=True, help='输出文件路径') parser.add_argument('--rules', default='pii_replace.yaml', help='脱敏规则文件') args = parser.parse_args() desensitizer = TrafficDesensitizer(args.rules) with open(args.input, 'r', encoding='utf-8') as fin, \ open(args.output, 'w', encoding='utf-8') as fout: for line in fin: clean_line = desensitizer.desensitize(line) fout.write(clean_line) print(f"脱敏完成: {args.input} -> {args.output}") if __name__ == '__main__': main()

3.2 分布式追踪 Span 集成

/** * AI 推理链路的 OpenTelemetry Span 集成。 * * 每个推理阶段创建独立的 Span，便于分析各环节耗时。 * 环节：参数校验 -> Prompt 构建 -> Embedding 检索 -> LLM 推理 -> 后处理 */ @Component @Slf4j public class AIChatService { @Autowired private Tracer tracer; @Autowired private PromptBuilder promptBuilder; @Autowired private VectorSearchService vectorSearch; @Autowired private LLMInferenceClient llmClient; /** * 带全链路追踪的 AI 对话处理。 * * @param request 用户对话请求 * @return AI 生成回复 * @throws AIChatException 如果推理失败 */ public ChatResponse chat(ChatRequest request) { // 创建根 Span Span rootSpan = tracer.spanBuilder("ai.chat.full") .setAttribute("user.id", request.getUserId()) .setAttribute("session.id", request.getSessionId()) .startSpan(); try (Scope scope = rootSpan.makeCurrent()) { // 阶段1：Prompt 构建 Span promptSpan = createChildSpan("ai.chat.prompt_build"); try (Scope s1 = promptSpan.makeCurrent()) { String prompt = promptBuilder.build(request); promptSpan.setAttribute("prompt.length", prompt.length()); promptSpan.setAttribute("message.count", request.getMessages().size()); } finally { promptSpan.end(); } // 阶段2：RAG 检索 Span retrievalSpan = createChildSpan("ai.chat.rag_retrieval"); List<String> contextDocs = List.of(); try (Scope s2 = retrievalSpan.makeCurrent()) { contextDocs = vectorSearch.search(request.getQuery(), 5); retrievalSpan.setAttribute("retrieved.docs", contextDocs.size()); } catch (Exception e) { retrievalSpan.recordException(e); retrievalSpan.setStatus(StatusCode.ERROR); throw new AIChatException("RAG检索失败: " + e.getMessage(), e); } finally { retrievalSpan.end(); } // 阶段3：LLM 推理 Span inferenceSpan = createChildSpan("ai.chat.llm_inference"); try (Scope s3 = inferenceSpan.makeCurrent()) { String response = llmClient.generate( request.getQuery(), contextDocs); inferenceSpan.setAttribute("output.length", response.length()); rootSpan.setStatus(StatusCode.OK); return new ChatResponse(response); } catch (Exception e) { inferenceSpan.recordException(e); inferenceSpan.setStatus(StatusCode.ERROR); throw new AIChatException("LLM推理失败: " + e.getMessage(), e); } finally { inferenceSpan.end(); } } finally { rootSpan.end(); } } private Span createChildSpan(String name) { return tracer.spanBuilder(name) .setParent(Context.current()) // 使用当前 Span 上下文作为父级 .startSpan(); } }

3.3 瓶颈定位——Jaeger 火焰图分析

# 从 Jaeger 导出指定时间段的 Trace 数据 curl "http://jaeger-query:16686/api/traces?\ service=ai-chat-service&\ operation=ai.chat.full&\ limit=100&\ lookback=1h&\ minDuration=5s" \ | jq '.data[].spans[] | {operationName, duration}' \ | sort -t: -k2 -n -r \ | head -20 # 输出示例（瓶颈定位依据）： # "ai.chat.llm_inference": 4200ms ← 推理耗时 4.2s，是主要瓶颈 # "ai.chat.prompt_build": 45ms # "ai.chat.rag_retrieval": 380ms ← 向量检索 380ms，可优化 # "ai.chat.post_process": 15ms

四、深度解析

4.1 瓶颈定位的典型模式

在我们的压测过程中，识别出三类典型瓶颈：

4.2 SSE 流式响应的压测难点

AI 对话通常采用 SSE（Server-Sent Events）流式返回。传统 JMeter 的 HTTP 采样器无法正确处理 SSE。我们基于 JMeter 的 JSR223 Sampler 实现了 SSE 客户端：

• 首 Token 时间（TTFT）：从发送请求到收到第一个 SSE 事件的延迟。
• Token 生成速率（TPS）：单位时间内生成的 Token 数。
• 流中断率：连接在完整响应前被意外关闭的比例。

4.3 压测环境与生产环境的差异校准

即使使用真实流量回放，压测环境与生产环境仍存在差异：硬件可能不一致、网络延迟不同、下游服务可能是 Mock。校准方法：选择 5% 的真实请求在压测环境回放，对比 Trace 数据中相同环节的延迟比例，得出校准系数后应用到整体分析中。

五、压测结果的可靠性分析

5.1 压测数据的代表性陷阱

压测数据的质量直接决定优化方向的准确性。在我们的实践中，发现录制的线上流量存在明显的"幸存者偏差"：录制的流量都是成功请求，不包含因超时、限流而被拒绝的请求，这些被拒绝的请求往往暴露了系统的真实瓶颈。

因此，在流量录制阶段，建议不仅录制成功请求，还通过镜像流量（Traffic Mirroring）的方式，将生产环境的真实负载（包括超时、失败）引入压测环境。此外，业务场景会随时间演变（如促销活动、新功能上线），压测数据集需要定期更新，避免使用 3 个月前的老数据评估当前系统。

一个实用的做法是：每次大版本发布前，从生产环境录制最近 7 天的流量作为压测数据集，并在压测报告中注明数据集的时间范围和业务场景分布。

5.2 压测环境与生产环境的隔离策略

全链路压测的核心原则是"压测环境与生产环境完全隔离"。但在 K8s 多租户集群中，这种隔离往往难以完美实现：如果压测环境和生产环境共享同一个 GPU 节点池，压测产生的高 GPU 利用率会影响生产服务的推理延迟。

我们的解决方案是：

1. 独立节点池：为压测环境分配独立的节点池（Node Pool），并通过 K8s 的节点亲和性（Node Affinity）确保压测 Pod 只调度到压测节点池；
2. 下游依赖隔离：在压测环境的服务配置中，将所有下游依赖（如向量数据库、Redis）指向压测环境的实例，避免压测流量意外访问生产环境的数据库；
3. 网络隔离：通过 K8s NetworkPolicy 限制压测环境 Pod 只能访问压测环境的 Service，防止配置错误导致的跨环境访问。

5.3 AI 推理质量的自动化验证

AI 推理服务的压测与传统 Web 服务有一个关键区别：推理质量难以通过自动化断言验证。传统压测可以通过 HTTP 状态码、响应体 JSON 结构来验证正确性，但 AI 推理的响应是自由文本，可能因模型版本升级、Prompt 调整而变化。

我们的做法是：在压测脚本中加入"采样验证"逻辑——每处理 100 个请求，随机抽取 1 个响应，由人工或自动化评估模型（如 BLEU、ROUGE）验证推理质量。如果推理质量下降超过阈值（如 BLEU 分数下降 >5%），立即中断压测并告警。

这能避免"压测性能达标但推理质量崩溃"的隐蔽问题。在生产环境中，我们曾遇到过一次模型升级后，压测 P99 延迟从 5s 降至 3s（性能提升），但因为新模型的 Temperature 参数设置过高，推理结果的随机性大幅增加，导致用户投诉率上升 300%。如果压测中包含了推理质量验证，这种问题可以被提前发现。

六、总结

AI 服务全链路压测的三个关键实践：

1. 流量录制要脱敏：PII 信息绝不能进入压测环境。脱敏脚本必须在录制阶段完成，不在存储后再处理。
2. Span 粒度要合理：按推理管线阶段切分 Span，而不是笼统的一个 "chat" Span。只有拆解得足够细，才能定位到具体瓶颈。
3. 瓶颈优化要有取舍：LLM 推理的 P99 延迟有物理下限（GPU 计算时间），如果 P99 已接近单次推理的理论下限，优化重心应转向请求调度（见第 1 篇文章的连续批处理策略）。

全程压测是手段，不是目的。最终交付的是"在给定 QPS 下，P99 延迟不超过 SLA 阈值"的能力验证，而不是一堆漂亮的 Grafana 截图。

作者：程序员鸭梨（李然），Java 架构师，专注 AI 推理服务性能工程与全链路可观测性实践。欢迎留言交流你的压测经验。