Agent 日志结构化方案便于事后审计与调试的工具调用记录Agent 调了什么工具、传了什么参数、返回了什么结果——如果这三件事没记清楚事后排查就是盲人摸象。一、场景痛点你上线了一个多步骤 Agent 系统用户反馈结果不对。你打开日志一看满屏都是console.log输出的 JSON 字符串工具调用记录和业务日志混在一起没有统一的字段名有的用tool_name有的用tool时间戳格式也不统一。要找到某次工具调用的完整链路你得手动 grep 三四个关键字再拼凑上下文。更麻烦的是合规场景。审计方要求你证明Agent 在处理用户数据时调用了哪些外部服务、传入了哪些参数。你翻遍日志只能给出一个模糊的说法因为没有结构化的调用记录参数和返回值散落在不同日志行里。核心矛盾Agent 的工具调用是离散的事件但调试和审计需要的是连续的链路。二、底层机制与原理剖析2.1 Agent 工具调用的生命周期一次完整的工具调用不是调一下就完了它包含六个阶段意图识别→参数提取→工具选择→调用执行→结果解析→结果注入。每个阶段都可能失败每个阶段的耗时都需要记录。2.2 结构化日志的数据模型结构化日志的核心是统一的数据模型。每个工具调用记录至少包含以下字段trace_id链路唯一标识贯穿整个 Agent 会话span_id单次工具调用的唯一标识parent_span_id父调用标识支持嵌套调用链tool_name工具名称与注册表中的名称一致tool_version工具版本用于追踪版本变更导致的行为差异input_params调用参数完整记录敏感字段脱敏output_result返回结果摘要大结果截断保留前 N 字符status调用状态——success / failure / timeouterror_detail失败时的错误信息start_time / end_time / duration_ms时间三元组token_usage本次调用涉及的 token 消耗如果有2.3 日志与链路追踪的关系结构化日志不等于链路追踪。链路追踪关注的是请求在服务间的流转路径日志关注的是事件的内容细节。两者互补trace_id 将它们关联起来你在 Jaeger 里看链路拓扑在 Elasticsearch 里看调用细节。三、生产级代码实现3.1 日志结构化工具类// agent-logger.ts —— Agent 工具调用结构化日志工具 import { v4 as uuidv4 } from uuid; /** 调用状态枚举 */ export enum CallStatus { SUCCESS success, FAILURE failure, TIMEOUT timeout, } /** 脱敏规则字段名 → 脱敏函数 */ type Sanitizer (value: unknown) string; const defaultSanitizers: Recordstring, Sanitizer { // 手机号脱敏保留前3后4 phone: (v) typeof v string ? v.replace(/(\d{3})\d{4}(\d{4})/, $1****$2) : [REDACTED], // 身份证脱敏保留前6后4 id_card: (v) typeof v string ? v.replace(/(\d{6})\d{8}(\d{4})/, $1********$2) : [REDACTED], // 通用敏感字段直接掩码 password: () [MASKED], api_key: () [MASKED], token: () [MASKED], }; /** 日志记录的数据结构 */ export interface ToolCallRecord { trace_id: string; span_id: string; parent_span_id: string | null; tool_name: string; tool_version: string; input_params: Recordstring, unknown; output_result: string; status: CallStatus; error_detail: string | null; start_time: string; // ISO 8601 end_time: string; // ISO 8601 duration_ms: number; token_usage: { input: number; output: number } | null; metadata: Recordstring, string; // 自定义扩展字段 } export class AgentLogger { private traceId: string; private sanitizers: Recordstring, Sanitizer; private outputMaxLength: number; // 日志输出通道可以替换为文件、HTTP、消息队列 private sink: (record: ToolCallRecord) void; constructor( traceId?: string, sanitizers?: Recordstring, Sanitizer, outputMaxLength 2000, sink?: (record: ToolCallRecord) void ) { // trace_id 不传则自动生成保证每条链路都有唯一标识 this.traceId traceId ?? uuidv4(); this.sanitizers { ...defaultSanitizers, ...sanitizers }; this.outputMaxLength outputMaxLength; // 默认 sink 输出到 stdout生产环境替换为结构化日志收集器 this.sink sink ?? ((r) console.info(JSON.stringify(r))); } /** 获取当前 trace_id便于在上下游传递 */ getTraceId(): string { return this.traceId; } /** 脱敏处理遍历参数对象对命中规则的字段执行脱敏 */ private sanitizeParams(params: Recordstring, unknown): Recordstring, unknown { const result: Recordstring, unknown {}; for (const [key, value] of Object.entries(params)) { // 字段名命中脱敏规则则执行脱敏函数否则原值保留 const sanitizer this.sanitizers[key]; result[key] sanitizer ? sanitizer(value) : value; } return result; } /** 截断大结果避免一条日志占满存储空间 */ private truncateOutput(output: string): string { if (output.length this.outputMaxLength) return output; return output.slice(0, this.outputMaxLength) ...[TRUNCATED, total ${output.length} chars]; } /** 记录一次工具调用的完整生命周期 */ async recordToolCall( toolName: string, toolVersion: string, inputParams: Recordstring, unknown, parentSpanId: string | null, executor: () Promisestring, tokenUsage?: { input: number; output: number }, metadata?: Recordstring, string ): PromiseToolCallRecord { const spanId uuidv4(); const startTime new Date().toISOString(); let status: CallStatus; let outputResult: string; let errorDetail: string | null null; let endTime: string; let durationMs: number; try { // 执行工具调用捕获结果 outputResult await executor(); status CallStatus.SUCCESS; endTime new Date().toISOString(); durationMs Date.parse(endTime) - Date.parse(startTime); } catch (err: unknown) { // 区分超时和一般错误 const errorMsg err instanceof Error ? err.message : String(err); if (errorMsg.includes(timeout) || errorMsg.includes(ETIMEDOUT)) { status CallStatus.TIMEOUT; } else { status CallStatus.FAILURE; } outputResult ; errorDetail errorMsg; endTime new Date().toISOString(); durationMs Date.parse(endTime) - Date.parse(startTime); } // 组装结构化记录脱敏参数 截断输出 const record: ToolCallRecord { trace_id: this.traceId, span_id: spanId, parent_span_id: parentSpanId, tool_name: toolName, tool_version: toolVersion, input_params: this.sanitizeParams(inputParams), output_result: this.truncateOutput(outputResult), status, error_detail: errorDetail, start_time: startTime, end_time: endTime, duration_ms: durationMs, token_usage: tokenUsage ?? null, metadata: metadata ?? {}, }; // 写入日志通道 this.sink(record); return record; } }3.2 与 Agent 执行引擎集成// agent-executor.ts —— Agent 执行引擎集成日志记录 import { AgentLogger, CallStatus } from ./agent-logger; interface ToolDefinition { name: string; version: string; execute: (params: Recordstring, unknown) Promisestring; } export class AgentExecutor { private tools: Mapstring, ToolDefinition new Map(); private logger: AgentLogger; // 全局超时单次工具调用最长等待时间 private toolTimeoutMs: number; constructor(logger: AgentLogger, toolTimeoutMs 30000) { this.logger logger; this.toolTimeoutMs toolTimeoutMs; } /** 注册工具名称与定义映射 */ registerTool(tool: ToolDefinition): void { this.tools.set(tool.name, tool); } /** 执行单步工具调用带超时保护 */ async invokeTool( toolName: string, params: Recordstring, unknown, parentSpanId: string | null ): Promise{ result: string; record: ToolCallRecord } { const tool this.tools.get(toolName); if (!tool) { // 工具不存在时直接记录失败避免静默跳过 const record await this.logger.recordToolCall( toolName, unknown, params, parentSpanId, () { throw new Error(Tool not found: ${toolName}); } ); return { result: , record }; } // 用 Promise.race 实现超时控制工具执行 vs 超时定时器 const record await this.logger.recordToolCall( tool.name, tool.version, params, parentSpanId, () Promise.race([ tool.execute(params), new Promisenever((_, reject) setTimeout(() reject(new Error(Tool timeout: ${toolName} exceeded ${this.toolTimeoutMs}ms)), this.toolTimeoutMs) ), ]), ); return { result: record.output_result, record }; } /** 执行多步骤 Agent 链路串行调用并串联 span_id */ async executeChain( steps: Array{ toolName: string; params: Recordstring, unknown } ): PromiseToolCallRecord[] { const records: ToolCallRecord[] []; let parentSpanId: string | null null; for (const step of steps) { const { record } await this.invokeTool(step.toolName, step.params, parentSpanId); records.push(record); // 成功则继续失败则终止链路避免后续步骤依赖错误数据 if (record.status ! CallStatus.SUCCESS) { break; } // 当前 span 作为下一步的 parent形成调用链 parentSpanId record.span_id; } return records; } }3.3 日志持久化到 Elasticsearch# log_sink.py —— 将结构化日志写入 Elasticsearch import json import logging from datetime import datetime from elasticsearch import Elasticsearch, helpers logger logging.getLogger(agent-log-sink) class ElasticsearchSink: Agent 日志写入 Elasticsearch支持批量提交 def __init__(self, es_url: str, index_name: str agent-tool-calls, bulk_size: int 100, flush_interval_sec: int 5): self.es Elasticsearch(es_url, timeout30) self.index_name index_name self.bulk_size bulk_size self.flush_interval_sec flush_interval_sec self._buffer: list[dict] [] # 确保索引存在指定映射以避免动态映射的类型猜测问题 self._ensure_index() def _ensure_index(self): 预定义索引 mapping保证字段类型一致 mapping { mappings: { properties: { trace_id: {type: keyword}, span_id: {type: keyword}, parent_span_id: {type: keyword}, tool_name: {type: keyword}, tool_version: {type: keyword}, input_params: {type: object, enabled: True}, output_result: {type: text, analyzer: standard}, status: {type: keyword}, error_detail: {type: text}, start_time: {type: date, format: strict_date_optional_time}, end_time: {type: date, format: strict_date_optional_time}, duration_ms: {type: long}, token_usage: {type: object}, metadata: {type: object, enabled: True}, } } } # 索引不存在则创建已存在则忽略 if not self.es.indices.exists(indexself.index_name): self.es.indices.create(indexself.index_name, bodymapping) logger.info(fCreated ES index: {self.index_name}) def write(self, record: dict): 单条写入攒到 bulk_size 后批量提交 # 添加 timestamp 字段便于 Kibana 按时间检索 record[timestamp] datetime.utcnow().isoformat() Z self._buffer.append(record) if len(self._buffer) self.bulk_size: self._flush() def _flush(self): 批量提交缓冲区中的日志 if not self._buffer: return try: # helpers.bulk 自动分批提交减少网络往返 actions [ { _index: self.index_name, _source: record, } for record in self._buffer ] success, errors helpers.bulk(self.es, actions, raise_on_errorFalse) if errors: logger.warning(fBulk write partial failure: {len(errors)} errors) logger.info(fFlushed {success} records to ES) self._buffer.clear() except Exception as e: # 写入失败不丢弃日志保留在缓冲区下次重试 logger.error(fES bulk write failed: {e}, keeping buffer for retry) def close(self): 关闭前强制 flush确保不丢数据 self._flush()四、边界分析与架构权衡4.1 结构化日志的存储成本每条工具调用日志约 1-5 KB一个 Agent 会话可能产生 10-50 条记录。日活 10 万用户、平均 20 次调用一天就是 200 万条 × 3 KB ≈ 6 GB。一个月 180 GB加上 ES 的索引膨胀约 1.5 倍存储量接近 270 GB。对策设置 TTL保留 30 天热数据超期归档到冷存储对 output_result 做截断大结果只保留摘要。4.2 脱敏规则的维护负担脱敏规则需要持续更新。新增一个工具它的参数里可能包含新的敏感字段。如果规则滞后隐私数据就泄露到日志里了。对策脱敏规则集中管理工具注册时必须声明参数中的敏感字段列表日志工具自动应用对应脱敏策略。4.3 适用边界与禁用场景适用需要审计合规的金融/医疗场景、多工具组合的复杂 Agent 链路、事后排查频率高的生产系统禁用单工具单步骤的简单 Agent日志开销大于收益、内网无合规要求的研发环境、实时性要求极高且日志写入影响延迟的场景4.4 与 OpenTelemetry 的重叠OpenTelemetry 已经提供了 span 属性机制你可以把工具调用信息塞进 OTel span 的 attributes 里不需要另建一套日志系统。区别在于OTel 的 span 数据量受采样策略限制结构化日志是全量记录。如果你需要 100% 的审计覆盖不能依赖采样必须走独立日志通道。五、总结Agent 工具调用日志的结构化不是一个加几行 console.log的小事它涉及数据模型定义、脱敏规则管理、存储成本控制、与链路追踪的互补设计。核心原则trace_id 串联链路、span_id 标识调用、脱敏保护隐私、截断控制成本。生产系统中这四个维度缺一个事后排查和合规审计就会出漏洞。