记住这个场景:你花了两周调通的 Agent 工作流,某天早上打开 IDE,发现代码文件被覆盖了,git 历史也被误操作重置。最后能证明“当时系统到底发生了什么”的,只剩下日志目录里那几个不断增长的 .log 文件。
这不是段子。在 Agent 开发里,代码丢失、状态漂移、LLM 输出覆盖已有逻辑,几乎每天都在发生。Agent 应用和普通后端应用有一个本质差异:普通应用里代码是稳定的“真相”,日志是辅助排错工具;在 Agent 应用里,代码本身可能是动态生成的、被模型反复改写、被运行时替换,日志反而是唯一可靠的事实来源。代码删光了不可怕,可怕的是日志也断了。
这篇文章围绕一个核心问题展开:如何在 Agent 项目里设计一套“代码删光也不慌”的日志体系。你会看到为什么 Agent 比普通服务更依赖可观测性,日志应该记录哪些层次,怎么用结构化日志和链路追踪回溯一次完整的 Agent 执行,以及当代码真的丢失时,怎么靠日志把状态找回来。
1. 这篇文章真正要解决的问题
先说清楚边界:这篇文章不是讲 LLM 推理优化,也不是讲某个特定 Agent 框架的 API 用法,而是讲 Agent 工程里的可观测性设计。
为什么这个问题值得单独写一篇?因为 Agent 应用和传统应用在故障域上有根本不同:
- 传统后端:代码固定,逻辑确定,故障来自外部输入或资源问题。日志的价值是定位“哪一行代码出了问题”。
- Agent 应用:代码可能是模型生成的、可能是工具临时拼装的、可能是多个自治子 Agent 互相调用的产物。故障经常是“系统做了什么根本不在预期里”。
这意味着三个结果:
第一,Agent 的“状态”散布在代码、环境变量、向量库里,代码被删或覆盖后,你的第一反应应该是去查日志,而不是去猜代码。
第二,Agent 执行链路的中间产物往往比最终结果更值钱。你看到的最终回复可能只是一句简单总结,但中间检索了什么、调用了哪几个工具、给模型喂了哪些上下文——这些全在日志里。
第三,通用日志库能解决“记录”问题,但解决不了“串起来”的问题。Agent 一次任务会横跨多轮 LLM 调用、多次工具调用、多次代码执行,没有 trace_id 贯通,日志就是一盘散沙。
所以这篇文章的读者画像是:
- 正在用 LangChain、CrewAI、自研 Agent 框架搭建业务的同学;
- 需要排查“Agent 为什么答非所问”“工具调用为什么失败”的开发者;
- 团队里 Agent 代码经常被自动改写,需要引入审计机制的工程负责人。
读完这篇文章,你能带走一套可以直接落地的 Agent 日志方案:包含结构化日志配置、trace_id 贯通、文件持久化、日志分析思路,以及一份常见问题的排查清单。
2. “代码是生成物,日志才是现场”的 Agent 可观测观念
2.1 为什么代码在 Agent 项目里不是唯一真相
传统软件开发有一个默认假设:源码是对的,运行时不符就修代码。但在某些 Agent 工作流里,这个假设不成立了。
典型场景是“自我修正型 Agent”:系统拿到任务后,先生成一段代码,执行后把报错信息喂给模型,模型再改代码,再执行。你原本手写的入口函数可能还在,但核心处理逻辑已经变成模型在运行时反复改出来的东西。这时候源码和实际跑的代码已经脱节。
另一种场景是工具链替换。Agent 框架在初始化时会根据配置动态加载工具类,有时候同一个函数名会被不同分支注册多次,代码仓库里看到的是一个版本,运行容器里加载的是另一个版本。如果你没有记录“运行时实际加载了哪些模块”,光看源码是永远对不上现场情况的。
这里需要建立一种观念:代码只是 Agent 系统的一次快照,日志才是发生在时间长河里的完整证据链。排查 Agent 问题时,源码用于确认“写的时候是什么意图”,日志才是确认“跑的时候发生了什么”。
2.2 Agent 日志的五个关键层次
Agent 的日志体系可以分成五层,每一层的价值不一样:
| 日志层次 | 记录内容 | 典型问题 | 关键字段 |
|---|---|---|---|
| 请求层 | 用户输入、会话 ID、任务参数 | 用户到底问了什么 | user_input、session_id、trace_id |
| 决策层 | LLM 输入输出、选择的动作 | 模型为什么这么决策 | prompt、response、model_name、tokens |
| 工具层 | 工具调用参数与返回结果 | 工具为什么失败 | tool_name、tool_args、tool_result |
| 代码执行层 | 生成的代码、执行结果、报错堆栈 | 代码为什么跑飞 | exec_code、stdout、stderr、exit_code |
| 状态层 | 内存键值、向量库写入、缓存变更 | 状态为什么漂移 | state_key、state_before、state_after |
传统后端日志通常只关心请求层和异常堆栈,Agent 必须五层都记录。因为 Agent 故障很少是单点崩溃,而是链路偏离:请求没问题,模型决策偏了,工具结果不对,状态没有同步,最后在某个环节爆出来。
2.3 trace_id 是贯穿所有层的线索
有了分层还不够,还得把一次任务的所有日志串起来。这就要引入 trace_id。
trace_id 是一个随任务生成、贯穿全链路的唯一标识。一次 Agent 任务从用户请求进来开始,生成一个 trace_id,后续所有环节——模型调用、工具执行、代码运行、状态变更——都把这个 ID 写进日志。排查时你只需要拿 trace_id 去日志中心搜一次,整条链路就完整还原了。
实现上有两种方式:
- 上下文传递:用一个自定义上下文对象把 trace_id 传遍所有函数。
- 线程局部/协程局部存储:用
contextvars或threading.local保存 trace_id,日志格式化器自动读取。
第二种方式更干净,业务代码不用到处手动传参。下面实操部分采用这种方式。
3. 环境准备与最小工程结构
先说环境要求。下面的示例基于 Python 3.9+,只用标准库实现,不依赖任何第三方 Agent 框架。这样设计的目的是让你把所有注意力放在日志机制本身,不被框架 API 干扰。你后续完全可以把这个机制移植到 LangChain、CrewAI 或自研框架中。
需要准备的工具:
- Python 3.9+
- 一个文本编辑器或 IDE
- 终端(Windows 推荐 PowerShell,macOS/Linux 直接 bash)
工程结构如下:
agent_log_demo/ ├── requirements.txt # 依赖声明 ├── main.py # 启动入口 ├── agent/ │ ├── __init__.py │ ├── logger.py # 日志配置与 trace_id 管理 │ ├── executor.py # 模拟 Agent 执行器 │ └── tools.py # 模拟工具调用 └── logs/ # 日志输出目录(运行后自动生成)这个结构刻意保持精简:一个入口文件,一个日志模块,一个模拟执行器,一个模拟工具模块。重点看 logger.py 的设计,其他模块都是用来演示日志链路的。
4. 核心流程拆解
一个完整的 Agent 任务日志体系,按照下面的流程设计:
用户请求 | v 生成 trace_id -------------------------> 写入 请求层日志 | v 构建并发送 LLM 请求 -------------------> 写入 决策层日志 | v 执行工具调用 --------------------------> 写入 工具层日志 | v 执行生成的代码 ------------------------> 写入 代码执行层日志 | v 更新上下文/状态 -----------------------> 写入 状态层日志 | v 返回最终结果 --------------------------> 写入 最终响应日志每一步的关键点:
- 生成 trace_id 要早:在进入任何业务逻辑之前生成,最好在请求入口处完成。
- LLM 请求和响应分开记录:prompt 可能很长,响应也可能很长,合并成一条日志会把单行长度撑爆,不方便检索。
- 工具调用记录参数和返回:工具调用的入参要记录,返回结果要截断后记录,后续排查“模型为什么乱回答”全看这里。
- 代码执行必须记录 exit_code:只记录 stdout/stderr 不够,exit_code 能告诉你进程到底成功没有。
- 状态变更记录 before/after:这是 Agent 项目最容易漏的日志。状态一旦错了,要能直接从日志里看出是哪一步把它改坏的。
下面按这个流程逐个实现。
5. 完整示例代码实现
5.1 第一步:结构化日志配置
文件路径:agent/logger.py
""" Agent 项目日志模块 设计目标: 1. 输出 JSON 格式结构化日志,方便后续采集和分析 2. 支持 thread-local / contextvars 级别的 trace_id 自动注入 3. 日志同时输出到控制台和文件,文件按大小轮转 """ import json import logging import os import socket import sys import uuid from contextvars import ContextVar from logging.handlers import RotatingFileHandler # 用 ContextVar 保存当前任务的 trace_id # Agent 应用中一次请求可能发生在异步上下文里,ContextVar 比 threading.local 更安全 current_trace_id: ContextVar[str] = ContextVar("current_trace_id", default="-") # 用 ContextVar 保存当前任务的 session_id current_session_id: ContextVar[str] = ContextVar("current_session_id", default="-") class JsonFormatter(logging.Formatter): """将日志记录格式化为 JSON 输出,自动注入 trace_id 和 session_id""" def format(self, record: logging.LogRecord) -> str: log_data = { "timestamp": self.formatTime(record, "%Y-%m-%d %H:%M:%S"), "level": record.levelname, "logger": record.name, "module": record.module, "function": record.funcName, "line": record.lineno, "trace_id": current_trace_id.get(), "session_id": current_session_id.get(), "message": record.getMessage(), } # 如果业务代码通过 extra 传了其他字段,合并进来 extra_fields = getattr(record, "extra_fields", None) if extra_fields: log_data.update(extra_fields) # 异常堆栈单独序列化 if record.exc_info: log_data["exc_info"] = self.formatException(record.exc_info) return json.dumps(log_data, ensure_ascii=False) def setup_logger(log_dir: str = "logs") -> logging.Logger: """初始化根日志器,返回一个可直接使用的 logger Args: log_dir: 日志文件输出目录,默认 logs Returns: logging.Logger: 配置完成的 logger 实例 """ os.makedirs(log_dir, exist_ok=True) logger = logging.getLogger("agent") logger.setLevel(logging.INFO) # 避免重复配置 handler if logger.handlers: return logger # 控制台输出 console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(logging.INFO) console_handler.setFormatter(JsonFormatter()) # 文件输出,按 10MB 轮转,保留 5 份历史 file_handler = RotatingFileHandler( os.path.join(log_dir, "agent.log"), maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8", ) file_handler.setLevel(logging.INFO) file_handler.setFormatter(JsonFormatter()) logger.addHandler(console_handler) logger.addHandler(file_handler) return logger def bind_trace_id(trace_id: str = None, session_id: str = None) -> str: """在请求入口调用,绑定本次任务的 trace_id 和 session_id Args: trace_id: 外部传入的链路 ID,为空则自动生成 session_id: 会话 ID,表示同一个用户的多轮对话 Returns: str: 本次任务的 trace_id """ if trace_id is None: trace_id = uuid.uuid4().hex[:16] if session_id is not None: current_session_id.set(session_id) current_trace_id.set(trace_id) return trace_id关键逻辑说明:
current_trace_id是ContextVar,在异步环境里每次协程切换时能正确恢复,比threading.local更适合现代 Python 并发模型。JsonFormatter在每个日志记录产生时自动读取 ContextVar 中的值,业务代码不需要手动传 trace_id。RotatingFileHandler防止日志无限增长,10MB 一个文件,留 5 份历史,生产环境可以加大这个阈值,但本地演示绰绰有余。
5.2 第二步:模拟工具模块
文件路径:agent/tools.py
""" 模拟 Agent 的工具调用层 实际项目中这里可能是: - 调用外部搜索 API - 查询数据库 - 读写文件 - 调用另一个 Agent 服务 """ import time from .logger import current_session_id, current_trace_id, setup_logger logger = setup_logger() def search_knowledge_base(query: str) -> str: """模拟检索知识库 Args: query: 用户查询字符串 Returns: str: 检索结果文本 """ logger.info( "工具调用开始", extra={"extra_fields": { "tool_name": "search_knowledge_base", "tool_args": {"query": query}, }}, ) # 模拟耗时 time.sleep(0.3) # 模拟检索结果 result = ( f"知识库中与「{query}」相关的内容:\n" "1. Agent 日志设计应包含结构化字段和 trace_id\n" "2. 日志轮转和持久化是生产系统的底线\n" "3. 排查 Agent 问题时,优先还原链路而不是只看最终输出" ) logger.info( "工具调用返回", extra={"extra_fields": { "tool_name": "search_knowledge_base", "tool_result": result, }}, ) return result def execute_code(code_snippet: str) -> dict: """模拟执行模型生成的代码 Args: code_snippet: 模型生成的代码字符串 Returns: dict: 包含执行输出和退出码 """ logger.info( "执行模型生成代码", extra={"extra_fields": { "exec_code": code_snippet, }}, ) # 真实场景中可能是在沙箱容器里跑这段代码 # 这里只是模拟一个成功执行的结果 stdout = "任务执行成功,产出 42" stderr = "" exit_code = 0 logger.info( "代码执行完成", extra={"extra_fields": { "stdout": stdout, "stderr": stderr, "exit_code": exit_code, }}, ) return { "stdout": stdout, "stderr": stderr, "exit_code": exit_code, }这里有一个工程细节:工具调用开始和结束分成两条日志,而不是合并成一条。原因很简单:工具可能卡死或者超时,如果只写一条“调用开始”,你根本不知道它到底有没有返回;拆成“开始”和“返回”两条,结合时间戳就能算出每个工具的耗时,也能识别超时场景。
5.3 第三步:模拟 Agent 执行器
文件路径:agent/executor.py
""" 模拟 Agent 执行器 负责一次任务的完整生命周期: 1. 接收用户请求 2. 模拟 LLM 决策 3. 调用工具 4. 汇总结果 """ from .logger import setup_logger from .tools import execute_code, search_knowledge_base logger = setup_logger() class AgentExecutor: """一个最小可用的 Agent 执行器""" def __init__(self, model_name: str = "demo-model"): self.model_name = model_name self.memory = {} def run(self, user_input: str, session_id: str = None) -> str: """执行一次 Agent 任务 Args: user_input: 用户输入 session_id: 会话 ID Returns: str: Agent 的最终回答 """ # 1. 记录请求层日志 logger.info( "收到用户请求", extra={"extra_fields": { "user_input": user_input, "session_id": session_id, }}, ) # 2. 模拟 LLM 决策:决定先检索知识库,再执行一段代码 decision = { "action": "search_then_execute", "reason": "用户请求包含需要代码验证的问题", } logger.info( "LLM 决策", extra={"extra_fields": { "model_name": self.model_name, "decision": decision, }}, ) # 3. 调用检索工具 knowledge = search_knowledge_base(query=user_input) # 4. 模拟模型生成一段用来验证的代码 code_snippet = "result = 40 + 2" execute_result = execute_code(code_snippet) # 5. 更新内部状态 state_before = dict(self.memory) self.memory["last_query"] = user_input self.memory["last_code_result"] = execute_result["stdout"] state_after = dict(self.memory) logger.info( "状态更新", extra={"extra_fields": { "state_before": state_before, "state_after": state_after, }}, ) # 6. 汇总最终回复 final_answer = f"执行完成,结果:{execute_result['stdout']},参考知识:{knowledge[:30]}..." logger.info( "返回最终结果", extra={"extra_fields": {"final_answer": final_answer}}, ) return final_answer这个执行器一共写了六层日志:请求、决策、检索工具开始/返回、代码执行、状态更新、最终结果。你注意观察,这类日志一旦落地,你用任何一个日志分析平台都能画出完整的时序图。
5.4 第四步:入口文件
文件路径:main.py
""" Agent 日志示例入口 用法: python main.py 会连续执行三次 Agent 任务,展示 trace_id 如何贯穿所有日志 """ from agent.executor import AgentExecutor from agent.logger import bind_trace_id, setup_logger logger = setup_logger() def handle_request(user_input: str, session_id: str = None): """模拟一个请求入口:先绑定 trace_id,再执行业务""" trace_id = bind_trace_id(session_id=session_id) logger.info("请求进入") executor = AgentExecutor() return trace_id, executor.run(user_input, session_id=session_id) if __name__ == "__main__": # 第一次请求:用户询问如何设计日志 tid1, res1 = handle_request("如何设计 Agent 日志体系", session_id="session-001") print(f"[task-1] trace_id={tid1}") print(f"[task-1] result={res1}") # 第二次请求:用户询问代码丢失问题 tid2, res2 = handle_request("代码被删了怎么排查", session_id="session-001") print(f"[task-2] trace_id={tid2}") print(f"[task-2] result={res2}") # 第三次请求:新会话用户 tid3, res3 = handle_request("如何让 Agent 记住用户偏好", session_id="session-002") print(f"[task-3] trace_id={tid3}") print(f"[task-3] result={res3}")入口文件演示的是最简单也最核心的调用顺序:先绑定 trace_id,再执行业务。这个顺序不能错。如果先执行业务再绑定 ID,前面那一大段日志就没有 trace_id,链路就断了一截。
5.5 运行方式
在项目根目录执行:
pip install -r requirements.txt python main.py注意:示例代码本身不需要任何第三方依赖,只是保留requirements.txt占位。如果你的项目后续接入分布式链路追踪组件(如 OpenTelemetry),再往里加依赖。
预期输出:
{"timestamp": "2025-01-01 10:00:00", "level": "INFO", "logger": "agent", "module": "logger", "function": "handle_request", "line": 15, "trace_id": "9f2c3d4e5a6b7c8d", "session_id": "session-001", "message": "请求进入"} {"timestamp": "2025-01-01 10:00:00", "level": "INFO", "logger": "agent", "module": "executor", "function": "run", "line": 38, "trace_id": "9f2c3d4e5a6b7c8d", "session_id": "session-001", "message": "收到用户请求", "user_input": "如何设计 Agent 日志体系", "session_id": "session-001"} {"timestamp": "2025-01-01 10:00:00", "level": "INFO", "logger": "agent", "module": "executor", "function": "run", "line": 48, "trace_id": "9f2c3d4e5a6b7c8d", "session_id": "session-001", "message": "LLM 决策", "model_name": "demo-model", "decision": {"action": "search_then_execute", "reason": "用户请求包含需要代码验证的问题"}} ...(后续日志省略)三条任务会生成三个不同的 trace_id。同一个 session-001 的两条任务 trace_id 不同但 session_id 相同,你既能按任务链路查,也能按用户会话查。
6. 运行结果与效果验证
代码跑通之后,怎么判断这套日志体系真的合格?我建议用三个维度验证:
6.1 验证 trace_id 贯穿性
用 grep 搜索刚才第一次请求的 trace_id:
grep "9f2c3d4e5a6b7c8d" logs/agent.log | jq .message如果配置正确,你会看到从“请求进入”到“返回最终结果”的完整消息序列。如果你看到某条日志没有 trace_id 或者 trace_id 是-,说明绑定时机太晚或者某些异步分支没有继承 ContextVar。
6.2 验证跨任务隔离
检查第三次请求的 trace_id 不会出现在第一次请求的链路里。用 grep 只能搜到对应任务自己的日志。这是检验 ContextVar 是否产生污染的关键指标。如果多个任务意外共享同一个 trace_id,你会看到日志链路交叉,排查时会完全分不清是哪次任务产生的调用。
6.3 验证日志可读性与轮转
打开logs/agent.log,看 JSON 是否合法,缩进是否正确。连续跑很多次任务后,观察文件是否按预期轮转。可以用循环模拟压力:
for i in $(seq 1 50); do python main.py > /dev/null; done跑完 50 次之后检查logs目录,正常情况下agent.log和多个agent.log.1、agent.log.2会在目录里,这证明文件轮转生效,不会出现单文件无限膨胀的问题。
7. 常见问题与排查思路
下面是 Agent 日志项目中最常见的问题,按出现的频率排序:
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
日志没有 trace_id,显示为- | 业务代码在 bind_trace_id 之前就打了日志 | 检查入口函数中 bind_trace_id 的调用位置 | 确保在所有业务逻辑之前调用 bind_trace_id |
| 异步任务之间 trace_id 串线 | 使用 threading.local 而不是 ContextVar,或新协程没有正确传播上下文 | 打印日志时检查 trace_id 是否频繁变化 | 改为 ContextVar,并封装传播函数 |
| 日志文件增长过快 | 没有配置轮转,或轮转阈值太大 | 查看文件大小和写入频率 | 使用 RotatingFileHandler,设置合理的 maxBytes |
| 日志内容仍是纯文本 | 没有应用 JsonFormatter | 检查 setup_logger 中 formatter 设置 | 确保所有 handler 都绑定 JsonFormatter |
| 工具层日志缺失 | 工具调用封装层没有使用同一 logger | 检查工具模块里的 logger 是logging.getLogger("agent")而不是__name__ | 统一使用 prefixed logger 名称 |
| LLM 响应太长导致日志臃肿 | 原样记录完整 prompt 和响应 | 查看单条日志大小 | 对长文本做截断,比如保留前 1000 字符 |
| 日志里有业务字段但搜索平台不展示 | extra_fields 字段名和搜索平台字段冲突 | 检查日志平台字段映射 | 统一 extra 字段命名规范,加上前缀 |
这里特别想强调第一行:trace_id 为-的排查。
很多开发者的做法是“先跑一下看效果”,如果发现日志里大量出现-,就会去翻 logger 的代码。但更高效的排查路径是先在入口函数看一眼 bind_trace_id 在哪一行调用。
有一次我帮同事排查,他写的 Agent 服务在收到消息后,先做了一堆数据清洗,再把 trace_id 绑定上。结果每次出问题时,数据清洗阶段产生的日志全部没有链路 ID,等于最关键的输入日志断了。这个错误很隐蔽,因为程序能跑,日志也能出,只是链路缺了一截。你只有把“绑定 trace_id”作为请求处理的第一行固化下来,才能真正避免。
8. 代码被删光之后:如何通过日志重建现场
回到文章标题的场景:代码真的被删了,只剩日志,怎么活下去?
先说清楚一个事实:日志不能百分之百恢复代码,但日志能给你三条线索,让你快速重新写回来。
8.1 从决策日志还原系统设计意图
决策层日志会记录模型选择了哪个 action、理由是什么。哪怕代码全没了,你翻日志能知道这个 Agent 的业务路径:用户输入进来之后是先检索还是先判断,遇到什么情况会执行代码,最终结果怎么汇总。这些信息拼起来,就是一份天然的设计文档。
8.2 从代码执行日志还原算法实现
代码执行层日志记录了exec_code字段。这等于把模型动态生成过的代码全部沉淀下来了。你把日志里所有 exec_code 字段提取出来,按时间排序,就能看到模型在反复迭代什么逻辑、最终收敛到哪个版本。
提取命令示例:
grep '"message": "执行模型生成代码"' logs/agent.log | jq .exec_code > recovered_code.txt如果你是自研 Agent,建议在代码执行日志里加上代码的 hash 值。一旦发现线上代码和日志里的 hash 对不上,就说明代码被外部修改了,可以立刻定位是哪次任务、哪个环节改的。
8.3 从状态日志还原上下文
状态层日志记录 state_before 和 state_after。当代码丢失导致内存状态不清时,你可以从日志的时间序列回放状态机的变迁过程。比如用户会话 session-001 的偏好设置是什么时候写入的、最后一次写入的值是什么,这些都能从状态日志里捞出来。
8.4 一套“从日志恢复”的实践流程
如果发生代码丢失事故,推荐按下面的顺序处理:
- 冻结现场:立即停止 Agent 服务,防止新的日志覆盖旧的轮转文件。
- 全量备份日志目录:把 logs 目录压缩存档,包括轮转文件。小心不要打开日志文件后顺手清理了什么。
- 按 trace_id 聚合链路:用日志平台或 grep 把最近一次成功运行的完整链路还原出来。
- 提取 exec_code 和 tool_args:把动态代码和工具调用参数整理成可执行的还原草稿。
- 结合状态日志补充上下文:在还原代码的同时,把会话状态和内存数据也恢复出来。
- 验证恢复结果:用旧日志里的输入用例重放一遍,对比最终回答是否接近。
这套流程不依赖任何特定技术栈,只要你的日志包含 trace_id、exec_code、tool_args、state_before/after 这几个核心字段,就能执行。
9. 生产级 Agent 日志的最佳实践
最后把工程经验收敛成几条可以直接执行的原则。
9.1 字段命名统一,提前预留扩展位
日志字段的命名最好从一开始就定好规范。比如统一用 snake_case,所有业务字段放在extra_fields中,不要直接在 message 里拼接 JSON。这样无论后续接 ELK、Loki 还是 ClickHouse,字段映射都简单。建议在日志配置里加入service_name、environment、version三个固定字段,发布新版本后可以直接从日志里区分新旧行为。
9.2 敏感信息必须过滤
Agent 日志里最容易出现三类敏感数据:用户输入的隐私文本、工具返回的敏感内容、模型 prompt 里内嵌的业务密钥。在JsonFormatter里加一层脱敏逻辑,规则很简单:
- 手机号、邮箱、身份证号用正则替换为
***; - 密钥字段名包含
token、api_key、secret时,值直接置空; - 用户原始输入超过 500 字符直接截断,长文本协议落到专门的审计日志。
9.3 日志要进中心化平台,不要只留在本地文件
本地文件日志的作用是开发和单机排查。生产环境一旦有多个 Agent 副本,日志必须进入中心化采集平台。你可以用 Filebeat 采集本地文件,也可以用日志库直接对接 Kafka、OpenTelemetry。中心化之后,按 trace_id 跨服务检索才可能实现,不然一个任务横跨多个容器,你只能一台台机器去翻。
9.4 代码执行日志要包含沙箱标识
如果 Agent 会动态执行代码,一定要记录这段代码运行在哪个沙箱、哪个容器、用了什么镜像。多个 Agent 同时跑代码时,如果日志里没有沙箱标识,出现问题后你根本不知道代码跑在哪里,更不用说回收资源。
9.5 日志存储时间要足够长
普通后端日志保留 30 天可能够了,Agent 日志建议至少保留 90 天到 180 天。原因很直接:Agent 的异常行为可能和很久之前的一次工具返回值或者状态变更有关,链路越长,回溯窗口就越长。存储成本可以用“分级别存储”来平衡:全量 JSON 日志放冷存储,trace_id 索引放热存储,只有被标记为异常的链路才完整加载分析。
9.6 定期做“日志自检”
每两周做一次日志质量检查,要点是:
- 检查有没有出现 trace_id 为
-的日志; - 检查是否有新代码路径没有记录任何日志;
- 检查日志是否因为格式变更导致采集端解析失败;
- 随机抽一条异常链路,看是否能用日志完整还原全过程。
日志自检和代码审查一样,投入不大,但能持续防止日志体系腐化。
10. 结尾
Agent 项目的代码会迭代、会重构、会被模型改写,甚至会在某个深夜被一条误操作清空。但日志不会,它忠实地记录着每一次请求、每一次模型决策、每一次工具调用、每一段代码执行。
把 trace_id 记进每一条日志,把工具调用拆成开始和结束,把代码执行记录下来,把状态变更前后翻开,做到这几点,你的 Agent 项目就有了自愈的能力。下次再遇到“代码被删光”的极端情况,你至少有日志这条退路。
建议收藏这篇文章,最方便的做法,是先把第 5 节的示例代码原地跑一遍——三分钟就能看见 trace_id 怎样把每一层日志串成一条可回溯的完整链路。