AI Agent 日志审计记录 agent 每次工具调用的入参、出参和耗时标签AI、AI Agent、日志、审计、可观测性一、Agent 到底干了什么测试环境里 Agent 处理一个工单它调了 5 个工具、跑了 30 秒最后返回已完成。表面上看好像没问题。但有一天我突然发现它在某一步用错了参数调用了错误的 API虽然最后还是返回了结果但这个结果是错的。我想追查发生了什么事。但回头一看——没有任何记录。日志里只有一条Agent 任务完成。至于它调了什么工具、传了什么参数、每一步用了多长时间、中间有没有出错——完全不知道。这件事让我意识到给 Agent 配齐日志审计系统不是可选的高级功能而是基本的安全底线。这篇文章就是我设计 Agent 审计日志系统的过程。不算什么高深的东西但是很实用。二、审计日志需要记录什么动手之前先想清楚要记录什么。不全不行太多又会影响性能。flowchart LR subgraph AgentLife[Agent 一次执行的生命周期] Start[用户输入任务] -- Plan[Agent 规划步骤] Plan -- Tool1[调用工具 A] Tool1 -- Observe1[观察结果] Observe1 -- Decision{继续还是完成} Decision --|继续| Tool2[调用工具 B] Decision --|完成| Output[输出最终结果] Tool2 -- Observe2[观察结果] Observe2 -- Decision2{继续还是完成} Decision2 --|继续| ToolN[调用更多工具...] Decision2 --|完成| Output end subgraph AuditPoints[需要审计的关键点] AP1[1. 任务开始输入、时间戳] AP2[2. 每次工具调用工具名、入参、出参、耗时、状态] AP3[3. 中间决策LLM 返回的思考过程] AP4[4. 任务结束最终输出、总耗时、工具调用次数] end Start -.- AP1 Tool1 -.- AP2 Tool2 -.- AP2 ToolN -.- AP2 Output -.- AP4 style AuditPoints fill:#e3f2fd我把审计日志分成四个维度维度内容用途请求级任务 ID、用户输入、会话 ID、开始时间关联一次完整的 Agent 调用工具调用级工具名称、入参 JSON、出参 JSON、耗时、成功/失败复现每一步的操作LLM 调用级模型名称、Prompt Tokens、Completion Tokens、总 Tokens、延迟分析成本和质量聚合级总耗时、总工具调用次数、总 Token 消耗宏观统计和计费三、实现用 decorator 模式包装工具调用我不想在每个工具函数里面加日志代码那样太恶心了。我用的是decorator 模式把所有工具调用包装一层自动记录。// // Agent 工具调用的审计日志系统 // 核心思路用一个 wrap 函数拦截所有工具调用 // 自动记录入参、出参、耗时、错误信息。 // 对业务代码零侵入。 // use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::sync::Arc; use std::time::Instant; use tokio::sync::Mutex; use tracing::{info, warn, error}; /// 单次工具调用的审计记录 #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ToolCallRecord { /// 调用的工具名称如 search_database, send_email pub tool_name: String, /// 工具调用的入参JSON 格式方便后续搜索分析 pub input_params: serde_json::Value, /// 工具调用的出参JSON 格式 pub output_result: Optionserde_json::Value, /// 是否成功None 未完成Some(true) 成功Some(false) 失败 pub success: Optionbool, /// 错误信息如果有的话 pub error_message: OptionString, /// 工具调用的开始时间毫秒时间戳 pub start_time_ms: u64, /// 工具调用的耗时毫秒 pub duration_ms: u64, /// 调用序号本次 agent 任务中第几次工具调用 pub call_index: usize, } /// Agent 一次任务的完整审计记录 #[derive(Debug, Clone, Serialize, Deserialize)] pub struct AgentAuditRecord { /// 全局唯一的任务 ID pub task_id: String, /// 用户的原始输入 pub user_input: String, /// 会话 ID如果一个对话中有多轮 agent 调用 pub session_id: String, /// 任务开始的时间戳毫秒 pub start_time_ms: u64, /// 任务结束的时间戳毫秒 pub end_time_ms: Optionu64, /// 任务状态running / completed / failed pub status: String, /// 所有工具调用的记录按时间顺序 pub tool_calls: VecToolCallRecord, /// 总的 Token 消耗 pub total_tokens: u64, /// Agent 的最终输出 pub final_output: OptionString, } /// 审计记录器全局单例或请求级实例 pub struct AuditLogger { /// 当前任务的审计记录 current_record: AgentAuditRecord, /// 工具调用计数器 call_counter: usize, } impl AuditLogger { /// 创建一个新的审计记录器 pub fn new(task_id: String, user_input: String, session_id: String) - Self { Self { current_record: AgentAuditRecord { task_id, user_input, session_id, start_time_ms: now_ms(), end_time_ms: None, status: running.to_string(), tool_calls: Vec::new(), total_tokens: 0, final_output: None, }, call_counter: 0, } } /// 包装一个工具调用自动记录审计日志 /// /// 用法 /// ignore /// let result logger /// .wrap_tool_call(search_database, json!({query: xxx}), /// || async { my_search_fn(xxx).await } /// ) /// .await; /// pub async fn wrap_tool_callF, Fut, T( mut self, tool_name: str, input_params: serde_json::Value, f: F, ) - ResultT, String where F: FnOnce() - Fut, Fut: std::future::FutureOutput ResultT, String, { self.call_counter 1; let index self.call_counter; // 记录开始时间 let start Instant::now(); let start_ms now_ms(); info!( task_id %self.current_record.task_id, tool %tool_name, call_index index, input %input_params, 工具调用开始 ); // 执行实际的工具调用 let result f().await; let duration start.elapsed(); let duration_ms duration.as_millis() as u64; // 根据结果创建记录 let record match result { Ok(value) { info!( task_id %self.current_record.task_id, tool %tool_name, duration_ms duration_ms, 工具调用成功 ); ToolCallRecord { tool_name: tool_name.to_string(), input_params, output_result: Some(serde_json::to_value(value).unwrap_or_default()), success: Some(true), error_message: None, start_time_ms: start_ms, duration_ms, call_index: index, } } Err(err) { error!( task_id %self.current_record.task_id, tool %tool_name, duration_ms duration_ms, error %err, 工具调用失败 ); ToolCallRecord { tool_name: tool_name.to_string(), input_params, output_result: None, success: Some(false), error_message: Some(err.clone()), start_time_ms: start_ms, duration_ms, call_index: index, } } }; // 追加到记录列表 self.current_record.tool_calls.push(record); result } /// 标记任务完成记录最终输出 pub fn complete(mut self, output: String) { self.current_record.status completed.to_string(); self.current_record.end_time_ms Some(now_ms()); self.current_record.final_output Some(output); let total_duration self.current_record.end_time_ms.unwrap() - self.current_record.start_time_ms; info!( task_id %self.current_record.task_id, total_duration_ms total_duration, tool_calls_count self.current_record.tool_calls.len(), Agent 任务完成 ); } /// 标记任务失败 pub fn fail(mut self, error: str) { self.current_record.status failed.to_string(); self.current_record.end_time_ms Some(now_ms()); self.current_record.final_output Some(error.to_string()); error!( task_id %self.current_record.task_id, error %error, Agent 任务失败 ); } /// 获取审计记录的副本 pub fn get_record(self) - AgentAuditRecord { self.current_record.clone() } } /// 获取当前毫秒时间戳 fn now_ms() - u64 { use std::time::SystemTime; SystemTime::now() .duration_since(SystemTime::UNIX_EPOCH) .unwrap_or_default() .as_millis() as u64 }使用示例/// 模拟一个 Agent 的执行流程 /// 展示如何在实际业务中使用 AuditLogger async fn agent_workflow(user_query: str, session_id: str) { // 创建一个新的审计记录器 let task_id uuid::Uuid::new_v4().to_string(); let mut logger AuditLogger::new( task_id.clone(), user_query.to_string(), session_id.to_string(), ); // 第一步查询数据库 let db_result logger .wrap_tool_call( search_database, // 工具名称 serde_json::json!({query: user_query}), // 入参JSON || async { // 实际工具调用的逻辑 // 这里用一个模拟的数据库查询代替 let result format!(查到 3 条关于 {} 的记录, user_query); Ok(result) }, ) .await; match db_result { Ok(data) { // 第二步基于第一步的结果调用外部 API let api_result logger .wrap_tool_call( call_external_api, serde_json::json!({endpoint: /analyze, data: data}), || async { // 模拟调用外部 API // 如果 API 返回错误会被自动记录到审计日志 Ok::_, String(format!(分析结果: {}, data)) }, ) .await; match api_result { Ok(analysis) { // 第三步发送通知 let _ logger .wrap_tool_call( send_notification, serde_json::json!({channel: email, content: analysis}), || async { Ok::_, String(通知已发送.to_string()) }, ) .await; // 标记任务完成 logger.complete(analysis); } Err(e) { logger.fail(format!(外部 API 调用失败: {}, e)); } } } Err(e) { logger.fail(format!(数据库查询失败: {}, e)); } } // 将审计记录存入数据库或日志系统 let audit_record logger.get_record(); save_audit_record(audit_record).await; } /// 模拟保存审计记录到数据库 async fn save_audit_record(record: AgentAuditRecord) { // 实际项目中这里会写入数据库 tracing::info!( 审计记录已保存: task_id{}, tool_calls{}, status{}, record.task_id, record.tool_calls.len(), record.status, ); }四、审计日志的存储和分析记录下来了怎么存怎么查flowchart TD Agent[Agent 执行] -- Logger[AuditLogger 记录] Logger -- Structured[结构化日志\nJSON 格式] Structured -- Path1[路径 1写入数据库\nPostgreSQL / ClickHouse] Structured -- Path2[路径 2写入文件\nJSON Lines 格式] Path1 -- Query[SQL 查询分析] Path2 -- Grep[grep / jq 快速检索] Query -- Dash[监控面板] Query -- Alert[告警规则] Grep -- Debug[本地调试] Alert -- Rules[异常检测规则\n1. 工具调用失败率 5%\n2. 单次工具调用 10s\n3. 单任务 Token 10000] style Logger fill:#e3f2fd style Dash fill:#c8e6c9 style Alert fill:#ffcdd2我目前的方案比较朴素——存到 PostgreSQL用几行 SQL 就能做基本的分析-- 审计日志分析 SQL 示例 -- 1. 查最近 1 小时失败的工具调用 SELECT task_id, tool_name, input_params, error_message, duration_ms FROM agent_tool_calls WHERE success false AND start_time_ms EXTRACT(EPOCH FROM NOW() - INTERVAL 1 hour) * 1000 ORDER BY start_time_ms DESC; -- 2. 找出耗时的工具调用 TOP 10 SELECT tool_name, AVG(duration_ms)::INTEGER AS avg_duration_ms, MAX(duration_ms) AS max_duration_ms, COUNT(*) AS call_count FROM agent_tool_calls WHERE start_time_ms EXTRACT(EPOCH FROM NOW() - INTERVAL 1 day) * 1000 GROUP BY tool_name ORDER BY avg_duration_ms DESC LIMIT 10; -- 3. 按任务统计 Token 消耗用于成本分析 SELECT DATE_TRUNC(hour, TO_TIMESTAMP(start_time_ms / 1000)) AS hour, SUM(total_tokens) AS total_tokens, COUNT(DISTINCT task_id) AS task_count FROM agent_audit_records WHERE start_time_ms EXTRACT(EPOCH FROM NOW() - INTERVAL 7 days) * 1000 GROUP BY hour ORDER BY hour DESC; -- 4. 找出异常工具调用次数异常多的任务 SELECT task_id, user_input, tool_calls_count, total_duration_ms, total_tokens FROM agent_audit_records WHERE tool_calls_count 10 -- 设置一个合理的阈值 ORDER BY tool_calls_count DESC LIMIT 20;告警规则基于审计日志可以设置这些告警单个工具调用失败率 5%→ 某个工具可能出问题了单次工具调用耗时 10 秒→ 外部依赖可能有性能问题单任务 Token 消耗 10,000→ Agent 可能陷入了推理循环五、总结审计日志这件事说简单也简单说重要也重要Agent 必须有可审计的操作记录。它不像传统程序——代码有确定的执行路径。Agent 的决策是黑箱你必须至少知道它调了什么工具、传了什么参数。用包装器模式比在每个函数里加日志好得多。业务逻辑和审计逻辑分离代码干净也不会有遗漏。记录什么要提前想清楚。入参、出参、耗时、错误——这四个维度是最基本的。Token 消耗和 LLM 延迟是进阶的。存储要结构化JSON。未来你需要按工具名称、时间段、Token 消耗做过滤和聚合这些分析能力来自于良好的数据结构设计。日志不是存了就完了。至少要配几条基本的 SQL 和告警规则。否则出了问题你还是得去翻日志而不是被日志主动告知出了问题。其实这些都是我在实战中摔打出来的教训。刚开始我以为审计日志就是打个日志后来发现它是一套完整的可观测性体系——采集、存储、查询、告警——缺一不可。如果你也在做 Agent 相关的东西希望我的经验能帮到你。我才 23 岁书读得不多不对的地方欢迎指正。