用mitmproxy反向代理破解LLM API联调与token观测难题

平时做大模型应用的同学应该都有这种体会：线上某个基于LLM的功能突然变慢，问了一圈只能得到“模型推理慢”这种模糊结论，再想往深了查，拿不出任何证据；联调阶段更是心累，明明prompt在网页上调试得好好的，换成代码调SDK就各种不对劲，又看不到真实请求里到底发了什么。我之前也一直被这个问题卡住，直到把 mitmproxy 以反向代理模式架在客户端和大模型API之间，把请求体、响应流、token消耗、时延全部变成可查的数据，才算是找到了一个顺手又足够透明的排查工具。这篇就把整个做法拆开讲清楚，包括启动姿势、抓包数据怎么读、以及怎么写一个自动归档的插件。

用名词解释一下这个方案的核心：mitmproxy 是一个支持脚本扩展的中间人代理工具，反向代理模式等于让它在本地监听一个端口，替客户端转发所有流量到远端的大模型API，这个过程中每个请求和响应都会被完整记录和解析。对于排查真实请求内容、统计token消耗、验证prompt是否被正确拼装，这套方案比单纯看日志靠谱得多。适合正在做LLM应用开发、AI平台稳定性保障、或者需要做数据合规审计的团队参考。

1. 为什么用Mitmproxy反向代理，而不是别的抓包姿势

1.1 三种抓包方案的选型对比

先说结论：抓大模型请求这件事，行业里主流的路子其实是三种，但我最终选了反向代理，不是因为别的路不行，而是因为这条路最省事、最贴合LLM API的调用方式。

第一种是正向代理。让客户端通过系统代理走mitmproxy，mitmproxy以普通代理身份接收CONNECT请求，再做中间人解密。这种方式适合浏览器、手机App这类能配置系统代理的应用，但对大模型SDK不一定友好。很多SDK在底层直接用了连接池和自定义TLS配置，对系统代理的支持参差不齐，而且代理一旦生效，会把整个机器上的所有HTTPS流量都揽进来，想看大模型请求得先在一片噪音里过滤，效率很低。

第二种是透明代理。把网关或者路由器的流量重定向到代理端口，客户端完全感知不到代理的存在。听起来很优雅，但落地成本高，要改iptables或者交换机策略，还得确保内网DNS解析正确，一不小心就把整个环境搞挂。对于只是要调试某个服务的开发或运维同学来说，这明显是过度设计。

第三种就是反向代理，也是我这篇的核心。客户端把目标地址直接改到本机mitmproxy监听的端口，mitmproxy再把流量转发给真正的上游API。对大模型SDK来说，只需要改一个base_url参数，其他什么都不用动，代理对调用方几乎是透明的。

三轮对比下来，反向代理在“改造成本”和“定位精准度”之间取得了最好的平衡。你只需要把SDK里的API地址从https://api.gateway.example.com换成http://127.0.0.1:8080，剩下的活儿全部交给mitmproxy。

1.2 反向代理模式下流量到底怎么走

画不出架构图，我就用文字把完整链路捋一遍。以某大模型API网关为例，正常情况下客户端的请求路径是这样的：

LLM应用 -> https://api.gateway.example.com/v1/chat/completions -> 大模型服务

接入mitmproxy反向代理之后，链路变成：

LLM应用 -> http://127.0.0.1:8080 -> mitmproxy本地端口 -> https://api.gateway.example.com/v1/chat/completions -> 大模型服务

注意两个关键点。第一，客户端和mitmproxy之间走的通常是HTTP明文，因为地址是本机回环，没有跨网络传输，不需要加密；第二，mitmproxy到上游大模型API之间是真实的TLS连接，证书校验照常进行。所以反向代理模式下根本不需要给客户端安装mitmproxy的CA证书，只有在客户端用https://访问本地代理端口时，才需要做证书信任，而这种场景在我们日常联调中基本用不上。

另外要解释一个容易混淆的概念：抓包解密到底解的是哪一段。普通正向代理抓HTTPS包，解的是“客户端到代理”这一段，代理用自己生成的CA证书冒充目标站点，骗过客户端的TLS校验，从而看到明文。反向代理模式解得也是这一段，但因为客户端访问的是本地HTTP，默认就是明文，压根不需要骗。代理到上游那一段，即使你是代理自己也看不到加密内容，只会把加密后的TLS字节流转发出去。这个细节理解清楚，后面遇到证书报错就知道该往哪个方向排查。

