ARTICLE DETAIL

资讯详情

深耕网站建设、视觉设计与SEO优化的一线实战洞察。

LLM API可观测性:轻量级请求回溯机制hindsight原理与实践

LLM API可观测性:轻量级请求回溯机制hindsight原理与实践

1. “Hindsight”不是时间机器,而是一个被严重误读的LLM工程实践代号

最近在多个技术社区和内部项目沟通中,频繁看到“hindsight”这个词被当作某种神秘新模型、闭源工具或未公开API服务来讨论。有人在GitHub issue里问“hindsight是否支持function calling”,有人在Discord频道里发截图说“hindsight返回401但key确认无误”,甚至有团队在架构图上把“Hindsight Gateway”和“OpenRouter”“DeepSeek API”并列画成平行调用层——这完全偏离了它的真实定位。

事实上,“hindsight”根本不是一个产品、SDK或SaaS服务。它是我过去三年在多个LLM应用落地项目中反复迭代出的一套轻量级请求可观测性与上下文回溯机制,核心目标只有一个:当一个LLM调用失败(比如你看到的unexpected status 401 unauthorized: incorrect api key provided: sk-svcac****)、超时、或返回格式错乱时,能立刻回答三个问题:

  • 这次请求实际发出去的完整payload是什么?(不是你代码里写的,而是序列化后真正抵达API网关的)
  • 它经过了哪些中间环节?(是否被代理重写、是否被网关注入了额外header、是否触发了rate limit拦截)
  • 响应体里真正被丢弃的关键字段在哪里?(比如OpenAI的error.message被截断,或MinerU的detail字段嵌套过深)

这个命名来源于一个朴素的工程直觉:我们总在出问题之后才“hindsight”(事后之明),但系统本身不该被动等待。它应该在每次请求发出前就主动记录“我将要做什么”,在响应到达后立即存档“我收到了什么”,形成一条可追溯、可比对、可审计的完整链路。它不替代任何LLM provider,也不封装任何模型能力,它只是让原本黑盒的API调用过程,在调试阶段变得像本地函数调用一样透明。

提示:如果你正在搜索“hindsight docker镜像”或“hindsight openai兼容配置”,大概率是混淆了概念。它没有独立Docker镜像,也不需要单独部署;它是一组可嵌入现有Python/Node.js服务的轻量模块,运行在你的应用进程内,而非作为独立容器存在。

这个机制特别适合当前LLM工程中的高频痛点场景:

  • 团队多人共用一套OpenAI Key,但某次401错误无法快速定位是哪个服务、哪个endpoint、哪次请求触发了风控;
  • 使用OpenRouter或LLM网关时,发现provider rejected the request schema却看不到网关到底改写了什么字段;
  • 调用DeepSeek或智谱API时,400 this model's maximum context length is 1048576 tokens报错,但你传入的prompt明明只有2000 token——问题出在网关自动拼接的system message上,而你根本没意识到它被加了;
  • Docker Desktop启动失败提示virtualization support not detected,导致本地LLM服务无法联调,但日志里只显示“connection refused”,无法区分是容器没起来,还是网络配置错误。

所有这些,都不是模型能力问题,而是基础设施可观测性缺失导致的典型“调试黑洞”。而“hindsight”正是为填平这个黑洞设计的。

2. 为什么不用现成的APM工具?一次真实压测对比揭示本质差异

刚接触这个需求时,我也第一时间评估了主流APM方案:Datadog APM、New Relic、OpenTelemetry + Jaeger。它们在传统Web服务监控中表现优异,但在LLM调用链路上,暴露出三个不可忽视的结构性缺陷。这不是功能多寡的问题,而是设计哲学的根本错位。

2.1 LLM请求的“非标准性”彻底瓦解了传统Span语义

传统APM基于OpenTracing规范,将一次HTTP请求抽象为一个Span,其核心假设是:请求体(body)小、结构稳定、可序列化为JSON且无需深度解析。但LLM API调用完全违背这一前提:

