企业官网建设流程全解析

1. 大模型推理可观测性到底在解决什么问题

大模型应用上线之后，最让人头疼的往往不是模型本身答得好不好，而是“钱花在哪了、慢在哪了、什么时候开始不对劲了”。我见过不少团队，模型接进业务跑了两周，账单突然翻了三倍，排查半天才发现是某个上游服务在疯狂重试，每次重试都带着完整的上下文重新推理一遍。也见过线上对话突然变慢，用户投诉不断，最后定位到是某个批次的请求上下文长度暴涨，把显存打满导致排队。这些问题的共同点是：如果没有一套围绕推理过程的观测体系，你根本不知道发生了什么。

所谓大模型日志与可观测性，说白了就是给每一次推理调用建立一份“体检报告”。这份报告里最核心的两个指标，一个是Token 消耗，一个是延迟。Token 消耗直接对应成本，延迟直接对应体验。把这两个指标按请求、按会话、按模型、按业务线拆开看，你才能回答“哪个功能最烧钱”“哪个环节最拖后腿”这类问题。

这套东西适合谁？如果你是刚把大模型 API 接进产品的后端开发，它能帮你快速建立成本意识；如果你是负责模型部署的运维或平台工程师，它能帮你定位推理引擎的性能瓶颈；如果你是技术负责人，它能让你在汇报时拿出真实数据而不是拍脑袋。哪怕你只是自己搭了个本地推理服务玩玩，加上这层观测也能让你清楚知道每次对话到底花了多少资源。

我自己的体会是，可观测性这件事，越早做越好。等到账单爆炸或者线上雪崩再去补日志，往往已经损失了一大批用户信任。下面我就按实际落地的思路，把整套方案拆开讲清楚。

2. 整体设计思路与关键选型考量

2.1 为什么不能只靠云厂商自带的用量面板

很多人第一反应是：我用的是云上的大模型服务，后台不是有用量统计吗？问题在于，云厂商的面板通常是按账号、按天聚合的，粒度太粗。你看到的是“今天总共消耗了 200 万 Token”，但你想知道的是“今天下午三点那波慢请求是哪个用户触发的、上下文有多长、用的是哪个模型”。这些细粒度信息，厂商面板给不了，必须自己在调用链路上埋点。

另一个现实问题是，很多团队并不是只用一家模型服务。可能主力用一家，备用用另一家，本地还部署了一个开源模型做兜底。这种情况下，统一的可观测层就更有必要了，否则你连横向对比都做不到。

2.2 埋点位置的选择：网关层还是业务层

埋点位置决定了你能拿到什么数据。常见做法有两种：一种是在 API 网关层统一拦截，另一种是在业务代码里逐个调用点埋。我的建议是以网关层为主、业务层为辅。

网关层的好处是统一、不易遗漏，所有经过网关的请求都能被记录，而且可以拿到完整的请求和响应体，方便计算 Token。但网关层拿不到业务语义，比如这次调用属于哪个功能模块、是哪个用户。所以业务层需要补充传递一些标签，比如在请求头里带上x-biz-module、x-user-id这类信息，网关记录时一并写入日志。

如果你们用的是类似 OpenAI 兼容协议的接口，网关层可以直接解析请求体里的messages字段和响应体里的usage字段。usage里通常包含prompt_tokens、completion_tokens、total_tokens，这是最准确的 Token 数据来源。如果某些自部署的推理引擎不返回usage，那就需要自己用分词器估算，这部分后面会细讲。

2.3 延迟指标怎么拆才有意义

延迟不是一个单一数字。用户感知的“慢”，可能来自网络传输、排队等待、模型首 Token 生成、后续 Token 流式输出等多个环节。如果只记录一个总耗时，排查问题时你会很痛苦。我习惯把延迟拆成这么几段：

• 请求排队时间：从请求到达网关到真正被推理引擎接收的时间，反映的是并发压力。
• 首 Token 延迟（TTFT）：从推理开始到第一个 Token 返回的时间，对流式对话体验影响最大。
• Token 间延迟（TPOT）：相邻 Token 之间的平均间隔，反映的是解码速度。
• 总耗时：从请求发出到最后一个 Token 返回的完整时间。

