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

我把审计日志分成四个维度：

三、实现：用 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: Option<serde_json::Value>,
    
    /// 是否成功（None = 未完成，Some(true) = 成功，Some(false) = 失败）
    pub success: Option<bool>,
    
    /// 错误信息（如果有的话）
    pub error_message: Option<String>,
    
    /// 工具调用的开始时间（毫秒时间戳）
    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: Option<u64>,
    
    /// 任务状态：running / completed / failed
    pub status: String,
    
    /// 所有工具调用的记录（按时间顺序）
    pub tool_calls: Vec<ToolCallRecord>,
    
    /// 总的 Token 消耗
    pub total_tokens: u64,
    
    /// Agent 的最终输出
    pub final_output: Option<String>,
}

/// 审计记录器（全局单例或请求级实例）
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_call<F, Fut, T>(
        &mut self,
        tool_name: &str,
        input_params: serde_json::Value,
        f: F,
    ) -> Result<T, String>
    where
        F: FnOnce() -> Fut,
        Fut: std::future::Future<Output = Result<T, 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["结构化日志\n（JSON 格式）"]
    
    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 可能陷入了推理循环

五、总结

审计日志这件事，说简单也简单，说重要也重要：

1. Agent 必须有可审计的操作记录。它不像传统程序——代码有确定的执行路径。Agent 的"决策"是黑箱，你必须至少知道它调了什么工具、传了什么参数。
2. 用包装器模式比在每个函数里加日志好得多。业务逻辑和审计逻辑分离，代码干净，也不会有遗漏。
3. 记录什么要提前想清楚。入参、出参、耗时、错误——这四个维度是最基本的。Token 消耗和 LLM 延迟是进阶的。
4. 存储要结构化（JSON）。未来你需要按工具名称、时间段、Token 消耗做过滤和聚合，这些分析能力来自于良好的数据结构设计。
5. 日志不是存了就完了。至少要配几条基本的 SQL 和告警规则。否则出了问题你还是得去翻日志，而不是被日志主动告知出了问题。

其实这些都是我在实战中摔打出来的教训。刚开始我以为审计日志就是"打个日志"，后来发现它是一套完整的可观测性体系——采集、存储、查询、告警——缺一不可。

如果你也在做 Agent 相关的东西，希望我的经验能帮到你。我才 23 岁，书读得不多，不对的地方欢迎指正。