企业官网建设流程全解析

如果你负责一个已经上线的 LLM 应用，排查过一次线上问题，大概率会撞上同一个尴尬：日志系统里明明记录了 prompt、模型返回和 token 用量，但你仍然无法完整还原“一次 LLM 调用到底发生了什么”。

问题不在于日志不够多，而在于——业界至今没有一个公认的 “LLM 调用到底该记录什么” 的标准。OpenTelemetry 在传统微服务领域统一了 trace 和日志，但放到 LLM 场景，各家方案仍然各写各的 schema。于是，我基于自己的项目实践构建了一个 record：它不是又一款可观测性平台，而是一套描述 LLM 调用记录的数据结构，并配套实现了一个尽量小的记录器。

这篇文章会讲清楚三件事：为什么传统日志和普通 APM 管理不了 LLM 调用；业界做 LLM 可观测性的工具走到哪一步，为什么仍然缺“标准”；我设计的一套 LLM call record 长什么样，怎么在真实项目里记录、怎么用它验证和排查问题。

1. 先从一个真实场景说起：查一次线上问题，为什么这么难

假设你负责一个已经上线的 LLM 客服系统，某天用户投诉，某个回答明显偏离了事实，甚至有“编造”的嫌疑。你第一时间去查日志，发现日志里信息很全：请求时间、模型名称、prompt、response、token 用量，一样不少。但真正开始排查时，你会遇到几个很难受的问题。

第一个问题：日志里的 prompt 不是真实请求。你在日志里记录的 prompt，是拼接后的最终字符串。但在生产系统里，这个 prompt 往往由系统提示词、用户问题、知识库检索片段、历史对话拼接而成。你只记录了最终 prompt，却没法确认知识库片段是怎么排序的、哪个片段对最终回答影响最大。想追溯模型为什么给出这个结论，缺少关键证据。

第二个问题：多个日志条目之间的关联方式很弱。一次 LLM 调用可能涉及多轮重试、流式输出、工具调用、后处理，会分散在十几个日志文件里。你靠一个 request_id 去 grep 半天，仍然拼不出一张完整的调用时序图。如果某个环节是异步执行的，连 request_id 都可能对不上。

第三个问题：每个系统记录的字段完全不一样。网关记 JSON 格式，业务服务记文本格式，模型供应商的控制台有一套自己的视图，第三方可观测平台又有另一套 schema。你想写一个统一的审计脚本，最后发现光是解析各种日志格式就花掉大半天。

这里最核心的问题不是日志打得不够多，而是我们缺少一个统一的“LLM 调用到底该记录什么”的规范。看这个现象，很容易形成一个错误判断：把 LLM 调用记录等同于普通的接口日志，用传统的 logging 库打几行 key-value 就行。但 LLM 调用远比普通 API 调用复杂，它携带的上下文、工具调用、流式输出、token 用量、模型版本、随机参数、成本等数据，远远超出了普通日志的承载能力。

换句话说，这不是一个“多打几行日志”就能解决的问题，而是一个数据建模问题。如果你不想等公共标准落地，那就需要自己先定义一套可扩展、可分析、可审计的 LLM call record。

2. 传统日志和普通 APM 为什么管不了 LLM 调用

2.1 LLM 调用与普通 REST 调用的差异

传统分布式系统里，一次 HTTP 调用通常可以用 trace 完整还原：客户端地址、请求路径、状态码、响应时间、响应体大小。中间件的调用关系是确定的，数据流是相对线性的。

但一次 LLM 调用往往不是一次普通 REST 请求可以概括的。我把它和普通 REST 调用做了个对比：

这张表说明，LLM 调用需要记录的信息不只多，而且类型不同。传统 APM 把重点放在调用链和服务节点上，默认一次调用是可重复、可确定追踪的，但 LLM 生成的随机性和上下文敏感性，意味着调用记录必须能还原“当时的上下文”。

2.2 普通日志的弱点

有人会问：用 logging.info 把请求和响应都打出来不行吗？可以，但它有几个很难补齐的短板。

一是缺少类型约束。prompt、completion、tool_call、usage 这些字段如果都塞进字符串日志，后续想按 model、provider、trace_id 过滤，只能靠正则，非常痛苦。二是缺少关联结构。一次 LLM 调用中包含多个事件（请求发出、流式首包、完成、工具调用、用户反馈），普通日志只能按时间先后平铺，很难表达这些事件之间的层级关系。三是缺少指标属性。业务日志不会天然计算 token 消耗、模型延迟、成本估算，这些计算逻辑需要在日志体系中额外实现。

所以，LLM 调用记录本质上更适合用“结构化事件 + 结构化字段”来承载，而不是简单的文本日志。这也是我把方案定义为 record 而不是 log 的原因：它是一份可解析、可计算的数据记录，不是给人类看的纯文本。

2.3 这也是标准要解决的核心问题

没有标准，意味着每个团队都要花精力去定义自己的字段名、字段类型、存储方式和汇总口径。你给 A 服务的 LLM 调用建了 schema，到了 B 服务又需要一套新的。