把这四个指标分开记录，你就能判断问题出在哪。比如 TTFT 正常但 TPOT 很高，说明模型解码慢，可能是显存带宽瓶颈；如果排队时间很长，说明并发数超了，需要扩容或限流。

2.4 日志存储与查询的取舍

日志量大的时候，存储和查询成本会迅速上升。我的经验是分层存储：热数据（最近 7 天）放在支持全文检索的日志系统里，方便快速排查；冷数据（7 天以上）归档到对象存储，只保留聚合后的指标。聚合指标可以按小时、按天 rollup，这样既能看趋势，又不会把存储撑爆。

查询方面，如果团队已经有 Elasticsearch 或 Loki 这类日志系统，直接复用即可。如果没有，初期用结构化日志写到文件、再用轻量工具做聚合也能顶一阵。关键是日志格式要统一，每条记录都是 JSON，字段名固定，这样后续换系统也不用改埋点代码。

3. 核心细节解析与实操要点

3.1 Token 统计的三种方式与准确性对比

Token 统计是成本核算的基础，但不同来源的数据准确性差异很大。我整理了一个对比表：

优先用usage字段，这是模型服务自己算的，最准。如果拿不到，就用对应模型的分词器本地算。注意不同模型的分词器不一样，比如有的用 BPE，有的用 SentencePiece，混用会导致统计偏差。粗略估算只适合做“今天用量是不是异常”这种预警，千万别拿它去跟业务方对账。

还有一个坑：流式响应下的 Token 统计。流式返回时，usage字段可能只在最后一个 chunk 里出现，也可能完全不出现。如果完全不出现，你只能在流结束后用分词器对完整输出做一次计算。这时候要注意把流式拼接的文本完整保存下来，否则算不准。

3.2 延迟埋点的时间戳怎么打才准

延迟计算的准确性取决于时间戳的精度和一致性。我建议在三个位置打时间戳：

1. 网关收到请求的瞬间（t0）
2. 推理引擎开始处理的瞬间（t1）
3. 收到第一个 Token 的瞬间（t2）
4. 收到最后一个 Token 的瞬间（t3）

这样 TTFT 就是t2 - t1，排队时间是t1 - t0，总耗时是t3 - t0。注意所有时间戳要用同一时钟源，跨机器的话最好用 NTP 同步，否则算出来的延迟可能是负数。

提示：如果推理引擎不暴露“开始处理”的时间点，可以用网关转发请求的时间近似代替，虽然有一点误差，但排查问题时够用了。

3.3 上下文长度对延迟的影响必须单独看

同样是一次推理，上下文 500 Token 和上下文 8000 Token 的延迟可能差好几倍。如果只统计平均延迟，长上下文请求会把短请求的数据淹没，你看不出真实分布。我的做法是按上下文长度分桶统计，比如 0-1K、1K-4K、4K-16K、16K 以上，每个桶单独算 P50、P95、P99 延迟。这样一眼就能看出长上下文是不是瓶颈。

Token 消耗同理，也要按桶看。很多时候成本超支就是因为某个功能悄悄把上下文拉长了，分桶统计能让你第一时间发现异常。

3.4 采样与全量记录的平衡

全量记录每次推理的详细日志，在请求量大的时候会产生海量数据。我的建议是指标全量、明细采样。也就是说，Token 数和延迟这类数值指标每次都记录并聚合，但完整的请求和响应内容只按一定比例采样保存，比如 1% 或 5%。采样比例可以根据流量动态调整，流量大时降低比例，流量小时提高比例。

采样时要注意保留异常请求。比如延迟超过阈值、Token 数异常大的请求，应该 100% 记录明细，方便事后分析。这个逻辑可以在网关层实现：先判断是否异常，异常就强制记录，正常则按概率采样。

4. 实操过程与核心环节实现

4.1 搭建一个最小可用的观测网关

下面用一个 Python 写的轻量网关示例，演示怎么在转发请求的同时记录 Token 和延迟。这个例子假设后端是 OpenAI 兼容接口，实际使用时把UPSTREAM_URL换成你的推理服务地址即可。