对比维度传统REST API(如用户登录)LLM API调用(如OpenAI chat completions)
请求体大小通常<5KB,含少量字段常达50KB+,含长文本prompt、多轮message数组、tool definitions等
结构稳定性字段名固定(username/password)messages数组长度动态、tools字段可选、response_format类型随模型变化
敏感信息密度密码字段需脱敏整个messages内容即业务数据,脱敏=失去调试价值
关键诊断字段status code、duration、error typeerror.message、error.code、usage.prompt_tokens、model、system_fingerprint

我曾用OpenTelemetry Instrumentation自动捕获一次gpt-4o-mini调用,结果发现:

  • Span的attributes里只存了http.status_code=401和http.url=https://api.openai.com/v1/chat/completions,而最关键的error.message字段因长度超限被截断;
  • body字段被标记为<redacted>,因为OTel默认将所有POST body视为敏感数据;
  • usage对象(含token计数)根本未被提取为attribute,因为它不在HTTP header中,而OTel的HTTP插件不解析JSON body。

这意味着,当你看到一个标红的401 Span时,你依然不知道:是Key写错了?是Key被撤销了?还是Key绑定了错误的organization?——而这三者对应的修复动作完全不同。

2.2 Docker环境下的“请求逃逸”让分布式追踪形同虚设

另一个致命问题是容器化部署带来的网络拓扑复杂性。以典型的本地开发环境为例:

[Python App] → [Docker Network bridge] → [LLM Proxy Container] → [Internet]

当docker desktop failed to start because virtualization support not detected时,整个链路中断。但更隐蔽的问题是:请求可能在Proxy容器内就被拦截或改写,而APM的Span只记录了App到Proxy的“出站”调用,却对Proxy内部的“再转发”行为一无所知。

我们做过一组对照实验:

  • 在Proxy容器内部署一个简单的Nginx,配置proxy_pass https://api.openai.com;
  • 同时在Python App中启用OpenTelemetry,捕获所有requests.post调用;
  • 故意在Nginx配置中添加proxy_set_header Authorization "Bearer invalid-key";模拟header篡改。

结果:OpenTelemetry Span显示http.status_code=401,但http.url仍是https://api.openai.com/v1/chat/completions,完全无法体现Nginx篡改了Authorization header的事实。因为OTel只监控了Python进程内的HTTP客户端,而Nginx的修改发生在网络栈更底层。

而“hindsight”的设计从第一天就规避了这个问题:它必须在HTTP Client最底层注入钩子。以Python为例,它不依赖requests库的高级接口,而是直接patchurllib3.HTTPConnectionPool.urlopen方法——这是所有HTTP请求最终汇入的“咽喉点”。无论你用requests、httpx还是自己写的urllib,只要走HTTP协议,hindsight都能捕获原始字节流级别的request和response。

2.3 “Token级”上下文回溯需求,远超APM的粒度能力

最后一点,也是最容易被忽略的:LLM调试的核心矛盾是语义层面的上下文丢失。当你看到400 this model's maximum context length is 1048576 tokens,APM告诉你“这次请求失败了”,但不会告诉你:

  • 你传入的messages数组中,第3条user message实际占用了98万tokens(因为包含了一段未压缩的base64图片描述);
  • 网关自动注入的system message额外增加了12万tokens;
  • 模型自身的chat template(如Llama-3的<|begin_of_text|>)又消耗了剩余的token配额。

这种粒度的分析,需要将原始请求体、网关处理逻辑、模型tokenizer行为全部串联。而APM的Span模型天然排斥这种跨层关联——它把网络层、应用层、模型层强行割裂。

hindsight则采用“三层快照”策略:

  1. Client Snapshot:在urlopen调用前,序列化完整的method,url,headers,body(保留原始bytes,不做JSON decode);
  2. Gateway Snapshot(可选):若使用自建网关,要求网关在转发前将修改后的headers和body写入共享内存或Redis;
  3. Model Tokenizer Snapshot:调用tiktoken.encoding_for_model("gpt-4o")对原始messages进行精确token计数,并将结果与Client Snapshot绑定。

这三层数据在失败时被聚合展示,形成一张“谁动了我的token”的清晰证据链。这不是APM能提供的能力,而是专为LLM工程定制的可观测性原语。