如果能有一个清晰的、最小化的 LLM call record 标准，保证所有 LLM 调用都能记录下：是谁调的、调的哪个模型、发出去什么、收到什么、花掉多少 token、耗时多少、最终状态如何，那么团队内部的数据建模、成本核算、审计复盘就会变得非常顺畅。标准不关心某个工具好不好用，它关心的是数据能不能在不同系统之间稳定地流转。

3. 一个 LLM 调用记录应该包含哪些维度

在写代码之前，先把“该记录什么”讲清楚。我把自己总结出来的 LLM 调用记录维度分成六类。

3.1 基本身份信息

这部分解决“这条记录是谁”的问题，包括 record_id、trace_id、provider、model、action。record_id 是单条记录的唯一 ID，trace_id 是整条业务链路的追踪 ID，provider 是模型提供方，model 是具体模型名，action 表示调用类型，例如 chat.completion、embedding、tool_call。

3.2 请求信息

请求信息解决“模型当时收到了什么”的问题。这里建议直接记录最终发给模型的完整请求体，包括 messages 列表、工具定义、模型参数（temperature、top_p、max_tokens），以及外部上下文来源。注意，外部上下文来源非常关键。如果你用了 RAG，最好把检索结果的 chunk_id、来源文档、相似度 score 也记下来，后续排查“模型是不是因为看到某篇资料才这样回答”时非常有用。

3.3 响应信息

响应信息解决“模型返回了什么”的问题，包括模型返回的完整内容、工具调用参数、finish_reason、是否流式输出。如果模型供应商返回了内容安全标记，也应该保留。

3.4 用量与性能

这部分是成本分析和性能分析的依据，包括 prompt_tokens、completion_tokens、total_tokens、latency_ms、首 token 延迟、缓存命中情况。在 Agent 或多轮对话场景下，token 用量的准确记录直接决定成本归因是否可靠。

3.5 状态与错误

状态与错误解决“这次调用是否成功”的问题，包括 HTTP 状态码、错误类型、重试次数、异常描述。我特别建议把错误信息作为结构化字段保存，而不是只记录一个 error_code。因为不同模型供应商的错误格式不一样，结构化字段能方便后续做统一的错误聚合分析。

3.6 元信息

元信息是给业务侧用的，包括 user_id、session_id、功能模块、环境、数据版本、知识库版本等。元信息的作用是让记录能在业务维度上做切片查询，例如“某个用户 24 小时内调用了多少次模型”“A/B 实验里新提示词的 token 消耗如何”。

汇总成一张表格就是：

设计格式时，建议遵守“分层记录、最少必要字段、统一命名”三个原则。最少必要字段是为了降低接入成本，统一命名是为了后续能够跨服务聚合。

4. 现状盘点：业界差“标准”到底差在哪

既然标题说“没有标准”，那很有必要看看业界目前有哪些尝试。

传统可观测性领域，OpenTelemetry 已经定义了比较完善的 trace、metrics、logs 语义约定。面向 LLM 和 GenAI 的新语义约定也在演进中，例如 GenAI 相关的 semantic conventions 已经覆盖了模型调用、Agent 等领域。这套规范的价值在于：如果所有厂商都遵循同一套 span 属性和命名，那么不同工具之间的数据是可以互通的。

但现实中，LLM 可观测领域的格局还比较分散。主流的 LLM 观测工具包括 Langfuse、LangSmith、Traceloop OpenLLMetry、Arize Phoenix、MLflow Tracing 等，它们的方向各有侧重：有的提供非常完整的 trace 可视化，适合调试 Agent 链路；有的擅长评测，可以拿线上日志回放来构造评测集；有的专注成本管理，适合看 token 消耗；有的作为开源库，接入了 OpenTelemetry 生态。

这些工具的共性，是都提供了“记录 LLM 调用”的能力，但没有一个能做到所有工具都使用同一套 record 结构。你在这家工具里导出的 trace JSON，到了另一家平台，仍然需要做字段映射。从工程视角看，这里真正缺少的不是又一个可视化平台，而是一个中立的、足够通用的 LLM 调用记录标准。

标准不解决工具好不好用的问题，它解决的是数据能不能互认的问题。只要你自己定义了一个清晰的记录格式，即便不接入商业平台，也可以把记录落成 JSONL、导入数据仓库、做成本审计、做回归评测。这也是我为什么要自己构建一套 record 的原因：等一个公共标准成熟可能要很久，但眼下项目就需要记录数据。自己先设计一套最小可用的格式，将来标准成熟了再迁移，成本也远小于从零开始。

5. 我的方案：LLM Call Record 的完整设计与实现

以下代码基于一个假设场景：你的应用通过一个 OpenAI 兼容的 chat/completions HTTP 端点访问大模型。我们用 Python 实现一个 record 记录器，把一次 LLM 调用的完整信息保存下来。

5.1 数据模型设计

先定义 LLMRecord 数据类，对应前面第 3 节的六个维度。使用 dataclass 实现，方便序列化和演进。