1.3 它适合做什么，不适合做什么

先说适合的场景。第一，本地联调验证prompt拼装逻辑，能清晰看到system、user、assistant消息到底按什么顺序发给模型。第二，测试环境观测线上问题复现，把用户的原始请求原样转发给测试账号，对比响应差异。第三，数据合规审计，记录每一次请求的完整内容，确认有没有敏感信息被意外发送。第四，token消耗审计，统计每个业务线调用了多少次、平均生成了多少token，用来对账。

但也有明确的不适合场景。如果某个核心业务对响应延迟极度敏感，或者流量已经到每秒上千次，不建议把mitmproxy直接架在生产链路上。因为代理本身是单进程事件模型，一旦在这个环节做body缓冲和落盘，会明显推高尾部延迟，尤其是流式响应场景，客户端会觉得首字出来变慢了一大截。这类需求更适合从网关侧采集日志，或者做旁路镜像，而不是用这种“中途插入”的代理方案。

2. 环境准备与启动配置

2.1 安装与版本确认

安装非常简单，mitmproxy 是一个Python包，直接通过包管理器装就行：

pip install mitmproxy

装好之后命令行里会出现三个可执行文件：mitmproxy是带交互界面的版本，mitmdump是不带界面的轻量版本，mitmweb是带Web界面的版本。我日常用mitmweb居多，因为浏览器界面看请求列表和body比终端更方便，还能直接搜索、过滤、导出。

版本方面提醒一下，mitmproxy的更新频率不算慢，不同版本的命令参数有过调整。比如旧版本里指定代理模式用的还是--reverse这种写法，新版本统一收敛成--mode reverse:...这种单个参数。如果你看老的教程遇到参数不识别，先执行mitmproxy --help确认一下当前版本的参数写法，不要照搬。

2.2 启动反向代理模式的正确姿势

核心命令只有一行：

mitmproxy --mode reverse:https://api.gateway.example.com --listen-port 8080

拆开解释每个参数的含义。--mode reverse:https://api.gateway.example.com表示以反向代理模式启动，并把所有接收到的请求转发到指定的上游地址。--listen-port 8080表示mitmproxy监听在本机的8080端口，客户端后续就访问这个地址。默认监听地址是127.0.0.1，也就是只有本机能连。

如果要多台机器共享这个代理，比如让其他同事的联调环境也指向你这边，需要加一个监听地址参数：

mitmproxy --mode reverse:https://api.gateway.example.com --listen-host 0.0.0.0 --listen-port 8080

这样同一内网的其他机器就能把base_url配置成http://你的开发机IP:8080来共享抓包能力。不过内网使用时要稍微注意一下，这个端口上没有认证机制，任何人知道地址就能用你的代理转发流量，所以在公司内网用完记得关掉，别长时间开着裸奔。

如果还想把原始流量存下来做离线分析，可以加一个保存参数：

mitmproxy --mode reverse:https://api.gateway.example.com --listen-port 8080 --save-stream-file llm_capture.flow

.flow文件是mitmproxy自己的二进制格式，可以用mitmdump -r llm_capture.flow离线回放里面的内容，非常实用。

2.3 配置客户端指向本机，并处理SSL的坑

大模型SDK一般都会提供一个设置base_url的入口。比如Python环境里常见的初始化方式是这样：

from llm_client import LLMClient client = LLMClient( base_url="http://127.0.0.1:8080", api_key="sk-test-key-only-for-local-debug", )

这里有个容易踩坑的细节：即使你只是本地测试，SDK依然会把Authorization请求头发出去。如果你的客户端代码里硬编码了一个线上环境的API Key，那这个Key就会随着流量经过mitmproxy，并且很可能被记录到日志里。所以我强烈建议本地调试用一个仅供测试的虚拟Key，或者让mitmproxy在转发时直接把Authorization改写掉。注意不要用生产Key做这种试验，一旦日志泄露，代价非常大。

如果上游是一个自建的、启用了双向TLS的网关，反向代理模式默认情况下会失败，因为mitmproxy发起的TLS连接没有携带客户端证书。这种场景需要写一个addon在TLS握手时注入证书，或者干脆换用SDK直连加日志的方式，不在代理层面硬磕。大部分云上大模型API都是单向TLS验证，不需要考虑这个。