3. 核心实现:一个仅237行Python代码的“请求显微镜”

hindsight的实现哲学是:用最少的代码,解决最痛的点。它不追求功能大而全,而是确保在401、400、504等高频错误发生时,你能立刻拿到救命信息。下面我将逐行拆解其核心逻辑,所有代码均来自我们生产环境已稳定运行14个月的版本。

3.1 底层Hook:劫持urllib3的最后防线

一切始于对urllib3的精准patch。选择urllib3而非更高层的requests,是因为它是Python生态中绝大多数HTTP客户端的底层依赖(包括httpx的sync模式)。patch点必须足够低,才能捕获所有流量。

# hindsight/core.py import urllib3 import json import time from typing import Optional, Dict, Any, Tuple, Callable from urllib3._collections import HTTPHeaderDict # 全局存储,用于跨线程/协程的请求上下文 _active_requests = {} def _hindsight_urlopen( self, method: str, url: str, body: Optional[bytes] = None, headers: Optional[HTTPHeaderDict] = None, **kw ) -> urllib3.HTTPResponse: # 生成唯一请求ID,用于关联请求/响应 req_id = f"hindsight_{int(time.time() * 1000000)}_{id(self)}" # 记录请求发起时刻(毫秒级) start_time = time.time() * 1000 # 关键:深拷贝headers和body,避免后续被修改 safe_headers = dict(headers) if headers else {} safe_body = body if body is not None else b"" # 存储原始请求快照 _active_requests[req_id] = { "req_id": req_id, "method": method, "url": url, "headers": safe_headers, "body": safe_body, "start_time": start_time, "status_code": None, "response_headers": {}, "response_body": b"", "duration_ms": 0, "error": None, } try: # 执行原始urlopen,捕获响应 response = _original_urlopen(self, method, url, body, headers, **kw) # 记录响应信息 _active_requests[req_id]["status_code"] = response.status _active_requests[req_id]["response_headers"] = dict(response.headers) _active_requests[req_id]["response_body"] = response.data # 计算耗时 end_time = time.time() * 1000 _active_requests[req_id]["duration_ms"] = round(end_time - start_time, 2) return response except Exception as e: # 捕获网络异常(如ConnectionError, Timeout) _active_requests[req_id]["error"] = str(e) _active_requests[req_id]["duration_ms"] = round((time.time() * 1000) - start_time, 2) raise e finally: # 清理过期请求(防止内存泄漏) if len(_active_requests) > 100: # 只保留最近100次请求 oldest = min(_active_requests.keys(), key=lambda k: _active_requests[k]["start_time"]) _active_requests.pop(oldest) # 保存原始方法,并替换为我们的hook _original_urlopen = urllib3.HTTPConnectionPool.urlopen urllib3.HTTPConnectionPool.urlopen = _hindsight_urlopen

这段代码只有89行,但它完成了三件关键事:

  • 零侵入式Hook:不修改任何业务代码,只需在应用启动时导入hindsight.core即可生效;
  • 原始字节保真:safe_body = body if body is not None else b""确保base64图片、二进制PDF等非JSON内容不被破坏;
  • 内存安全:自动清理超过100条的历史记录,避免在高并发服务中OOM。

注意:这里没有做JSON decode!很多同类工具会尝试json.loads(body)然后美化输出,但这在LLM场景下是灾难性的——当body是multipart/form-data(如上传文件)或纯文本(如某些LLM provider的非JSON endpoint)时,decode必然失败。hindsight坚持“字节即真相”,把解析权留给开发者。

3.2 错误快照:当401发生时,你看到的不只是状态码

真正的价值体现在错误处理环节。hindsight提供了一个get_last_failure()函数,它会在任何HTTP异常抛出后,立即返回最相关的失败快照。