# llm_record/models.py from dataclasses import dataclass, field, asdict from datetime import datetime, timezone from typing import Any, Dict, List, Optional @dataclass class LLMRecord: record_id: str trace_id: str provider: str model: str action: str request_payload: Dict[str, Any] response_payload: Optional[Dict[str, Any]] = None prompt_tokens: Optional[int] = None completion_tokens: Optional[int] = None total_tokens: Optional[int] = None latency_ms: float = 0.0 status: str = "pending" # pending / success / error error: Optional[str] = None retry_count: int = 0 user_id: Optional[str] = None session_id: Optional[str] = None context_sources: List[Dict[str, Any]] = field(default_factory=list) metadata: Dict[str, Any] = field(default_factory=dict) created_at: str = field( default_factory=lambda: datetime.now(timezone.utc).isoformat() ) def to_dict(self) -> Dict[str, Any]: return asdict(self) def to_json(self) -> str: import json return json.dumps(asdict(self), ensure_ascii=False, indent=2)

这里有几个设计细节。

request_payload 和 response_payload 都保存完整的原始结构，不把它拍平成字符串。这样后续可以用 jq 或 Python 脚本对不同类型的调用做二次分析。token 用量拆成 prompt_tokens / completion_tokens / total_tokens，而不是存一个嵌套对象，便于在数据仓库里做 SQL 查询。context_sources 字段用来记录 RAG 检索结果或外部工具来源。status 字段默认设置为 pending，调用结束后再更新为 success 或 error，这样即使在调用过程中崩溃，也能保留一条半成品记录用于排查。

5.2 记录写入器

接着实现一个简单的 JSONL 写入器。JSONL 每行一个 JSON 对象，写入简单、追加方便，可以配合 DuckDB、Spark 或日志采集器直接分析，是落地成本最低的存储形态。

# llm_record/writer.py import json from pathlib import Path from threading import Lock from typing import Dict, List from llm_record.models import LLMRecord class JSONLRecordWriter: """将 LLMRecord 追加写入 JSONL 文件。""" def __init__(self, write_dir: str = "records", file_prefix: str = "llm"): self.write_dir = Path(write_dir) self.write_dir.mkdir(parents=True, exist_ok=True) self.file_prefix = file_prefix self._lock = Lock() self._buffer: List[Dict[str, Any]] = [] def _get_output_path(self) -> Path: from datetime import datetime, timezone day = datetime.now(timezone.utc).strftime("%Y%m%d") return self.write_dir / f"{self.file_prefix}-{day}.jsonl" def append(self, record: LLMRecord) -> None: with self._lock: self._buffer.append(record.to_dict()) self._flush_if_needed() def _flush_if_needed(self) -> None: if len(self._buffer) < 10: return self.flush() def flush(self) -> None: with self._lock: if not self._buffer: return path = self._get_output_path() with path.open("a", encoding="utf-8") as fp: for item in self._buffer: fp.write(json.dumps(item, ensure_ascii=False) + "\n") self._buffer = []

注意这里用 lock 保证多线程安全，用 buffer 做批量写入，每 10 条或主动 flush 时写一次。实际生产场景可以考虑替换成异步队列，或者直接发送到消息中间件，避免记录动作阻塞模型调用主链路。

5.3 完整的 LLM 调用记录封装

下面实现核心函数：调用模型 + 自动生成记录。为了减小依赖，使用标准库 urllib 发起 HTTP 请求，这样你直接复制代码到项目里就能跑，不需要安装额外 SDK。

# examples/demo_record.py import json import time import uuid from urllib.request import Request, urlopen from urllib.error import HTTPError from llm_record.models import LLMRecord from llm_record.writer import JSONLRecordWriter def create_llm_record( trace_id: str, provider: str, model: str, payload: dict, *, user_id: str = None, session_id: str = None, context_sources: list = None, ) -> LLMRecord: return LLMRecord( record_id=uuid.uuid4().hex, trace_id=trace_id, provider=provider, model=model, action="chat.completion", request_payload=payload, user_id=user_id, session_id=session_id, context_sources=context_sources or [], ) def call_llm_and_record( endpoint: str, api_key: str, model: str, messages: list, *, trace_id: str = None, temperature: float = 0.7, user_id: str = None, session_id: str = None, context_sources: list = None, writer: JSONLRecordWriter = None, ): trace_id = trace_id or uuid.uuid4().hex payload = { "model": model, "messages": messages, "temperature": temperature, } record = create_llm_record( trace_id=trace_id, provider="openai-compatible", model=model, payload=payload, user_id=user_id, session_id=session_id, context_sources=context_sources, ) start = time.perf_counter() body = json.dumps(payload).encode("utf-8") req = Request( endpoint, data=body, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {api_key}", }, method="POST", ) try: with urlopen(req, timeout=60) as resp: response_json = json.loads(resp.read().decode("utf-8")) record.latency_ms = (time.perf_counter() - start) * 1000 record.response_payload = response_json usage = response_json.get("usage", {}) record.prompt_tokens = usage.get("prompt_tokens") record.completion_tokens = usage.get("completion_tokens") record.total_tokens = usage.get("total_tokens") record.status = "success" except HTTPError as e: record.latency_ms = (time.perf_counter() - start) * 1000