import time import json import uuid import httpx from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse app = FastAPI() UPSTREAM_URL = "http://localhost:8000/v1/chat/completions" @app.post("/v1/chat/completions") async def proxy(request: Request): body = await request.json() request_id = str(uuid.uuid4()) t0 = time.time() # 记录请求元信息 log_entry = { "request_id": request_id, "model": body.get("model"), "stream": body.get("stream", False), "message_count": len(body.get("messages", [])), "biz_module": request.headers.get("x-biz-module", "unknown"), "user_id": request.headers.get("x-user-id", "anonymous"), } async with httpx.AsyncClient(timeout=120) as client: if body.get("stream"): # 流式处理 async def stream_generator(): t1 = time.time() first_token_time = None last_chunk = None async with client.stream("POST", UPSTREAM_URL, json=body) as resp: async for line in resp.aiter_lines(): if line.startswith("data: ") and line != "data: [DONE]": if first_token_time is None: first_token_time = time.time() last_chunk = line yield line + "\n" t3 = time.time() # 从最后一个 chunk 提取 usage usage = {} if last_chunk: try: data = json.loads(last_chunk[6:]) usage = data.get("usage", {}) except Exception: pass log_entry.update({ "queue_time_ms": round((t1 - t0) * 1000, 2), "ttft_ms": round((first_token_time - t1) * 1000, 2) if first_token_time else None, "total_ms": round((t3 - t0) * 1000, 2), "prompt_tokens": usage.get("prompt_tokens"), "completion_tokens": usage.get("completion_tokens"), "total_tokens": usage.get("total_tokens"), }) write_log(log_entry) return StreamingResponse(stream_generator(), media_type="text/event-stream") else: t1 = time.time() resp = await client.post(UPSTREAM_URL, json=body) t3 = time.time() data = resp.json() usage = data.get("usage", {}) log_entry.update({ "queue_time_ms": round((t1 - t0) * 1000, 2), "ttft_ms": round((t3 - t1) * 1000, 2), "total_ms": round((t3 - t0) * 1000, 2), "prompt_tokens": usage.get("prompt_tokens"), "completion_tokens": usage.get("completion_tokens"), "total_tokens": usage.get("total_tokens"), }) write_log(log_entry) return data def write_log(entry): # 实际使用时替换为写入日志系统或消息队列 print(json.dumps(entry, ensure_ascii=False))

这段代码的核心思路是：请求进来先打时间戳，转发给上游，流式响应时逐行透传并记录首 Token 时间，结束后从最后一个 chunk 提取 usage 并写入日志。非流式则直接等响应返回后记录。

4.2 用分词器兜底计算 Token

如果上游不返回 usage，就需要本地计算。以 HuggingFace 的 tokenizer 为例：

from transformers import AutoTokenizer tokenizer = AutoTokenizer.from_pretrained("your-model-path") def count_tokens(messages): # 简单拼接，实际应按模型的 chat template 处理 text = "".join([m["content"] for m in messages]) return len(tokenizer.encode(text))

注意这里只是演示，实际计算时要按模型的 chat template 来拼接，否则算出来的数和模型真实消耗会有偏差。不同模型的 template 不一样，有的加特殊 token，有的不加，这个细节会直接影响统计准确性。

4.3 延迟分桶统计的实现

分桶统计可以在日志写入后由聚合任务完成。下面是一个简单的聚合逻辑：

def bucket_by_context_length(logs): buckets = {"0-1K": [], "1K-4K": [], "4K-16K": [], "16K+": []} for log in logs: tokens = log.get("prompt_tokens") or 0 if tokens < 1000: buckets["0-1K"].append(log) elif tokens < 4000: buckets["1K-4K"].append(log) elif tokens < 16000: buckets["4K-16K"].append(log) else: buckets["16K+"].append(log) return buckets def percentile(values, p): if not values: return None values = sorted(values) idx = int(len(values) * p / 100) return values[min(idx, len(values) - 1)]

