1. 从一次线上事故说起LLM 应用为什么这么难调试先讲一个真实场景。某天晚上一个基于大语言模型LLM的工单分类服务突然出问题用户提交的工单被分到了完全无关的类别。第一反应是看日志结果发现日志里只记录了“调用成功返回结果xxx”完全没有记录当时传进去的上下文窗口里到底有什么内容。你想复现这个 Bug但用同样的 Prompt 测了几次结果又都是对的。这就是 LLM 应用开发最大的痛点之一结果非确定性。同样的输入温度参数不为 0 时模型可能给出不同答案上下文窗口里的历史消息、系统提示词、工具返回结果之间一旦互相干扰问题就更难定位。传统软件开发里我们可以靠日志、断点、单元测试、覆盖率来逼近问题根因。但到了 LLM 应用里程序的行为有一半发生在“模型内部”这一部分是黑盒我们无法直接观测。不过还有一半行为是我们可以观测的那就是模型调用前后的数据流——你发给模型什么、模型返回什么、中间经过了多少轮上下文截断、每轮消耗了多少 Token、哪个环节改变了消息结构。把这些可观测的数据完整记录下来再像“程序分析”一样去检查它们就能让 LLM 应用的调试从“猜”变成“查”。这就是 Lemmalog 的出发点把 LLM 的记忆上下文、状态、调用轨迹变成可查询、可分析的程序数据。这篇文章会围绕 Lemmalog 的设计思路讲三件事为什么普通日志系统不足以支撑 LLM 应用排错。Lemmalog 如何设计“LLM 记忆采集 结构化存储 程序分析”三层结构。一个最小可运行的 Python 实现帮你理解核心原理。无论你是在做 AI Agent、RAG 应用还是想给团队搭一套 LLM 可观测性基础工具这篇内容都能给你一些可落地的思路。2. 核心概念LLM 记忆、程序分析与 Lemmalog 的定位2.1 什么是“LLM 记忆”在传统程序里程序的“状态”保存在变量、数据库、缓存中。而 LLM 应用的“状态”本质上由三部分组成上下文窗口Context Window当前请求的所有消息包括 system prompt、历史对话、工具返回值、检索到的文档片段。会话状态Session State跨请求保留的变量比如用户 ID、会话 ID、当前步骤、临时选择的结构化数据。外部记忆External Memory向量数据库、KV 缓存、普通数据库中的内容它们会在需要时被注入到上下文窗口里。为什么“记忆”这么重要因为在 LLM 应用里模型的输出完全由输入决定而输入中真正不可控的往往就是记忆部分。上下文被截断、历史消息顺序错乱、检索结果重复注入这些问题几乎不会在代码层面报错但会导致模型行为异常。Lemmalog 要做的第一件事就是把上面这三类记忆“快照”下来让它们和每一次模型调用关联起来。2.2 什么是“程序分析”程序分析Program Analysis并不是一个新概念。传统的程序分析包括静态分析不运行代码直接扫描源码找问题和动态分析运行代码跟踪变量、路径、性能。它关注的问题是这段程序里是否存在某种状态异常某个变量在哪里被赋值在哪里被读取是否可能为 null某些条件分支是否永远无法到达当我们把 LLM 应用的每次调用想象成一次“函数执行”把上下文窗口想象成“入参”把模型输出想象成“返回值”程序分析的方法论就可以平移过来调用轨迹一次用户请求到底触发了几次 LLM 调用数据流哪些系统变量进入了 Prompt它们是如何被修改的状态检查某次调用的 Token 数是否异常历史消息是否被错误截断一致性分析多轮对话中同一类请求的 Prompt 结构是否保持稳定Lemmalog 这个名字就是“Lemma引理 Log日志”的合成词。它想表达的是一种态度用证明引理的严谨性去记录每一次模型交互。2.3 Lemmalog 与现有方案的区别市面上其实已经有 LangSmith、Langfuse 这类 LLM 可观测平台。它们的核心能力是追踪Tracing、监控Monitoring和可视化Visualization适合在团队内快速搭建可观测性。Lemmalog 的定位不太一样。它更强调“分析”而不是“展示”。它会把采集到的记忆数据落成结构化的关系型数据让你可以用 SQL 或者简单脚本去执行一些程序分析任务。比如找出所有“在同一会话内第二次调用时历史消息超过 50 条”的请求。分析某种工具返回结果被注入 System Prompt 后模型输出格式是否发生变化。统计不同 Prompt 模板的 Token 消耗中位数定位成本异常。换句话说Lemmalog 不是一个完整平台而是一种设计模式和轻量级落地框架。它强调你不需要一开始就上重平台你可以先在自己的代码里实现一条“记忆 → 日志 → 分析”的流水线后续再扩展成完整服务。3. 环境准备与工程结构3.1 环境版本本文示例代码使用 Python 编写核心运行环境如下依赖项版本建议说明Python3.10使用了新版类型注解语法 Xopenai按你实际使用的版本本文用官方 SDK 做示例重点不在具体模型pydantic2.x用于配置解析和数据校验SQLite3Python 内置零依赖存储后端方便本地运行不同版本的依赖库在 API 上可能存在差异。如果你使用的 OpenAI SDK 版本较早方法签名可能是openai.ChatCompletion.create(...)而不是client.chat.completions.create(...)。读者需要根据自己的实际环境调整。3.2 示例项目结构lemmalog-demo/ ├── lemmalog/ │ ├── __init__.py │ ├── trace_context.py # 调用链上下文 │ ├── storage.py # SQLite 存储层 │ ├── analyzer.py # 分析器查询 检查规则 │ └── collector.py # LLM 调用采集装饰器 ├── examples/ │ └── chat_demo.py # 完整可运行的示例 └── requirements.txt这个结构刻意保持精简。实际项目中你可以把 storage 层替换成 PostgreSQL把 analyzer 层替换成定时任务或事件回调。3.3 设计目标在动手写代码前先明确 Lemmalog 的三个设计原则非侵入性业务代码不需要大规模改造。理想状态下只需要在 LLM 调用处加一行装饰器或者在请求入口设置一个上下文。持久化优先所有记忆数据必须落盘不能只存在内存里或 Redis 里。因为故障分析往往发生在事故之后。分析友好存储格式要方便后续做切片、聚合、关联查询。因此使用关系型结构而不是纯文本日志。4. 核心原理拆解一条 LLM 调用是如何被“分析化”的4.1 从调用到轨迹一次最简单的 LLM 调用在程序层面可以看成response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, )要分析这次调用我们需要采集以下数据数据项示例值用途会话 IDsession_123关联多轮对话请求 IDreq_456唯一标识一次请求模型名称gpt-4o-mini迭代对比不同模型行为请求消息序列化后的 messages 数组重放现场检查 Prompt 结构响应结果完整响应内容检查输出质量Token 使用输入 1024输出 512成本分析与截断检测耗时1234 ms性能监控上下文快照当时的会话状态、检索到的文档片段排查历史消息干扰Lemmalog 的采集层就是围绕这些字段设计的。4.2 从轨迹到分析采集到数据只是第一步。关键是“分析”。我们可以定义三种分析粒度第一层单次调用分析检查一次调用中是否存在异常。例如输入 Token 数是否超过模型最大限制输出是否被截断响应是否因为max_tokens设置太小而中断第二层调用序列分析检查一个会话内的多次调用是否连贯。例如第二轮调用是否携带了第一轮的完整历史工具调用返回的临时数据有没有在下一次 Prompt 中被错误保留第三层跨会话统计分析把大量会话放在一起对比。例如某个 Prompt 模板在历史消息长度超过阈值之后模型输出是否出现了格式漂移Lemmalog 的分析器就是围绕这三个层级去设计查询接口的。4.3 核心抽象TraceContext在代码实现中我们使用一个TraceContext对象来管理记忆快照。它的职责是在请求开始时创建。在整个调用链中传递通过参数或者上下文变量。每次 LLM 调用都会向当前 TraceContext 追加一条 Trace 记录。在请求结束时把整个 Trace 批量写入存储层。下面进入代码实现部分。5. Lemmalog 最小实现实战5.1 创建项目与依赖先创建虚拟环境并安装依赖。mkdir lemmalog-demo cd lemmalog-demo python3 -m venv .venv source .venv/bin/activate pip install openai pydantic如果你没有 OpenAI API Key也可以把核心代码跑通只需要在调用采集部分用 mock 数据代替真实调用。5.2 定义 Trace 数据结构首先定义数据类。项目路径lemmalog/trace_models.py。# lemmalog/trace_models.py from __future__ import annotations import time import uuid from typing import Any, Optional from pydantic import BaseModel, Field class TraceRecord(BaseModel): 单次 LLM 调用轨迹记录 trace_id: str Field(default_factorylambda: uuid.uuid4().hex) session_id: str request_id: str model: str messages: list[dict[str, Any]] response: Optional[str] None prompt_tokens: int 0 completion_tokens: int 0 total_tokens: int 0 latency_ms: int 0 context_snapshot: dict[str, Any] Field(default_factorydict) timestamp: float Field(default_factorytime.time) status: str success error_message: Optional[str] None class TraceSession(BaseModel): 一个业务请求对应的完整跟踪会话 session_id: str request_id: str start_time: float Field(default_factorytime.time) end_time: Optional[float] None records: list[TraceRecord] Field(default_factorylist) def add_record(self, record: TraceRecord) - None: self.records.append(record) def complete(self) - None: self.end_time time.time()这里的核心是TraceRecord。可以看到它不仅记录了常见的输入输出还记录了context_snapshot这用来保存调用发生时外部记忆的内容。5.3 实现存储层使用 SQLite 作为存储。这个模块负责建表、插入记录、提供基础查询接口。项目路径lemmalog/storage.py。# lemmalog/storage.py import json import sqlite3 from pathlib import Path from typing import Any from lemmalog.trace_models import TraceRecord, TraceSession class SQLiteStorage: SQLite 存储实现 def __init__(self, db_path: str lemmalog.db): self.db_path db_path self._init_db() def _init_db(self) - None: with self._connect() as conn: conn.execute( CREATE TABLE IF NOT EXISTS llm_trace ( trace_id TEXT PRIMARY KEY, session_id TEXT NOT NULL, request_id TEXT NOT NULL, model TEXT NOT NULL, messages TEXT NOT NULL, response TEXT, prompt_tokens INTEGER, completion_tokens INTEGER, total_tokens INTEGER, latency_ms INTEGER, context_snapshot TEXT, timestamp REAL, status TEXT, error_message TEXT ) ) conn.execute( CREATE INDEX IF NOT EXISTS idx_session_id ON llm_trace(session_id) ) conn.execute( CREATE INDEX IF NOT EXISTS idx_request_id ON llm_trace(request_id) ) conn.commit() def _connect(self) - sqlite3.Connection: conn sqlite3.connect(self.db_path) conn.row_factory sqlite3.Row return conn def save_record(self, record: TraceRecord) - None: with self._connect() as conn: conn.execute( INSERT INTO llm_trace ( trace_id, session_id, request_id, model, messages, response, prompt_tokens, completion_tokens, total_tokens, latency_ms, context_snapshot, timestamp, status, error_message ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( record.trace_id, record.session_id, record.request_id, record.model, json.dumps(record.messages, ensure_asciiFalse), record.response, record.prompt_tokens, record.completion_tokens, record.total_tokens, record.latency_ms, json.dumps(record.context_snapshot, ensure_asciiFalse), record.timestamp, record.status, record.error_message, ), ) conn.commit() def query_by_session(self, session_id: str) - list[sqlite3.Row]: with self._connect() as conn: rows conn.execute( SELECT * FROM llm_trace WHERE session_id ? ORDER BY timestamp ASC, (session_id,), ).fetchall() return rows def query_all(self, limit: int 100) - list[sqlite3.Row]: with self._connect() as conn: rows conn.execute( SELECT * FROM llm_trace ORDER BY timestamp DESC LIMIT ?, (limit,), ).fetchall() return rows这个存储层足够简单但已经能满足“落盘 查询”的需求。生产环境中你可以把_connect替换成连接池或者换用 PostgreSQL。5.4 实现 LLM 调用采集装饰器现在实现最核心的部分在不改动业务代码的前提下自动采集一次 LLM 调用的完整信息。项目路径lemmalog/collector.py。# lemmalog/collector.py import functools import json import threading import time from typing import Any, Callable, Optional from openai import OpenAI from lemmalog.storage import SQLiteStorage from lemmalog.trace_models import TraceRecord, TraceSession # 用线程局部变量保存当前请求的 TraceSession _context_local threading.local() class TraceContextManager: 负责在请求生命周期内管理 TraceSession def __init__(self, storage: SQLiteStorage): self.storage storage def __enter__(self) - TraceSession: session TraceSession( session_idsession_demo_001, request_idreq_demo_001, ) _context_local.session session return session def __exit__(self, exc_type, exc_val, exc_tb) - None: session getattr(_context_local, session, None) if session is not None: session.complete() for record in session.records: self.storage.save_record(record) _context_local.session None def get_current_session() - Optional[TraceSession]: return getattr(_context_local, session, None) def trace_llm_call(func: Callable) - Callable: 装饰器自动记录 LLM 调用前后的数据 functools.wraps(func) def wrapper(*args, **kwargs): session get_current_session() if session is None: # 没有上下文时直接执行不采集 return func(*args, **kwargs) start time.time() model kwargs.get(model, unknown) messages kwargs.get(messages, []) # 尝试从 client.chat.completions.create 的参数中读取消息 # 由于不同 SDK 的调用方式不同这里做一个兼容处理 trace TraceRecord( session_idsession.session_id, request_idsession.request_id, modelmodel, messagesmessages, context_snapshot{ has_session: True, record_count: len(session.records), }, ) try: result func(*args, **kwargs) # 兼容不同 SDK 返回对象 if hasattr(result, model_dump): result_dict result.model_dump() elif hasattr(result, dict): result_dict result.dict() else: result_dict result trace.response json.dumps(result_dict, ensure_asciiFalse, defaultstr) trace.prompt_tokens result_dict.get(usage, {}).get(prompt_tokens, 0) trace.completion_tokens result_dict.get(usage, {}).get(completion_tokens, 0) trace.total_tokens result_dict.get(usage, {}).get(total_tokens, 0) trace.status success except Exception as e: trace.status error trace.error_message str(e) raise finally: trace.latency_ms int((time.time() - start) * 1000) session.add_record(trace) return result return wrapper这个装饰器的核心思路是在调用前后分别采集数据然后封装成TraceRecord挂到当前会话上。业务代码本身不需要感知采集逻辑。5.5 编写一个完整可运行示例现在写一个示例程序模拟一个简单的多轮对话服务。项目路径examples/chat_demo.py。# examples/chat_demo.py import os from openai import OpenAI from lemmalog.collector import TraceContextManager, trace_llm_call from lemmalog.storage import SQLiteStorage # 模拟业务函数这是原本的业务代码 trace_llm_call def chat_once(client: OpenAI, messages: list[dict]) - str: 发送一次聊天请求 response client.chat.completions.create( modelos.getenv(LLM_MODEL, gpt-4o-mini), messagesmessages, temperature0.7, ) return response.choices[0].message.content def build_messages(user_input: str, history: list[dict] | None None) - list[dict]: 构造消息列表 messages [ {role: system, content: 你是一个智能客服助手请使用中文回答用户问题。}, ] if history: messages.extend(history) messages.append({role: user, content: user_input}) return messages def main() - None: storage SQLiteStorage(lemmalog_demo.db) client OpenAI(api_keyos.getenv(OPENAI_API_KEY, sk-demo)) # 模拟历史对话 history [ {role: user, content: 我的订单一直显示运输中怎么办}, {role: assistant, content: 您好请问订单号是多少我可以帮您查询物流信息。}, ] # 使用 TraceContextManager 开启一次业务请求的追踪 with TraceContextManager(storage): messages build_messages(订单号是 20241201帮我看看到哪里了, history) answer chat_once(client, messages) print(AI 回复:, answer) # 查询保存的轨迹数据 print(\n 查询数据库中的轨迹记录 ) rows storage.query_all() for row in rows: print(f会话: {row[session_id]}, 模型: {row[model]}, f耗时: {row[latency_ms]}ms, Token: {row[total_tokens]}, f状态: {row[status]}) if __name__ __main__: main()运行方式python examples/chat_demo.py如果你没有真实 API Key程序会在调用 OpenAI 接口时报错但trace仍然会记录一条error状态的记录。这是验证采集流程是否正常工作的最简单方式。5.6 分析器用程序分析的思路查问题有了采集和存储之后就需要一个分析层。这里提供一个简单的分析器它包含了几类常见的检查规则。项目路径lemmalog/analyzer.py。# lemmalog/analyzer.py import json import sqlite3 from collections import Counter from lemmalog.storage import SQLiteStorage class LemmalogAnalyzer: 基于 SQLite 的轻量分析器 def __init__(self, storage: SQLiteStorage): self.storage storage def analyze_session(self, session_id: str) - dict: 分析单个会话的调用序列 rows self.storage.query_by_session(session_id) if not rows: return {error: session not found} total_tokens sum(row[total_tokens] for row in rows) total_latency sum(row[latency_ms] for row in rows) error_count sum(1 for row in rows if row[status] error) # 检查历史消息是否存在数量突增 msg_counts [] for row in rows: messages json.loads(row[messages]) msg_counts.append(len(messages)) return { session_id: session_id, call_count: len(rows), total_tokens: total_tokens, total_latency_ms: total_latency, error_count: error_count, message_count_sequence: msg_counts, } def find_context_overflow_risk(self, max_messages: int 20) - list[dict]: 找出消息数量超过阈值的调用这类调用容易触发上下文截断 risks [] for row in self.storage.query_all(limit500): messages json.loads(row[messages]) if len(messages) max_messages: risks.append({ trace_id: row[trace_id], session_id: row[session_id], message_count: len(messages), total_tokens: row[total_tokens], }) return risks def analyze_model_usage(self) - dict: 统计不同模型的使用频率和成本趋势 rows self.storage.query_all(limit1000) model_counter Counter() token_counter Counter() for row in rows: model row[model] model_counter[model] 1 token_counter[model] row[total_tokens] return { model_counts: dict(model_counter), model_tokens: dict(token_counter), }这个分析器已经可以回答一些实际问题某个会话内一共发生了多少次 LLM 调用哪些调用存在消息数量超限的风险不同模型的使用频次和 Token 消耗分布如何5.7 运行与验证为了验证整个流程可以写一个 mock 版本不需要真实调用 OpenAI。# examples/run_mock.py from openai import OpenAI from lemmalog.analyzer import LemmalogAnalyzer from lemmalog.collector import TraceContextManager, trace_llm_call from lemmalog.storage import SQLiteStorage trace_llm_call def fake_chat(client: OpenAI, messages: list[dict]) - str: 模拟一次 LLM 调用不真实请求 OpenAI return 这是一个模拟回复 def build_messages(user_input: str, history: list[dict] | None None) - list[dict]: messages [{role: system, content: 你是一个智能客服助手}] if history: messages.extend(history) messages.append({role: user, content: user_input}) return messages def main() - None: storage SQLiteStorage(lemmalog_mock.db) client OpenAI(api_keysk-mock) with TraceContextManager(storage): # 第一次调用没有历史 msg1 build_messages(你好) fake_chat(client, msg1) # 第二次调用带着历史 history [ {role: user, content: 你好}, {role: assistant, content: 您好请问有什么可以帮助您}, ] msg2 build_messages(我想查询订单, history) fake_chat(client, msg2) print( 简单分析结果 ) analyzer LemmalogAnalyzer(storage) # 分析第一个会话 result analyzer.analyze_session(session_demo_001) print(json.dumps(result, ensure_asciiFalse, indent2)) print(\n 调用链明细 ) rows storage.query_all() for i, row in enumerate(rows, 1): print(f\n--- 第 {i} 次调用 ---) print(f模型: {row[model]}) print(f消息数: {len(json.loads(row[messages]))}) print(f响应: {row[response]}) print(f耗时: {row[latency_ms]}ms) print(f状态: {row[status]}) if __name__ __main__: main()运行python examples/run_mock.py预期输出类似 简单分析结果 { session_id: session_demo_001, call_count: 2, total_tokens: 0, total_latency_ms: 0, error_count: 0, message_count_sequence: [3, 5] }从这个输出可以直观看到第二次调用的消息数量比第一次多这正是历史会话注入上下文的表现。如果消息数量增长到远超预期就可能是没有做历史截断。6. 用 Lemmalog 解决实际问题的三个场景上面实现了一个最小闭环下面用三个真实场景说明 Lemmalog 的用法。6.1 场景一定位“上下文被覆盖”问题在 AI Agent 应用中经常需要把工具返回结果写入上下文。但如果你在每一轮中直接向messages列表追加内容而不清理上一轮的临时数据就会出现“上下文污染”。使用 Lemmalog 的find_context_overflow_risk方法可以快速找出消息数量激增的会话risks analyzer.find_context_overflow_risk(max_messages20) for risk in risks: print(f会话 {risk[session_id]} 中调用 {risk[trace_id]} 包含 {risk[message_count]} 条消息)一旦定位到具体调用你就可以反查messages字段的完整内容看看到底是哪一轮循环把临时数据留在了上下文里。6.2 场景二多轮对话行为漂移你可能遇到过这样的问题第一轮模型回答得很准确第三轮就开始“失忆”。使用 Lemmalog 可以对比同一个会话内不同调用的messages结构rows storage.query_by_session(session_demo_001) for i, row in enumerate(rows): messages json.loads(row[messages]) system_prompts [m[content] for m in messages if m[role] system] print(f第 {i1} 次调用 system prompt 数量: {len(system_prompts)})如果第一次调用system_prompts是 1第二次变成了 2说明有模块在向消息列表里重复添加 System Prompt。这是一个典型的程序分析思路检查状态在调用链中的传播路径。6.3 场景三Token 消耗异常分析成本失控是 LLM 应用的常见运维问题。Lemmalog 的analyze_model_usage方法可以帮你统计usage analyzer.analyze_model_usage() print(usage)如果某个模型的 Token 消耗突然暴涨可以继续按session_id分组查询找出 Token 消耗最高的业务请求。进一步分析这些请求的messages序列通常会发现是检索上下文重复注入或者历史消息无限累积导致的。7. 常见问题与排查思路问题现象可能原因解决思路运行时提示module openai has no attribute ChatCompletionopenai SDK 版本过旧升级到 1.x或改用client.chat.completions.create新接口装饰器没有采集到任何记录调用 LLM 时不在TraceContextManager的上下文范围内检查业务代码是否在with块之外调用了chat_once数据库中没有记录但代码没报错存储层保存逻辑被短路检查TraceContextManager.__exit__是否正常执行session 是否为 None查询结果中messages字段是字符串无法直接用存储时用json.dumps序列化读取时没解析读取时执行json.loads(row[messages])多线程请求互相干扰 trace 数据使用了全局变量保存 session没有用线程隔离改用threading.local()或者使用上下文变量contextvars真实 API 调用时耗时很长影响业务采集逻辑在请求链路上同步执行采集与保存解耦先写入内存队列再由后台线程批量落盘如果你在接入 Lemmalog 时遇到“数据不完整”或者“没有记录”建议按以下顺序排查确认调用 LLM 的代码是否经过了trace_llm_call装饰器。确认同一线程内是否创建了TraceContextManager上下文。确认 SQLite 数据库文件路径是否有写权限。打开数据库手动查看llm_trace表是否有数据。sqlite3 lemmalog.db select session_id, model, status from llm_trace limit 10;8. 从 Demo 到生产最佳实践与工程建议8.1 存储选型Demo 中使用 SQLite方便本地调试。生产环境中建议更换为 PostgreSQL原因有三支持并发写入不会被单文件锁阻塞。可以按session_id做分区查询性能更好。可以和业务系统共用数据库权限体系便于运维。8.2 敏感信息脱敏LLM 调用记录中可能包含用户个人信息、业务敏感数据。在设计采集层时必须加入脱敏逻辑。建议在TraceRecord写入存储前对messages和context_snapshot做以下处理移除包含手机号、邮箱、身份证号等正则匹配的字段。System Prompt 中如果包含密钥或内部 API 地址必须过滤后才落盘。对用户 ID 做哈希化处理避免明文存储。import re def mask_sensitive_content(text: str) - str: # 手机号脱敏 text re.sub(r1[3-9]\d{9}, 138****0000, text) # 邮箱脱敏 text re.sub(r\w\w\.\w, userexample.com, text) return text8.3 性能优化LLM 应用的响应时间通常在几百毫秒到几秒一次额外的 JSON 序列化和 SQLite 写入对延迟影响不大。但如果你的调用频率很高比如每秒数十次就需要考虑异步化内存队列采集结果先写入queue.Queue后台线程批量消费写入。采样率控制生产环境可以设定 10% 或 1% 的采样率降低存储压力。历史数据归档超过 30 天的轨迹数据归档到冷存储避免在线库膨胀。8.4 分析规则的可扩展性不要把分析逻辑写死在业务代码里。建议把分析规则拆成独立的 Python 模块每个规则实现一个统一接口class BaseRule: def check(self, trace_session) - list[str]: 返回违规描述列表空列表表示无问题 raise NotImplementedError这样你就可以像写单元测试一样写分析规则配合定时任务做巡检在问题影响用户之前提前发现风险。8.5 合法合规与权限边界LLM 调用轨迹本质上属于系统日志涉及用户行为数据。在采集和存储时必须注意遵循最小化原则只记录排错必需字段不要额外抓取用户聊天内容用于他处。涉及数据库变更和生产环境部署时先在测试环境验证完整流程。分析数据仅供内部排错使用不应用于用户画像分析或外部商业合作。9. 总结与进阶方向Lemmalog 的核心思路可以概括为一句话把 LLM 应用运行时的“记忆”变成可查询、可分析、可复现的结构化数据再用程序分析的方式去定位问题。本文实现了从采集、存储到分析的最小闭环。你已经可以做到记录每次 LLM 调用的完整输入输出。将多次调用关联到同一个业务请求。通过分析器发现上下文超长、Token 消耗异常等问题。在真实事故发生后通过session_id复盘完整调用轨迹。下一步可以往这些方向继续深入接入真实的大模型链路把装饰器从chat.completions.create扩展到 Embeding、Function Calling 等场景。加入流式响应分析流式接口返回的是 Generator需要单独处理响应聚合。搭建可视化界面用 FastAPI 前端表格展示 session 调用链。引入静态分析在部署前扫描代码库找出哪些 Prompt 模板没有被trace_llm_call覆盖形成 LLM 应用的“代码覆盖率”。在动手改进之前先做一件事把你当前项目的 LLM 调用入口列出来然后手动在入口包一层日志记录。你可能会发现很多平时说不清楚的“玄学问题”其实只是上下文数据流的程序问题。如果这篇文章对你有帮助可以先收藏备用。后续我会继续整理 Lemmalog 在 Agent 场景下的实践以及如何把分析结果对接告警系统。 LLM应用调试不再靠猜：用Lemmalog把记忆变成可分析数据
1. 从一次线上事故说起LLM 应用为什么这么难调试先讲一个真实场景。某天晚上一个基于大语言模型LLM的工单分类服务突然出问题用户提交的工单被分到了完全无关的类别。第一反应是看日志结果发现日志里只记录了“调用成功返回结果xxx”完全没有记录当时传进去的上下文窗口里到底有什么内容。你想复现这个 Bug但用同样的 Prompt 测了几次结果又都是对的。这就是 LLM 应用开发最大的痛点之一结果非确定性。同样的输入温度参数不为 0 时模型可能给出不同答案上下文窗口里的历史消息、系统提示词、工具返回结果之间一旦互相干扰问题就更难定位。传统软件开发里我们可以靠日志、断点、单元测试、覆盖率来逼近问题根因。但到了 LLM 应用里程序的行为有一半发生在“模型内部”这一部分是黑盒我们无法直接观测。不过还有一半行为是我们可以观测的那就是模型调用前后的数据流——你发给模型什么、模型返回什么、中间经过了多少轮上下文截断、每轮消耗了多少 Token、哪个环节改变了消息结构。把这些可观测的数据完整记录下来再像“程序分析”一样去检查它们就能让 LLM 应用的调试从“猜”变成“查”。这就是 Lemmalog 的出发点把 LLM 的记忆上下文、状态、调用轨迹变成可查询、可分析的程序数据。这篇文章会围绕 Lemmalog 的设计思路讲三件事为什么普通日志系统不足以支撑 LLM 应用排错。Lemmalog 如何设计“LLM 记忆采集 结构化存储 程序分析”三层结构。一个最小可运行的 Python 实现帮你理解核心原理。无论你是在做 AI Agent、RAG 应用还是想给团队搭一套 LLM 可观测性基础工具这篇内容都能给你一些可落地的思路。2. 核心概念LLM 记忆、程序分析与 Lemmalog 的定位2.1 什么是“LLM 记忆”在传统程序里程序的“状态”保存在变量、数据库、缓存中。而 LLM 应用的“状态”本质上由三部分组成上下文窗口Context Window当前请求的所有消息包括 system prompt、历史对话、工具返回值、检索到的文档片段。会话状态Session State跨请求保留的变量比如用户 ID、会话 ID、当前步骤、临时选择的结构化数据。外部记忆External Memory向量数据库、KV 缓存、普通数据库中的内容它们会在需要时被注入到上下文窗口里。为什么“记忆”这么重要因为在 LLM 应用里模型的输出完全由输入决定而输入中真正不可控的往往就是记忆部分。上下文被截断、历史消息顺序错乱、检索结果重复注入这些问题几乎不会在代码层面报错但会导致模型行为异常。Lemmalog 要做的第一件事就是把上面这三类记忆“快照”下来让它们和每一次模型调用关联起来。2.2 什么是“程序分析”程序分析Program Analysis并不是一个新概念。传统的程序分析包括静态分析不运行代码直接扫描源码找问题和动态分析运行代码跟踪变量、路径、性能。它关注的问题是这段程序里是否存在某种状态异常某个变量在哪里被赋值在哪里被读取是否可能为 null某些条件分支是否永远无法到达当我们把 LLM 应用的每次调用想象成一次“函数执行”把上下文窗口想象成“入参”把模型输出想象成“返回值”程序分析的方法论就可以平移过来调用轨迹一次用户请求到底触发了几次 LLM 调用数据流哪些系统变量进入了 Prompt它们是如何被修改的状态检查某次调用的 Token 数是否异常历史消息是否被错误截断一致性分析多轮对话中同一类请求的 Prompt 结构是否保持稳定Lemmalog 这个名字就是“Lemma引理 Log日志”的合成词。它想表达的是一种态度用证明引理的严谨性去记录每一次模型交互。2.3 Lemmalog 与现有方案的区别市面上其实已经有 LangSmith、Langfuse 这类 LLM 可观测平台。它们的核心能力是追踪Tracing、监控Monitoring和可视化Visualization适合在团队内快速搭建可观测性。Lemmalog 的定位不太一样。它更强调“分析”而不是“展示”。它会把采集到的记忆数据落成结构化的关系型数据让你可以用 SQL 或者简单脚本去执行一些程序分析任务。比如找出所有“在同一会话内第二次调用时历史消息超过 50 条”的请求。分析某种工具返回结果被注入 System Prompt 后模型输出格式是否发生变化。统计不同 Prompt 模板的 Token 消耗中位数定位成本异常。换句话说Lemmalog 不是一个完整平台而是一种设计模式和轻量级落地框架。它强调你不需要一开始就上重平台你可以先在自己的代码里实现一条“记忆 → 日志 → 分析”的流水线后续再扩展成完整服务。3. 环境准备与工程结构3.1 环境版本本文示例代码使用 Python 编写核心运行环境如下依赖项版本建议说明Python3.10使用了新版类型注解语法 Xopenai按你实际使用的版本本文用官方 SDK 做示例重点不在具体模型pydantic2.x用于配置解析和数据校验SQLite3Python 内置零依赖存储后端方便本地运行不同版本的依赖库在 API 上可能存在差异。如果你使用的 OpenAI SDK 版本较早方法签名可能是openai.ChatCompletion.create(...)而不是client.chat.completions.create(...)。读者需要根据自己的实际环境调整。3.2 示例项目结构lemmalog-demo/ ├── lemmalog/ │ ├── __init__.py │ ├── trace_context.py # 调用链上下文 │ ├── storage.py # SQLite 存储层 │ ├── analyzer.py # 分析器查询 检查规则 │ └── collector.py # LLM 调用采集装饰器 ├── examples/ │ └── chat_demo.py # 完整可运行的示例 └── requirements.txt这个结构刻意保持精简。实际项目中你可以把 storage 层替换成 PostgreSQL把 analyzer 层替换成定时任务或事件回调。3.3 设计目标在动手写代码前先明确 Lemmalog 的三个设计原则非侵入性业务代码不需要大规模改造。理想状态下只需要在 LLM 调用处加一行装饰器或者在请求入口设置一个上下文。持久化优先所有记忆数据必须落盘不能只存在内存里或 Redis 里。因为故障分析往往发生在事故之后。分析友好存储格式要方便后续做切片、聚合、关联查询。因此使用关系型结构而不是纯文本日志。4. 核心原理拆解一条 LLM 调用是如何被“分析化”的4.1 从调用到轨迹一次最简单的 LLM 调用在程序层面可以看成response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, )要分析这次调用我们需要采集以下数据数据项示例值用途会话 IDsession_123关联多轮对话请求 IDreq_456唯一标识一次请求模型名称gpt-4o-mini迭代对比不同模型行为请求消息序列化后的 messages 数组重放现场检查 Prompt 结构响应结果完整响应内容检查输出质量Token 使用输入 1024输出 512成本分析与截断检测耗时1234 ms性能监控上下文快照当时的会话状态、检索到的文档片段排查历史消息干扰Lemmalog 的采集层就是围绕这些字段设计的。4.2 从轨迹到分析采集到数据只是第一步。关键是“分析”。我们可以定义三种分析粒度第一层单次调用分析检查一次调用中是否存在异常。例如输入 Token 数是否超过模型最大限制输出是否被截断响应是否因为max_tokens设置太小而中断第二层调用序列分析检查一个会话内的多次调用是否连贯。例如第二轮调用是否携带了第一轮的完整历史工具调用返回的临时数据有没有在下一次 Prompt 中被错误保留第三层跨会话统计分析把大量会话放在一起对比。例如某个 Prompt 模板在历史消息长度超过阈值之后模型输出是否出现了格式漂移Lemmalog 的分析器就是围绕这三个层级去设计查询接口的。4.3 核心抽象TraceContext在代码实现中我们使用一个TraceContext对象来管理记忆快照。它的职责是在请求开始时创建。在整个调用链中传递通过参数或者上下文变量。每次 LLM 调用都会向当前 TraceContext 追加一条 Trace 记录。在请求结束时把整个 Trace 批量写入存储层。下面进入代码实现部分。5. Lemmalog 最小实现实战5.1 创建项目与依赖先创建虚拟环境并安装依赖。mkdir lemmalog-demo cd lemmalog-demo python3 -m venv .venv source .venv/bin/activate pip install openai pydantic如果你没有 OpenAI API Key也可以把核心代码跑通只需要在调用采集部分用 mock 数据代替真实调用。5.2 定义 Trace 数据结构首先定义数据类。项目路径lemmalog/trace_models.py。# lemmalog/trace_models.py from __future__ import annotations import time import uuid from typing import Any, Optional from pydantic import BaseModel, Field class TraceRecord(BaseModel): 单次 LLM 调用轨迹记录 trace_id: str Field(default_factorylambda: uuid.uuid4().hex) session_id: str request_id: str model: str messages: list[dict[str, Any]] response: Optional[str] None prompt_tokens: int 0 completion_tokens: int 0 total_tokens: int 0 latency_ms: int 0 context_snapshot: dict[str, Any] Field(default_factorydict) timestamp: float Field(default_factorytime.time) status: str success error_message: Optional[str] None class TraceSession(BaseModel): 一个业务请求对应的完整跟踪会话 session_id: str request_id: str start_time: float Field(default_factorytime.time) end_time: Optional[float] None records: list[TraceRecord] Field(default_factorylist) def add_record(self, record: TraceRecord) - None: self.records.append(record) def complete(self) - None: self.end_time time.time()这里的核心是TraceRecord。可以看到它不仅记录了常见的输入输出还记录了context_snapshot这用来保存调用发生时外部记忆的内容。5.3 实现存储层使用 SQLite 作为存储。这个模块负责建表、插入记录、提供基础查询接口。项目路径lemmalog/storage.py。# lemmalog/storage.py import json import sqlite3 from pathlib import Path from typing import Any from lemmalog.trace_models import TraceRecord, TraceSession class SQLiteStorage: SQLite 存储实现 def __init__(self, db_path: str lemmalog.db): self.db_path db_path self._init_db() def _init_db(self) - None: with self._connect() as conn: conn.execute( CREATE TABLE IF NOT EXISTS llm_trace ( trace_id TEXT PRIMARY KEY, session_id TEXT NOT NULL, request_id TEXT NOT NULL, model TEXT NOT NULL, messages TEXT NOT NULL, response TEXT, prompt_tokens INTEGER, completion_tokens INTEGER, total_tokens INTEGER, latency_ms INTEGER, context_snapshot TEXT, timestamp REAL, status TEXT, error_message TEXT ) ) conn.execute( CREATE INDEX IF NOT EXISTS idx_session_id ON llm_trace(session_id) ) conn.execute( CREATE INDEX IF NOT EXISTS idx_request_id ON llm_trace(request_id) ) conn.commit() def _connect(self) - sqlite3.Connection: conn sqlite3.connect(self.db_path) conn.row_factory sqlite3.Row return conn def save_record(self, record: TraceRecord) - None: with self._connect() as conn: conn.execute( INSERT INTO llm_trace ( trace_id, session_id, request_id, model, messages, response, prompt_tokens, completion_tokens, total_tokens, latency_ms, context_snapshot, timestamp, status, error_message ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( record.trace_id, record.session_id, record.request_id, record.model, json.dumps(record.messages, ensure_asciiFalse), record.response, record.prompt_tokens, record.completion_tokens, record.total_tokens, record.latency_ms, json.dumps(record.context_snapshot, ensure_asciiFalse), record.timestamp, record.status, record.error_message, ), ) conn.commit() def query_by_session(self, session_id: str) - list[sqlite3.Row]: with self._connect() as conn: rows conn.execute( SELECT * FROM llm_trace WHERE session_id ? ORDER BY timestamp ASC, (session_id,), ).fetchall() return rows def query_all(self, limit: int 100) - list[sqlite3.Row]: with self._connect() as conn: rows conn.execute( SELECT * FROM llm_trace ORDER BY timestamp DESC LIMIT ?, (limit,), ).fetchall() return rows这个存储层足够简单但已经能满足“落盘 查询”的需求。生产环境中你可以把_connect替换成连接池或者换用 PostgreSQL。5.4 实现 LLM 调用采集装饰器现在实现最核心的部分在不改动业务代码的前提下自动采集一次 LLM 调用的完整信息。项目路径lemmalog/collector.py。# lemmalog/collector.py import functools import json import threading import time from typing import Any, Callable, Optional from openai import OpenAI from lemmalog.storage import SQLiteStorage from lemmalog.trace_models import TraceRecord, TraceSession # 用线程局部变量保存当前请求的 TraceSession _context_local threading.local() class TraceContextManager: 负责在请求生命周期内管理 TraceSession def __init__(self, storage: SQLiteStorage): self.storage storage def __enter__(self) - TraceSession: session TraceSession( session_idsession_demo_001, request_idreq_demo_001, ) _context_local.session session return session def __exit__(self, exc_type, exc_val, exc_tb) - None: session getattr(_context_local, session, None) if session is not None: session.complete() for record in session.records: self.storage.save_record(record) _context_local.session None def get_current_session() - Optional[TraceSession]: return getattr(_context_local, session, None) def trace_llm_call(func: Callable) - Callable: 装饰器自动记录 LLM 调用前后的数据 functools.wraps(func) def wrapper(*args, **kwargs): session get_current_session() if session is None: # 没有上下文时直接执行不采集 return func(*args, **kwargs) start time.time() model kwargs.get(model, unknown) messages kwargs.get(messages, []) # 尝试从 client.chat.completions.create 的参数中读取消息 # 由于不同 SDK 的调用方式不同这里做一个兼容处理 trace TraceRecord( session_idsession.session_id, request_idsession.request_id, modelmodel, messagesmessages, context_snapshot{ has_session: True, record_count: len(session.records), }, ) try: result func(*args, **kwargs) # 兼容不同 SDK 返回对象 if hasattr(result, model_dump): result_dict result.model_dump() elif hasattr(result, dict): result_dict result.dict() else: result_dict result trace.response json.dumps(result_dict, ensure_asciiFalse, defaultstr) trace.prompt_tokens result_dict.get(usage, {}).get(prompt_tokens, 0) trace.completion_tokens result_dict.get(usage, {}).get(completion_tokens, 0) trace.total_tokens result_dict.get(usage, {}).get(total_tokens, 0) trace.status success except Exception as e: trace.status error trace.error_message str(e) raise finally: trace.latency_ms int((time.time() - start) * 1000) session.add_record(trace) return result return wrapper这个装饰器的核心思路是在调用前后分别采集数据然后封装成TraceRecord挂到当前会话上。业务代码本身不需要感知采集逻辑。5.5 编写一个完整可运行示例现在写一个示例程序模拟一个简单的多轮对话服务。项目路径examples/chat_demo.py。# examples/chat_demo.py import os from openai import OpenAI from lemmalog.collector import TraceContextManager, trace_llm_call from lemmalog.storage import SQLiteStorage # 模拟业务函数这是原本的业务代码 trace_llm_call def chat_once(client: OpenAI, messages: list[dict]) - str: 发送一次聊天请求 response client.chat.completions.create( modelos.getenv(LLM_MODEL, gpt-4o-mini), messagesmessages, temperature0.7, ) return response.choices[0].message.content def build_messages(user_input: str, history: list[dict] | None None) - list[dict]: 构造消息列表 messages [ {role: system, content: 你是一个智能客服助手请使用中文回答用户问题。}, ] if history: messages.extend(history) messages.append({role: user, content: user_input}) return messages def main() - None: storage SQLiteStorage(lemmalog_demo.db) client OpenAI(api_keyos.getenv(OPENAI_API_KEY, sk-demo)) # 模拟历史对话 history [ {role: user, content: 我的订单一直显示运输中怎么办}, {role: assistant, content: 您好请问订单号是多少我可以帮您查询物流信息。}, ] # 使用 TraceContextManager 开启一次业务请求的追踪 with TraceContextManager(storage): messages build_messages(订单号是 20241201帮我看看到哪里了, history) answer chat_once(client, messages) print(AI 回复:, answer) # 查询保存的轨迹数据 print(\n 查询数据库中的轨迹记录 ) rows storage.query_all() for row in rows: print(f会话: {row[session_id]}, 模型: {row[model]}, f耗时: {row[latency_ms]}ms, Token: {row[total_tokens]}, f状态: {row[status]}) if __name__ __main__: main()运行方式python examples/chat_demo.py如果你没有真实 API Key程序会在调用 OpenAI 接口时报错但trace仍然会记录一条error状态的记录。这是验证采集流程是否正常工作的最简单方式。5.6 分析器用程序分析的思路查问题有了采集和存储之后就需要一个分析层。这里提供一个简单的分析器它包含了几类常见的检查规则。项目路径lemmalog/analyzer.py。# lemmalog/analyzer.py import json import sqlite3 from collections import Counter from lemmalog.storage import SQLiteStorage class LemmalogAnalyzer: 基于 SQLite 的轻量分析器 def __init__(self, storage: SQLiteStorage): self.storage storage def analyze_session(self, session_id: str) - dict: 分析单个会话的调用序列 rows self.storage.query_by_session(session_id) if not rows: return {error: session not found} total_tokens sum(row[total_tokens] for row in rows) total_latency sum(row[latency_ms] for row in rows) error_count sum(1 for row in rows if row[status] error) # 检查历史消息是否存在数量突增 msg_counts [] for row in rows: messages json.loads(row[messages]) msg_counts.append(len(messages)) return { session_id: session_id, call_count: len(rows), total_tokens: total_tokens, total_latency_ms: total_latency, error_count: error_count, message_count_sequence: msg_counts, } def find_context_overflow_risk(self, max_messages: int 20) - list[dict]: 找出消息数量超过阈值的调用这类调用容易触发上下文截断 risks [] for row in self.storage.query_all(limit500): messages json.loads(row[messages]) if len(messages) max_messages: risks.append({ trace_id: row[trace_id], session_id: row[session_id], message_count: len(messages), total_tokens: row[total_tokens], }) return risks def analyze_model_usage(self) - dict: 统计不同模型的使用频率和成本趋势 rows self.storage.query_all(limit1000) model_counter Counter() token_counter Counter() for row in rows: model row[model] model_counter[model] 1 token_counter[model] row[total_tokens] return { model_counts: dict(model_counter), model_tokens: dict(token_counter), }这个分析器已经可以回答一些实际问题某个会话内一共发生了多少次 LLM 调用哪些调用存在消息数量超限的风险不同模型的使用频次和 Token 消耗分布如何5.7 运行与验证为了验证整个流程可以写一个 mock 版本不需要真实调用 OpenAI。# examples/run_mock.py from openai import OpenAI from lemmalog.analyzer import LemmalogAnalyzer from lemmalog.collector import TraceContextManager, trace_llm_call from lemmalog.storage import SQLiteStorage trace_llm_call def fake_chat(client: OpenAI, messages: list[dict]) - str: 模拟一次 LLM 调用不真实请求 OpenAI return 这是一个模拟回复 def build_messages(user_input: str, history: list[dict] | None None) - list[dict]: messages [{role: system, content: 你是一个智能客服助手}] if history: messages.extend(history) messages.append({role: user, content: user_input}) return messages def main() - None: storage SQLiteStorage(lemmalog_mock.db) client OpenAI(api_keysk-mock) with TraceContextManager(storage): # 第一次调用没有历史 msg1 build_messages(你好) fake_chat(client, msg1) # 第二次调用带着历史 history [ {role: user, content: 你好}, {role: assistant, content: 您好请问有什么可以帮助您}, ] msg2 build_messages(我想查询订单, history) fake_chat(client, msg2) print( 简单分析结果 ) analyzer LemmalogAnalyzer(storage) # 分析第一个会话 result analyzer.analyze_session(session_demo_001) print(json.dumps(result, ensure_asciiFalse, indent2)) print(\n 调用链明细 ) rows storage.query_all() for i, row in enumerate(rows, 1): print(f\n--- 第 {i} 次调用 ---) print(f模型: {row[model]}) print(f消息数: {len(json.loads(row[messages]))}) print(f响应: {row[response]}) print(f耗时: {row[latency_ms]}ms) print(f状态: {row[status]}) if __name__ __main__: main()运行python examples/run_mock.py预期输出类似 简单分析结果 { session_id: session_demo_001, call_count: 2, total_tokens: 0, total_latency_ms: 0, error_count: 0, message_count_sequence: [3, 5] }从这个输出可以直观看到第二次调用的消息数量比第一次多这正是历史会话注入上下文的表现。如果消息数量增长到远超预期就可能是没有做历史截断。6. 用 Lemmalog 解决实际问题的三个场景上面实现了一个最小闭环下面用三个真实场景说明 Lemmalog 的用法。6.1 场景一定位“上下文被覆盖”问题在 AI Agent 应用中经常需要把工具返回结果写入上下文。但如果你在每一轮中直接向messages列表追加内容而不清理上一轮的临时数据就会出现“上下文污染”。使用 Lemmalog 的find_context_overflow_risk方法可以快速找出消息数量激增的会话risks analyzer.find_context_overflow_risk(max_messages20) for risk in risks: print(f会话 {risk[session_id]} 中调用 {risk[trace_id]} 包含 {risk[message_count]} 条消息)一旦定位到具体调用你就可以反查messages字段的完整内容看看到底是哪一轮循环把临时数据留在了上下文里。6.2 场景二多轮对话行为漂移你可能遇到过这样的问题第一轮模型回答得很准确第三轮就开始“失忆”。使用 Lemmalog 可以对比同一个会话内不同调用的messages结构rows storage.query_by_session(session_demo_001) for i, row in enumerate(rows): messages json.loads(row[messages]) system_prompts [m[content] for m in messages if m[role] system] print(f第 {i1} 次调用 system prompt 数量: {len(system_prompts)})如果第一次调用system_prompts是 1第二次变成了 2说明有模块在向消息列表里重复添加 System Prompt。这是一个典型的程序分析思路检查状态在调用链中的传播路径。6.3 场景三Token 消耗异常分析成本失控是 LLM 应用的常见运维问题。Lemmalog 的analyze_model_usage方法可以帮你统计usage analyzer.analyze_model_usage() print(usage)如果某个模型的 Token 消耗突然暴涨可以继续按session_id分组查询找出 Token 消耗最高的业务请求。进一步分析这些请求的messages序列通常会发现是检索上下文重复注入或者历史消息无限累积导致的。7. 常见问题与排查思路问题现象可能原因解决思路运行时提示module openai has no attribute ChatCompletionopenai SDK 版本过旧升级到 1.x或改用client.chat.completions.create新接口装饰器没有采集到任何记录调用 LLM 时不在TraceContextManager的上下文范围内检查业务代码是否在with块之外调用了chat_once数据库中没有记录但代码没报错存储层保存逻辑被短路检查TraceContextManager.__exit__是否正常执行session 是否为 None查询结果中messages字段是字符串无法直接用存储时用json.dumps序列化读取时没解析读取时执行json.loads(row[messages])多线程请求互相干扰 trace 数据使用了全局变量保存 session没有用线程隔离改用threading.local()或者使用上下文变量contextvars真实 API 调用时耗时很长影响业务采集逻辑在请求链路上同步执行采集与保存解耦先写入内存队列再由后台线程批量落盘如果你在接入 Lemmalog 时遇到“数据不完整”或者“没有记录”建议按以下顺序排查确认调用 LLM 的代码是否经过了trace_llm_call装饰器。确认同一线程内是否创建了TraceContextManager上下文。确认 SQLite 数据库文件路径是否有写权限。打开数据库手动查看llm_trace表是否有数据。sqlite3 lemmalog.db select session_id, model, status from llm_trace limit 10;8. 从 Demo 到生产最佳实践与工程建议8.1 存储选型Demo 中使用 SQLite方便本地调试。生产环境中建议更换为 PostgreSQL原因有三支持并发写入不会被单文件锁阻塞。可以按session_id做分区查询性能更好。可以和业务系统共用数据库权限体系便于运维。8.2 敏感信息脱敏LLM 调用记录中可能包含用户个人信息、业务敏感数据。在设计采集层时必须加入脱敏逻辑。建议在TraceRecord写入存储前对messages和context_snapshot做以下处理移除包含手机号、邮箱、身份证号等正则匹配的字段。System Prompt 中如果包含密钥或内部 API 地址必须过滤后才落盘。对用户 ID 做哈希化处理避免明文存储。import re def mask_sensitive_content(text: str) - str: # 手机号脱敏 text re.sub(r1[3-9]\d{9}, 138****0000, text) # 邮箱脱敏 text re.sub(r\w\w\.\w, userexample.com, text) return text8.3 性能优化LLM 应用的响应时间通常在几百毫秒到几秒一次额外的 JSON 序列化和 SQLite 写入对延迟影响不大。但如果你的调用频率很高比如每秒数十次就需要考虑异步化内存队列采集结果先写入queue.Queue后台线程批量消费写入。采样率控制生产环境可以设定 10% 或 1% 的采样率降低存储压力。历史数据归档超过 30 天的轨迹数据归档到冷存储避免在线库膨胀。8.4 分析规则的可扩展性不要把分析逻辑写死在业务代码里。建议把分析规则拆成独立的 Python 模块每个规则实现一个统一接口class BaseRule: def check(self, trace_session) - list[str]: 返回违规描述列表空列表表示无问题 raise NotImplementedError这样你就可以像写单元测试一样写分析规则配合定时任务做巡检在问题影响用户之前提前发现风险。8.5 合法合规与权限边界LLM 调用轨迹本质上属于系统日志涉及用户行为数据。在采集和存储时必须注意遵循最小化原则只记录排错必需字段不要额外抓取用户聊天内容用于他处。涉及数据库变更和生产环境部署时先在测试环境验证完整流程。分析数据仅供内部排错使用不应用于用户画像分析或外部商业合作。9. 总结与进阶方向Lemmalog 的核心思路可以概括为一句话把 LLM 应用运行时的“记忆”变成可查询、可分析、可复现的结构化数据再用程序分析的方式去定位问题。本文实现了从采集、存储到分析的最小闭环。你已经可以做到记录每次 LLM 调用的完整输入输出。将多次调用关联到同一个业务请求。通过分析器发现上下文超长、Token 消耗异常等问题。在真实事故发生后通过session_id复盘完整调用轨迹。下一步可以往这些方向继续深入接入真实的大模型链路把装饰器从chat.completions.create扩展到 Embeding、Function Calling 等场景。加入流式响应分析流式接口返回的是 Generator需要单独处理响应聚合。搭建可视化界面用 FastAPI 前端表格展示 session 调用链。引入静态分析在部署前扫描代码库找出哪些 Prompt 模板没有被trace_llm_call覆盖形成 LLM 应用的“代码覆盖率”。在动手改进之前先做一件事把你当前项目的 LLM 调用入口列出来然后手动在入口包一层日志记录。你可能会发现很多平时说不清楚的“玄学问题”其实只是上下文数据流的程序问题。如果这篇文章对你有帮助可以先收藏备用。后续我会继续整理 Lemmalog 在 Agent 场景下的实践以及如何把分析结果对接告警系统。