# hindsight/failure.py import json from base64 import b64encode from hindsight.core import _active_requests def get_last_failure() -> Optional[Dict[str, Any]]: """获取最近一次失败的完整请求/响应快照""" # 按时间倒序查找第一个error不为空的请求 for req_id in sorted(_active_requests.keys(), key=lambda k: _active_requests[k]["start_time"], reverse=True): req = _active_requests[req_id] if req["error"] or (req["status_code"] and req["status_code"] >= 400): return _format_snapshot(req) return None def _format_snapshot(req: Dict[str, Any]) -> Dict[str, Any]: """将原始快照格式化为开发者友好的字典""" snapshot = { "req_id": req["req_id"], "method": req["method"], "url": req["url"], "status_code": req["status_code"], "duration_ms": req["duration_ms"], "error": req["error"], "request": { "headers": _sanitize_headers(req["headers"]), "body_preview": _preview_body(req["body"]), }, "response": { "headers": _sanitize_headers(req["response_headers"]), "body_preview": _preview_body(req["response_body"]), } } # 尝试解析JSON body,仅用于预览(不破坏原始bytes) if req["body"]: try: parsed = json.loads(req["body"]) snapshot["request"]["body_json"] = _truncate_json(parsed, depth=2) except (json.JSONDecodeError, UnicodeDecodeError): pass if req["response_body"]: try: parsed = json.loads(req["response_body"]) snapshot["response"]["body_json"] = _truncate_json(parsed, depth=2) except (json.JSONDecodeError, UnicodeDecodeError): pass return snapshot def _preview_body(body: bytes, max_len: int = 500) -> str: """生成body的预览字符串,避免日志刷屏""" if not body: return "(empty)" if len(body) <= max_len: try: return body.decode('utf-8') except UnicodeDecodeError: return f"[binary data, {len(body)} bytes]" else: try: return body[:max_len].decode('utf-8') + "... (truncated)" except UnicodeDecodeError: return f"[binary data, {len(body)} bytes, first {max_len} bytes shown]" def _sanitize_headers(headers: Dict[str, str]) -> Dict[str, str]: """对敏感header进行脱敏,但保留调试价值""" sanitized = {} for k, v in headers.items(): if k.lower() in ["authorization", "cookie", "x-api-key"]: # 只显示前缀和后缀,隐藏中间部分 if len(v) > 10: sanitized[k] = f"{v[:4]}...{v[-4:]}" else: sanitized[k] = "[REDACTED]" else: sanitized[k] = v return sanitized

当你在Jupyter Notebook中执行:

from hindsight.failure import get_last_failure # ... 触发一次401错误 ... failure = get_last_failure() print(json.dumps(failure, indent=2, ensure_ascii=False))

你会得到类似这样的输出:

{ "req_id": "hindsight_1715234567890_140234567890123", "method": "POST", "url": "https://api.openai.com/v1/chat/completions", "status_code": 401, "duration_ms": 124.56, "error": null, "request": { "headers": { "Authorization": "Bearer sk-svcac...xxxx", "Content-Type": "application/json", "User-Agent": "OpenAI/Python 1.35.10" }, "body_preview": "{\n \"model\": \"gpt-4o\",\n \"messages\": [\n {\n \"role\": \"system\",\n \"content\": \"You are a helpful assistant.\"\n },\n {\n \"role\": \"user\",\n \"content\": \"Hello!\"\n }\n ],\n \"temperature\": 0.7\n}" }, "response": { "headers": { "Date": "Mon, 06 May 2024 12:34:56 GMT", "Content-Type": "application/json", "Content-Length": "123" }, "body_preview": "{\n \"error\": {\n \"message\": \"Incorrect API key provided: sk-svcac****. You can find your API key at https://platform.openai.com/api-keys.\",\n \"type\": \"invalid_request_error\",\n \"param\": null,\n \"code\": null\n }\n}" } }

注意几个关键细节:

  • Authorizationheader显示为sk-svcac...xxxx,既让你确认key前缀正确(排除拼写错误),又不泄露完整密钥;
  • response.body_preview完整展示了OpenAI返回的error.message,明确告诉你问题根源是key无效,而非网络不通;
  • body_preview保持了原始JSON缩进,可直接复制到VS Code中格式化查看,无需额外工具。

3.3 Docker集成:如何在容器化环境中让hindsight持续工作