拿到分桶后的数据，就可以分别算每个桶的 P50、P95、P99 延迟，以及平均 Token 消耗。这样报表出来，长上下文的影响一目了然。

4.4 告警规则怎么设才不吵人

告警设得太敏感，天天响，大家就麻木了；设得太迟钝，真出问题又发现不了。我的经验是设两级告警：

• 预警：P95 延迟超过基线 1.5 倍，或小时级 Token 消耗超过预算 80%，发到团队频道，不强制响应。
• 严重告警：P99 延迟超过基线 3 倍，或小时级 Token 消耗超过预算 120%，或错误率超过 5%，直接打电话。

基线怎么定？上线后先跑一周，取正常时段的 P95 作为基线，之后每周滚动更新。这样基线会随着业务变化自动调整，不会因为业务增长而误报。

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

5.1 Token 数和账单对不上怎么办

这是最常见的问题。可能的原因有几个：一是统计口径不一致，你算的是输入加输出，账单可能只算输出；二是重试请求被重复计费，但你的日志只记了一次；三是流式响应的 usage 没拿到，用了估算值导致偏差。

排查时先确认口径，再检查是否有重试逻辑。重试一定要在日志里标记出来，比如加一个retry_count字段，这样对账时能把重试的消耗单独拎出来看。

5.2 延迟突然升高但找不到原因

延迟升高通常有几个方向：上游模型服务变慢、网络抖动、并发排队、上下文变长。排查顺序建议是：先看排队时间，如果排队时间长，说明并发压力大；再看 TTFT，如果 TTFT 高但排队正常，说明模型侧慢；最后看上下文分桶，如果长上下文请求占比突然升高，那就是业务侧的问题。

我遇到过一次延迟飙升，最后发现是某个新上线的功能把系统提示词写得太长，每次请求都多带了两千 Token。这种问题不看分桶统计根本发现不了。

5.3 流式响应下首 Token 时间记录不准

流式响应时，首 Token 时间应该以第一个包含实际内容的 chunk 为准，而不是第一个 SSE 事件。有些服务会先发一个空的 role chunk，再发内容 chunk，如果把空 chunk 当首 Token，TTFT 会偏小。处理时判断一下 chunk 里是否有实际文本内容。

5.4 日志量太大导致存储成本失控

前面提过采样，这里补充一个技巧：对日志字段做裁剪。完整的请求和响应内容很占空间，但排查问题时往往只需要看前几百个字符。可以在写入前对内容字段做截断，比如只保留前 500 字符，超出部分用省略号代替。这样既能保留关键信息，又能大幅降低存储量。

5.5 常见问题速查表

注意：排查问题时优先看聚合指标，定位到异常时间段后再去翻明细日志。反过来做效率极低。

6. 我踩过的坑和几条实用建议

第一个坑是时间戳精度。早期我用秒级时间戳，结果很多请求的总耗时算出来是 0，因为太快了。后来统一改成毫秒级，问题解决。如果你的服务延迟本身就在几十毫秒级别，秒级时间戳完全不够用。

第二个坑是流式响应的日志写入时机。一开始我在流结束后才写日志，结果如果客户端提前断开连接，日志就丢了。后来改成在生成器里用 try-finally 包裹，确保无论正常结束还是异常断开都能写入日志。

第三个坑是多模型混用时的 Token 统计。不同模型的分词器不一样，用同一个分词器算所有模型的 Token 会不准。后来我给每个模型配置了对应的分词器，统计才准确。

最后分享一个我觉得很实用的小技巧：在日志里加一个trace_id，从网关一直透传到业务代码和下游服务。这样排查问题时，你可以用这一个 ID 把整条链路的日志串起来，不用在多个系统之间来回跳。这个习惯一旦养成，排查效率会提升一大截。

这套观测体系我前后迭代了三四版，从最开始只记总耗时和总 Token，到现在分桶、分段、采样、告警一应俱全。每次迭代都是被实际问题逼出来的。如果你刚开始做，不用一步到位，先把 Token 和总延迟记起来，跑一段时间有了数据，再逐步细化。数据本身会告诉你下一步该补什么。