如果你负责一个已经上线的 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 调用做了个对比对比维度普通 REST 调用LLM 调用请求体结构化参数字段固定自然语言 prompt结构松散可能包含系统提示、历史消息、工具定义、RAG 片段响应体结构化数据可直接解析自然语言生成结果可能流式返回可能包含工具调用指令状态HTTP 状态码即可表示需要记录 finish_reason、内容审核标志、流式结束状态等成本基本可忽略需要记录 token 用量、模型单价、缓存命中情况上下文依赖无状态或通过请求头传递依赖会话历史、向量检索结果、外部工具返回重试行为简单重试即可重试可能带来额外 token 消耗和不同的生成结果这张表说明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 是单条记录的唯一 IDtrace_id 是整条业务链路的追踪 IDprovider 是模型提供方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 消耗如何”。汇总成一张表格就是类别字段示例说明基本身份record_id, trace_id, provider, model, action决定这是一条什么样的记录以及它在全局哪里请求信息messages, tools, params, context_sources还原模型收到什么响应信息content, tool_calls, finish_reason还原模型返回什么用量性能prompt_tokens, completion_tokens, latency_ms, cost做成本核算和性能分析状态错误status_code, error_type, retry_count做故障排查元信息user_id, session_id, env, version做业务分析和审计设计格式时建议遵守“分层记录、最少必要字段、统一命名”三个原则。最少必要字段是为了降低接入成本统一命名是为了后续能够跨服务聚合。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_factorylist) metadata: Dict[str, Any] field(default_factorydict) created_at: str field( default_factorylambda: 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_asciiFalse, indent2)这里有几个设计细节。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(parentsTrue, exist_okTrue) 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, encodingutf-8) as fp: for item in self._buffer: fp.write(json.dumps(item, ensure_asciiFalse) \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_iduuid.uuid4().hex, trace_idtrace_id, providerprovider, modelmodel, actionchat.completion, request_payloadpayload, user_iduser_id, session_idsession_id, context_sourcescontext_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_idtrace_id, provideropenai-compatible, modelmodel, payloadpayload, user_iduser_id, session_idsession_id, context_sourcescontext_sources, ) start time.perf_counter() body json.dumps(payload).encode(utf-8) req Request( endpoint, databody, headers{ Content-Type: application/json, Authorization: fBearer {api_key}, }, methodPOST, ) try: with urlopen(req, timeout60) 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
大模型调用日志怎么记？一套可落地的LLM Call Record标准与实现
如果你负责一个已经上线的 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 调用做了个对比对比维度普通 REST 调用LLM 调用请求体结构化参数字段固定自然语言 prompt结构松散可能包含系统提示、历史消息、工具定义、RAG 片段响应体结构化数据可直接解析自然语言生成结果可能流式返回可能包含工具调用指令状态HTTP 状态码即可表示需要记录 finish_reason、内容审核标志、流式结束状态等成本基本可忽略需要记录 token 用量、模型单价、缓存命中情况上下文依赖无状态或通过请求头传递依赖会话历史、向量检索结果、外部工具返回重试行为简单重试即可重试可能带来额外 token 消耗和不同的生成结果这张表说明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 是单条记录的唯一 IDtrace_id 是整条业务链路的追踪 IDprovider 是模型提供方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 消耗如何”。汇总成一张表格就是类别字段示例说明基本身份record_id, trace_id, provider, model, action决定这是一条什么样的记录以及它在全局哪里请求信息messages, tools, params, context_sources还原模型收到什么响应信息content, tool_calls, finish_reason还原模型返回什么用量性能prompt_tokens, completion_tokens, latency_ms, cost做成本核算和性能分析状态错误status_code, error_type, retry_count做故障排查元信息user_id, session_id, env, version做业务分析和审计设计格式时建议遵守“分层记录、最少必要字段、统一命名”三个原则。最少必要字段是为了降低接入成本统一命名是为了后续能够跨服务聚合。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_factorylist) metadata: Dict[str, Any] field(default_factorydict) created_at: str field( default_factorylambda: 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_asciiFalse, indent2)这里有几个设计细节。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(parentsTrue, exist_okTrue) 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, encodingutf-8) as fp: for item in self._buffer: fp.write(json.dumps(item, ensure_asciiFalse) \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_iduuid.uuid4().hex, trace_idtrace_id, providerprovider, modelmodel, actionchat.completion, request_payloadpayload, user_iduser_id, session_idsession_id, context_sourcescontext_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_idtrace_id, provideropenai-compatible, modelmodel, payloadpayload, user_iduser_id, session_idsession_id, context_sourcescontext_sources, ) start time.perf_counter() body json.dumps(payload).encode(utf-8) req Request( endpoint, databody, headers{ Content-Type: application/json, Authorization: fBearer {api_key}, }, methodPOST, ) try: with urlopen(req, timeout60) 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