3. 拆解一次大模型请求的完整“档案”

3.1 请求头里有哪些值得记录的信息

用一次真实的debug记录举例。在mitmweb的请求列表里点开一次POST /v1/chat/completions请求，会看到一堆请求头，我平时重点看这几个：

很多大模型平台会在响应头里回一个X-Request-Id或者类似字段，这个一定要记下来。线上用户报问题的时候，你拿着这个ID去平台侧查日志，直接就能定位到对应的推理日志，比让用户复述“大概什么时间点了什么按钮”高效得多。所以从我的习惯来看，每次请求是否带Request-ID，是判断一个SDK配置是否合理的重要标准。

3.2 请求体里prompt与参数的设计细节

请求体是整个抓包档案里信息量最大的地方。拿一个典型的对话请求举例：

{ "model": "llm-model-x", "messages": [ { "role": "system", "content": "你是一名电商平台的售后客服，只回答与订单、物流、退换货相关的问题。" }, { "role": "user", "content": "我的订单为什么还没有发货？已经过去三天了。" } ], "temperature": 0.7, "max_tokens": 1024, "stream": true }

这里几个字段要重点解释。model字段决定了模型的版本和计费标准，排查线上问题时，先确认请求是不是打到了预期的模型上，很多“线上和线下结果不一致”的问题，本质就是请求的模型不同。messages数组是整个prompt的核心，注意它的顺序性：system消息放在最前面，用来设定总体人格和约束；user消息是用户输入；assistant消息则是之前模型生成的内容。如果某个应用是多轮对话，每一次轮次都会把历史消息全量带上，你从抓包里能看到上下文有没有被正确截断。

temperature这个参数容易被人忽略，但它在抓包里很有价值。你发现某个请求的回复内容异常固定或异常发散时，先看一眼请求体里temperature到底传的什么值。max_tokens决定了单次生成的最大长度，如果你发现响应在某个位置突然中断，而且输出字数刚好接近max_tokens，那基本就是触顶了。

再补充一个稍微进阶一点的信息：很多新版本API把函数调用或者工具调用也放进请求体里，格式通常是一个tools数组。如果这个数组在每次请求里都被塞入大量工具定义，请求体的体积会迅速膨胀。抓包最大的好处就是能直观地看到body大小，我见过某个应用因为塞了太多工具定义，单次请求直接超过几百KB，带宽和延迟双双爆炸，这种问题不看真实请求是根本察觉不到的。

3.3 流式响应与usage统计

当请求体里stream为true时，大模型API返回的不是一个完整的JSON，而是一连串SSE格式的数据块，每个块以data:开头。抓包里的实际内容大概是这样的：