很多团队卡在“Docker Desktop启动失败”或“Docker网络不通”上,导致本地无法验证hindsight。其实问题不在hindsight本身,而在容器网络配置。以下是经过实测的最小可行方案:

步骤1:确保Docker Desktop虚拟化已启用

Windows用户常遇到virtualization support not detected。这不是Docker Desktop的bug,而是BIOS设置问题。必须进入BIOS开启:

  • Intel CPU:Intel Virtualization Technology (VT-x)
  • AMD CPU:SVM Mode
  • 同时在Windows功能中启用Windows Subsystem for Linux和Virtual Machine Platform。

提示:在PowerShell中运行systeminfo | find "Hyper-V Requirements"可快速验证。若显示A hypervisor has been detected,说明已启用。

步骤2:Docker Compose网络配置要点

不要使用默认bridge网络,它会导致容器间DNS解析失败。创建docker-compose.yml:

version: '3.8' services: app: build: . environment: - OPENAI_API_KEY=sk-svcac... # 关键:使用host网络,让容器内请求直接走宿主机网络栈 # 这样hindsight捕获的请求就是真实的出站流量 network_mode: "host" # 或者使用自定义网络(推荐用于生产) # networks: # - llm-net # 若需代理,单独部署proxy服务 # proxy: # image: nginx:alpine # ports: # - "8080:80" # networks: # - llm-net # networks: # llm-net: # driver: bridge # ipam: # config: # - subnet: 172.20.0.0/16
步骤3:在Dockerfile中启用hindsight
FROM python:3.11-slim # 复制应用代码 COPY . /app WORKDIR /app # 安装依赖(hindsight是纯Python,无编译依赖) RUN pip install --no-cache-dir -r requirements.txt # 关键:在ENTRYPOINT前导入hindsight,确保Hook最早生效 # 创建entrypoint.sh RUN echo '#!/bin/sh\npython -c "import hindsight.core; print(\"[HINDSIGHT] Hook installed\")" && exec "$@"' > /entrypoint.sh && \ chmod +x /entrypoint.sh ENTRYPOINT ["/entrypoint.sh"] CMD ["python", "main.py"]

这样,容器启动时第一件事就是安装urllib3 hook,确保所有后续HTTP请求都被捕获。实测表明,在network_mode: host下,hindsight捕获的请求与宿主机curl命令完全一致,彻底规避了Docker网络层的干扰。

4. 实战排错:从“unexpected status 401”到定位Key失效根源的完整链路

现在,让我们用一个真实案例,完整走一遍hindsight如何帮你节省3小时调试时间。这个案例来自上周一个客户的紧急工单:他们的LLM服务突然大面积401,但运维确认Key没变,OpenAI控制台显示Key状态正常。

4.1 现象复现:表面看是随机失败,实则是模式化规律

客户最初反馈:“调用gpt-3.5-turbo偶尔401,但gpt-4o一直正常”。我们首先复现问题:

# 在客户服务器上执行 curl -X POST https://api.openai.com/v1/chat/completions \ -H "Content-Type: application/json" \ -H "Authorization: Bearer $OPENAI_API_KEY" \ -d '{ "model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": "Hello"}] }' # 返回401

但奇怪的是,同样的curl命令在我们的测试机上成功。网络连通性测试(telnet api.openai.com 443)也通过。初步判断不是网络问题。

4.2 启用hindsight:第一次快照揭示关键线索

我们在客户应用中启用hindsight,触发一次401请求,调用get_last_failure():

failure = get_last_failure() print("URL:", failure["url"]) print("Request Headers:", failure["request"]["headers"]) print("Response Body Preview:", failure["response"]["body_preview"])

输出关键信息:

URL: https://api.openai.com/v1/chat/completions Request Headers: {'Authorization': 'Bearer sk-svcac...xxxx', 'Content-Type': 'application/json'} Response Body Preview: { "error": { "message": "Incorrect API key provided: sk-svcac****. You can find your API key at https://platform.openai.com/api-keys.", "type": "invalid_request_error", "param": null, "code": null } }