data: {"id":"chatcmpl-123","object":"chat.completion.chunk","model":"llm-model-x","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]} data: {"id":"chatcmpl-123","object":"chat.completion.chunk","model":"llm-model-x","choices":[{"index":0,"delta":{"content":"您的"},"finish_reason":null}]} data: {"id":"chatcmpl-123","object":"chat.completion.chunk","model":"llm-model-x","choices":[{"index":0,"delta":{"content":"订单"},"finish_reason":null}]} data: {"id":"chatcmpl-123","object":"chat.completion.chunk","model":"llm-model-x","choices":[{"index":0,"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":86,"completion_tokens":153,"total_tokens":239}} data: [DONE]

这个抓包片段能透露很多信息。第一，整个响应是分多次到达的，你可以在mitmweb里看到每次数据传输的时间间隔，从而判断模型的“首字延迟”和“平均生成速度”。第二，最后一块数据里带了usage字段，里面是token消耗明细。这个字段非常有用，等于平台直接告诉你这次请求花了多少token。如果拿到的响应里没有usage，可以用一个土办法估算：按中文字符数除以1.5左右、英文按字符数除以3.5左右粗略估算，但精度远不如直接解析usage。

另外，流式响应下还有一个观察点：delta.content字段是增量内容，不是完整回复。把每个chunk里的content字段按顺序拼接起来，才能得到完整的模型回复。所以做抓包分析时，不能只看一个chunk就下结论，得把整个流全部拼完才拿到完整的生成结果。

4. 写一个自动归档Flow数据的Addon

4.1 Mitmproxy插件运行机制

如果只是临时抓一次包，直接看mitmweb界面就够了。但要持续观测、批量归集、定期分析的时候，人眼盯界面显然不现实。mitmproxy提供了一个很强的扩展机制——addon，本质上就是一个Python类，通过定义特定名字的方法来响应代理事件。

这里要先讲清楚一个关键的事件时序，否则写出来的插件会踩坑。以一次HTTP请求为例，addon里会依次触发这些方法：request（请求头发起时）、responseheaders（响应头返回时）、response（响应完整接收后）。但对于大模型流式响应，默认情况下response事件触发时，响应body并不会完整加载到内存里，而是边接收边转发，你在response里调用get_text()拿到的往往是不完整的。为了拿到完整的流式文本，需要在responseheaders阶段把该响应标记为非流式缓冲。

4.2 关键代码实现（记录、脱敏、落库）

下面是一个可以直接跑的插件示例，功能包括：识别大模型请求、脱敏敏感信息、解析usage字段、把关键数据写入本地SQLite。

import json import re import sqlite3 import time from mitmproxy import http, ctx SENSITIVE_PATTERNS = [ (re.compile(r"1[3-9]\d{9}"), "[手机号]"), (re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+"), "[邮箱]"), (re.compile(r"Bearer\s+[\w.-]+"), "Bearer [已脱敏]"), ] def mask_sensitive(text: str) -> str: for pattern, replacement in SENSITIVE_PATTERNS: text = pattern.sub(replacement, text) return text class LLMLogger: def __init__(self): self.conn = sqlite3.connect("llm_flows.db") self.conn.execute( """ CREATE TABLE IF NOT EXISTS llm_flows ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts REAL, path TEXT, model TEXT, prompt_preview TEXT, response_preview TEXT, status_code INTEGER, latency_ms INTEGER, prompt_tokens INTEGER, completion_tokens INTEGER, total_tokens INTEGER ) """ ) self.conn.commit() def request(self, flow: http.HTTPFlow) -> None: flow.metadata["llm_start_ts"] = time.time() flow.metadata["llm_request_body"] = flow.request.get_text() def responseheaders(self, flow: http.HTTPFlow) -> None: content_type = flow.response.headers.get("content-type", "") if "text/event-stream" in content_type: flow.response.stream = False def response(self, flow: http.HTTPFlow) -> None: if "/chat/completions" not in flow.request.pretty_url: return req_body = flow.metadata.get("llm_request_body") or "{}" resp_body = flow.response.get_text() or "" try: req_json = json.loads(req_body) model = req_json.get("model", "unknown") except json.JSONDecodeError: model = "unknown" usage = {} usage_match = re.search(r'"usage":\s*(\{[^{}]*\})', resp_body) if usage_match: try: usage = json.loads(usage_match.group(1)) except json.JSONDecodeError: usage = {} latency_ms = int((time.time() - flow.metadata.get("llm_start_ts", time.time())) * 1000) prompt_preview = mask_sensitive(req_body[:200]) response_preview = mask_sensitive(resp_body[:200]) self.conn.execute( """ INSERT INTO llm_flows ( ts, path, model, prompt_preview, response_preview, status_code, latency_ms, prompt_tokens, completion_tokens, total_tokens ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( time.time(), flow.request.path, model, prompt_preview, response_preview, flow.response.status_code, latency_ms, usage.get("prompt_tokens", 0), usage.get("completion_tokens", 0), usage.get("total_tokens", 0), ), ) self.conn.commit() ctx.log.info(f"recorded llm request: model={model}, latency={latency_ms}ms") addons = [LLMLogger()]

这段代码里几个细节值得展开说明。

第一个是脱敏逻辑。mask_sensitive函数在写入数据库之前对请求体和响应体做了清洗，把手机号替换成[手机号]、邮箱替换成[邮箱]、Bearer后面的Key替换成[已脱敏]。这一步一定要做，因为大模型请求体里经常携带真实的用户信息，如果日志库没有严格的权限管控，这就是一次隐私事故。

第二个是关闭流式响应。responseheaders里看到text/event-stream就设置flow.response.stream = False，这样mitmproxy会把整个SSE流缓存下来，后续response事件里就能拿到完整文本。但要注意，这个设置会增加下游客户端的等待时间，因为它会等所有chunk都到齐之后再一次性转发，客户端看到的不是“打字机效果”。所以生产链路上不要开这个，只有做分析时才开启。

第三个是usage解析。我的处理方式简单粗暴，用正则从SSE全文里搜索"usage": {...}字段，因为流式场景下usage只会在最后一个chunk里出现一次，用正则匹配不会漏。拿到usage之后就能计算出每次调用的准确token成本。

启动方式是在启动命令后面挂上这个脚本：

mitmproxy --mode reverse:https://api.gateway.example.com --listen-port 8080 -s llm_logger.py

启动后每次大模型请求落地，本地就会多一条llm_flows.db记录。

4.3 数据落库之后怎么分析

数据进了SQLite之后，分析就变成写SQL的事。我平时用得最多的是这几种查询。

按模型维度统计一段时间的调用情况：

SELECT model, COUNT(*) AS call_count, ROUND(AVG(latency_ms), 1) AS avg_latency_ms, SUM(total_tokens) AS total_tokens, SUM(prompt_tokens) AS sum_prompt_tokens, SUM(completion_tokens) AS sum_completion_tokens FROM llm_flows GROUP BY model ORDER BY call_count DESC;

这条SQL可以直接回答几个运营侧关心的问题：哪个模型被调用最多、平均延迟多少、token消耗的大头在prompt还是completion。如果发现prompt_tokens占比畸高，说明上下文中塞了太多历史内容或者太长指令，有优化空间。

按小时维度看调用趋势：

SELECT strftime('%Y-%m-%d %H:00', ts, 'unixepoch') AS hour_bucket, COUNT(*) AS call_count, AVG(latency_ms) AS avg_latency_ms FROM llm_flows GROUP BY hour_bucket ORDER BY hour_bucket;

这种时间序列查询很适合观测业务是否存在明显的峰谷，也能在排查故障时快速确认某段时间是否有异常流量。

如果你不想自己写SQL，直接用sqlite3命令行也可以交互查询：

sqlite3 llm_flows.db "SELECT model, COUNT(*), AVG(latency_ms) FROM llm_flows GROUP BY model;"

5. 常见问题与排查技巧实录

5.1 典型问题速查表

我把实际操作中最常碰到的问题整理成一个表，遇到类似现象可以直接对照处理。

5.2 值得单独说说的两个坑

第一个坑是SDK自动重试造成的“幽灵请求”。某大模型SDK默认在网络超时或返回5xx时会自动重试，而且每次重试都是完整的新请求。如果你只盯着mitmweb看，会看到同一时刻冒出好几个一模一样的POST请求，这时候第一反应往往是“是不是代理重复转发了”，其实并不是代理的问题，而是SDK自己在重试。排查重点是看这些请求的时间戳是否相差一个固定的超时时间，如果是，基本可以确定是SDK重试机制。这时候去SDK的配置里关掉重试，或者把超时时间调大，就能避免重复请求对上游造成的压力。

第二个坑是流式缓冲对TTFB的影响。我前面提到为了看到完整响应体需要关闭流式转发，这会让客户端感受到的“首字返回时间”大幅增加。有一次我在测试环境开着这个插件，结果业务方反馈说首字延迟从500ms变成5秒，我一度以为是上游出问题了，最后才发现是我插件里关了stream导致整个响应体被缓冲后才一次性转发给客户端。所以这里要有一个明确的心智模型：分析模式下的缓冲是为了拿数据，用户体验自然会受损，做完分析一定要记得把插件摘掉。

5.3 个人使用习惯

最后分享一个我自己的使用习惯作为收尾。我平时不会长时间挂着这个反向代理，而是采用“按需开启、用完即关”的方式。需要排查问题或者做例行审计的时候，启动代理开个十来分钟，把关键请求记录下来，然后立刻关掉。生成的llm_flows.db文件我一般会定期归档到独立存储，保留时间按团队合规要求来，不需要无限期保存。还有一个很顺手的小技巧：如果把.flow文件通过mitmdump -r llm_capture.flow -n离线回放，就能在不开服务的情况下反复查看当时抓到的流量，这在跨团队核对问题时特别好用，直接把文件发给对方，对方本地就能查看，不需要部署任何环境。说到底，抓包工具的价值不在于看得多全，而在于让你在真正需要证据的时候，拿得出结构化、可追溯的事实。