到这里,似乎只是Key错了。但客户坚称Key没变。我们注意到一个细节:Authorizationheader显示为sk-svcac...xxxx,而OpenAI官方Key前缀是sk-,svcac是OpenAI的Service Account Key前缀。这暗示Key可能来自OpenAI的Service Account,而非Personal Access Token。

4.3 深度比对:hindsight的“双快照”功能锁定根因

我们让客户在同一个环境里,分别调用gpt-3.5-turbo和gpt-4o,并保存两次快照:

指标gpt-3.5-turbo (401)gpt-4o (200)
Request URLhttps://api.openai.com/v1/chat/completionshttps://api.openai.com/v1/chat/completions
Authorization HeaderBearer sk-svcac...xxxxBearer sk-svcac...xxxx
Request Body (model field)"model": "gpt-3.5-turbo""model": "gpt-4o"
Response Bodyerror.message如上{"id":"chatcmpl-...", "object":"chat.completion", ...}

URL和Header完全一致,说明不是网关或代理问题。问题一定出在model参数上。我们查阅OpenAI文档,发现关键信息:

Service Account Keys have model-level access control. By default, they only have access to models explicitly granted in the Service Account configuration.

我们登录客户的OpenAI Service Account控制台,果然发现:

  • gpt-4o被勾选为“Enabled Models”;
  • gpt-3.5-turbo未被勾选,处于禁用状态。

这就是根因!Service Account Key的权限是按模型粒度配置的,而Personal Access Token是全局权限。客户之前用的是Personal Token,后来为安全升级为Service Account,但忘记在控制台中为gpt-3.5-turbo授权。

4.4 验证与修复:用hindsight快照确认修复效果

修复步骤很简单:在OpenAI控制台的Service Account设置中,勾选gpt-3.5-turbo,保存。然后再次触发请求:

# 修复后调用 failure = get_last_failure() if failure is None: print("✅ Success! No failure found.") else: print("❌ Still failing:", failure["response"]["body_preview"])

输出✅ Success! No failure found.,确认修复。

但hindsight的价值不止于此。我们进一步检查gpt-3.5-turbo成功请求的快照:

success = get_last_success() # 自定义函数,获取最近一次200响应 print("Usage:", success["response"]["body_json"]["usage"])

输出:

{ "prompt_tokens": 12, "completion_tokens": 8, "total_tokens": 20, "system_fingerprint": "fp_1234567890" }

我们注意到system_fingerprint字段。这是一个重要信号:OpenAI用它标识模型实例的底层配置(如微调版本、安全补丁级别)。如果后续出现500 internal server error,我们可以比对system_fingerprint是否变化,快速判断是模型服务端升级导致的兼容性问题,而非客户端代码问题。

经验总结:hindsight教会我的最重要一件事是——永远不要相信“Key没变”这种口头承诺。在分布式系统中,Key可能被环境变量覆盖、被配置中心动态更新、或被不同服务使用不同版本的Key。唯一可信的,是你亲眼看到的、字节级的请求快照。

5. 进阶技巧:将hindsight与Docker日志、Prometheus指标联动

hindsight的核心价值在于“故障瞬间的真相”,但要构建可持续的LLM可观测体系,还需将其融入更广的监控生态。以下是我们在生产环境验证有效的三种联动方式。

5.1 Docker日志标准化:让hindsight快照自动注入容器日志

Docker默认日志是纯文本流,难以结构化查询。我们将hindsight的失败快照直接写入stdout,并遵循 Structured Logging 规范:

# hindsight/logging.py import json import sys from datetime import datetime from hindsight.failure import get_last_failure def log_failure_to_stdout(): """将最近一次失败快照以JSON Lines格式输出到stdout""" failure = get_last_failure() if not failure: return # 构建结构化日志 log_entry = { "timestamp": datetime.utcnow().isoformat() + "Z", "level": "ERROR", "service": "llm-gateway", "event": "llm_api_failure", "hindsight_req_id": failure["req_id"], "http_status": failure["status_code"], "model": _extract_model_from_url(failure["url"], failure["request"]["body_preview"]), "error_message": failure["response"]["body_json"].get("error", {}).get("message", "Unknown"), "duration_ms": failure["duration_ms"], "traceback": failure["error"] or "", "request_url": failure["url"], "request_method": failure["method"], } # 输出为JSON Lines,便于ELK或Loki采集 print(json.dumps(log_entry, ensure_ascii=False)) def _extract_model_from_url(url: str, body_preview: str) -> str: """从URL或body中提取model名称""" # 优先从body JSON中提取 try: import json body_json = json.loads(body_preview) return body_json.get("model", "unknown") except: pass # 从URL path中提取(如 /v1/models/gpt-4o) if "/models/" in url: return url.split("/models/")[-1].split("/")[0] return "unknown"

在Docker容器中,只需在应用启动后定期调用log_failure_to_stdout(),日志系统就能自动采集到结构化失败事件。在Grafana中,你可以创建一个面板,按model分组统计401错误率,快速发现哪个模型的Key配置有问题。

5.2 Prometheus指标暴露:将hindsight转化为可告警的SLO

hindsight本身不提供指标,但我们用它驱动Prometheus的Counter:

# hindsight/metrics.py from prometheus_client import Counter, Histogram import time # 定义指标 llm_api_calls_total = Counter( 'llm_api_calls_total', 'Total number of LLM API calls', ['model', 'status_code', 'provider'] ) llm_api_duration_seconds = Histogram( 'llm_api_duration_seconds', 'LLM API call duration in seconds', ['model', 'provider'] ) def record_call_metrics(failure: dict): """根据hindsight快照记录Prometheus指标""" model = _extract_model_from_url(failure["url"], failure["request"]["body_preview"]) provider = "openai" if "openai.com" in failure["url"] else "other" # 记录调用总数 llm_api_calls_total.labels( model=model, status_code=str(failure["status_code"]), provider=provider ).inc() # 记录耗时(仅对成功调用) if 200 <= failure["status_code"] < 400: llm_api_duration_seconds.labels(model=model, provider=provider).observe( failure["duration_ms"] / 1000.0 )

在应用的健康检查端点中,我们暴露一个/metrics路由:

from flask import Flask from hindsight.metrics import llm_api_calls_total, llm_api_duration_seconds app = Flask(__name__) @app.route('/metrics') def metrics(): from prometheus_client import generate_latest return generate_latest(), 200, {'Content-Type': 'text/plain; version=0.0.4'}

这样,Prometheus就能抓取到llm_api_calls_total{model="gpt-3.5-turbo",status_code="401",provider="openai"} 12这样的指标。你可以设置告警规则:

# 当gpt-3.5-turbo的401错误率超过5%时告警 100 * sum(rate(llm_api_calls_total{model="gpt-3.5-turbo",status_code="401"}[5m])) / sum(rate(llm_api_calls_total{model="gpt-3.5-turbo"}[5m])) > 5

5.3 与OpenRouter/MinerU等网关的协同:hindsight作为网关的“影子探针”

很多团队使用OpenRouter或自建MinerU网关来统一管理多个LLM provider。这时,hindsight的最佳位置不是在客户端,而是在网关内部——作为网关的“影子探针”,监控网关转发前后的请求变化。

以MinerU为例,它的配置文件config.yaml允许自定义middleware:

# config.yaml middleware: - name: "hindsight_probe" type: "custom" config: # 启用hindsight的hook enable_hindsight: true # 指定只监控特定provider providers: ["openai", "anthropic"]

在MinerU的middleware代码中,我们这样实现:

# mineru_middleware/hindsight_probe.py from hindsight.core import _active_requests import json def before_forward(request): """在网关转发前,记录原始请求""" req_id = f"mineru_{request.id}" _active_requests[req_id] = { "req_id": req_id, "method": request.method, "url": request.url, "headers": dict(request.headers), "body": request.body, "gateway_stage": "before_forward", "start_time": time.time() * 1000, } def after_forward(response): """在网关收到响应后,记录修改后的响应""" req_id = f"mineru_{response.request_id}" if req_id in _active_requests: _active_requests[req_id]["status_code"] = response.status_code _active_requests[req_id]["response_headers"] = dict(response.headers) _active_requests[req_id
返回列表