0Pricing
AI Agents · 课时

智能体的不可变操作日志

仅追加日志、加密签名和可检测篡改的审计轨迹

智能体的不可变操作日志 是 CoddyKit 上的免费 AI Agents 课时。 这是第 1 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

代理为什么需要不可变日志

代理会自主执行操作——通常不会有人审核每一步。当出现问题时，您需要回答：代理做了什么、何时做的、代表谁做的，以及为什么这样做？

不可变且可发现篡改的日志可以防止审计记录被事后修改。

日志条目结构

每个日志条目都会记录：时间戳、代理身份、用户身份、操作类型、参数、结果，以及将其连接到前一个条目的哈希值。这种结构同时支持搜索和篡改检测。

from dataclasses import dataclass
from typing import Any

@dataclass
class AuditEntry:
    timestamp:   float       # Unix time (UTC)
    agent_id:    str
    user_id:     str
    action_type: str
    parameters:  dict
    result:      dict
    session_id:  str
    prev_hash:   str         # hash of previous entry
    entry_hash:  str = ''    # computed after construction

if __name__ == '__main__':
    entry = AuditEntry(
        timestamp=1717000000.0, agent_id='agent-1', user_id='user-42',
        action_type='send_email', parameters={'to': 'a@b.com'},
        result={'status': 'sent'}, session_id='sess-1', prev_hash='0' * 64
    )
    print('Audit entry created:')
    print(' agent_id:', entry.agent_id)
    print(' action_type:', entry.action_type)
    print(' prev_hash:', entry.prev_hash)

用于篡改检测的哈希链

每个条目的哈希值都根据其内容加上前一个条目的哈希值计算得出。如果某个条目事后被修改，其哈希值就会发生变化，从而使后续所有哈希值失效。因此可以检测出篡改行为。

import hashlib, json, time

def compute_entry_hash(entry: dict, prev_hash: str) -> str:
    content = json.dumps(entry, sort_keys=True, default=str)
    payload = f'{prev_hash}:{content}'
    return hashlib.sha256(payload.encode()).hexdigest()

def create_log_entry(agent_id: str, user_id: str, action_type: str,
                     parameters: dict, result: dict,
                     session_id: str, prev_hash: str) -> dict:
    entry = {
        'timestamp':   time.time(),
        'agent_id':    agent_id,
        'user_id':     user_id,
        'action_type': action_type,
        'parameters':  parameters,
        'result':      result,
        'session_id':  session_id,
        'prev_hash':   prev_hash
    }
    entry['entry_hash'] = compute_entry_hash(
        {k: v for k, v in entry.items() if k != 'entry_hash'},
        prev_hash
    )
    return entry

if __name__ == '__main__':
    entry = create_log_entry('agent-1', 'user-42', 'send_email',
                              {'to': 'a@b.com'}, {'status': 'sent'},
                              'sess-1', prev_hash='0' * 64)
    print('New log entry hash:', entry['entry_hash'])
    print('Chained from prev_hash:', entry['prev_hash'])

进程内仅追加日志

最简单的实现方式是使用一个内存列表，并提供一个只允许追加内容的方法。不公开删除或更新方法。在生产环境中，这种实现由数据库或对象存储提供支持。

class AppendOnlyLog:
    def __init__(self):
        self._entries: list[dict] = []
        self._last_hash = 'GENESIS'

    def append(self, agent_id: str, user_id: str, action_type: str,
               parameters: dict, result: dict, session_id: str) -> dict:
        entry = create_log_entry(
            agent_id, user_id, action_type,
            parameters, result, session_id,
            self._last_hash
        )
        self._entries.append(entry)
        self._last_hash = entry['entry_hash']
        return entry

    def verify_integrity(self) -> bool:
        running_hash = 'GENESIS'
        for e in self._entries:
            expected = compute_entry_hash(
                {k: v for k, v in e.items() if k != 'entry_hash'},
                running_hash
            )
            if expected != e['entry_hash']:
                return False
            running_hash = e['entry_hash']
        return True

    @property
    def entries(self):
        return list(self._entries)  # return copy, not reference

使用追加语义写入 PostgreSQL

在生产环境中，应将日志写入数据库表，并且不要向应用程序用户授予 UPDATE 或 DELETE 权限。在数据库层面撤销这些权限，才能真正实现仅追加行为。

import psycopg2, json

# Table DDL (run once):
# CREATE TABLE agent_audit_log (
#   id          BIGSERIAL PRIMARY KEY,
#   timestamp   DOUBLE PRECISION NOT NULL,
#   agent_id    TEXT NOT NULL,
#   user_id     TEXT NOT NULL,
#   action_type TEXT NOT NULL,
#   parameters  JSONB NOT NULL,
#   result      JSONB NOT NULL,
#   session_id  TEXT NOT NULL,
#   prev_hash   TEXT NOT NULL,
#   entry_hash  TEXT NOT NULL UNIQUE
# );
# REVOKE UPDATE, DELETE ON agent_audit_log FROM app_user;

def write_to_db(conn, entry: dict):
    with conn.cursor() as cur:
        cur.execute(
            'INSERT INTO agent_audit_log '
            '(timestamp,agent_id,user_id,action_type,parameters,result,session_id,prev_hash,entry_hash) '
            'VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)',
            (entry['timestamp'], entry['agent_id'], entry['user_id'],
             entry['action_type'], json.dumps(entry['parameters']),
             json.dumps(entry['result']), entry['session_id'],
             entry['prev_hash'], entry['entry_hash'])
        )
    conn.commit()

使用对象锁（WORM）写入 S3

采用 COMPLIANCE 模式的 AWS S3 对象锁可以防止任何用户（包括账户根用户）在保留期限内删除或覆盖对象。这是真正用于审计日志的一次写入存储。

import boto3, json, time

s3 = boto3.client('s3', region_name='us-east-1')
BUCKET = 'your-audit-log-bucket-worm'

def write_to_s3_worm(entry: dict):
    key = f'audit/{entry["agent_id"]}/{int(entry["timestamp"])}/{entry["entry_hash"][:16]}.json'
    s3.put_object(
        Bucket=BUCKET,
        Key=key,
        Body=json.dumps(entry).encode(),
        ContentType='application/json',
        ObjectLockMode='COMPLIANCE',
        ObjectLockRetainUntilDate='2030-01-01T00:00:00Z'  # 4+ year retention
    )
    return key

查询审计日志

审计日志不仅用于合规，也具有实际的运维价值。您可以查询它来调试代理行为、重建会话，或回答“代理为什么执行了 X？”

def query_session(log: AppendOnlyLog, session_id: str) -> list[dict]:
    return [
        e for e in log.entries
        if e['session_id'] == session_id
    ]

def query_user_actions(log: AppendOnlyLog, user_id: str,
                       action_type: str = None) -> list[dict]:
    entries = [e for e in log.entries if e['user_id'] == user_id]
    if action_type:
        entries = [e for e in entries if e['action_type'] == action_type]
    return sorted(entries, key=lambda e: e['timestamp'])

def count_actions_by_type(log: AppendOnlyLog) -> dict:
    from collections import Counter
    return dict(Counter(e['action_type'] for e in log.entries))

记录日志前清理敏感数据

审计日志应该记录代理做了什么，而不是暴露敏感数据。写入前，请从参数和结果中移除 PII（密码、接口密钥、信用卡号）。

import re

SENSITIVE_KEYS = {'password', 'api_key', 'secret', 'token', 'credit_card', 'ssn'}

def sanitize(obj, depth: int = 0) -> dict | list | str:
    if depth > 5:
        return '[MAX_DEPTH]'
    if isinstance(obj, dict):
        return {
            k: '[REDACTED]' if k.lower() in SENSITIVE_KEYS
               else sanitize(v, depth + 1)
            for k, v in obj.items()
        }
    elif isinstance(obj, list):
        return [sanitize(i, depth + 1) for i in obj]
    elif isinstance(obj, str):
        # Redact anything that looks like an API key
        return re.sub(r'(sk-|Bearer\s)[A-Za-z0-9_-]{16,}', '[REDACTED]', obj)
    return obj

if __name__ == '__main__':
    record = {'user': 'alice', 'password': 'hunter2', 'note': 'call me at sk-abcdefghijklmnopqrstuv'}
    print('Sanitized record:', sanitize(record))

日志轮换与归档

活动日志会无限增长。请对其进行轮换：将超过 90 天的条目移至冷存储（例如 S3 Glacier），并进行压缩。通过保留每个轮换片段的最后一个哈希值，确保哈希链保持完整。

import time, json, gzip

def rotate_log(log: AppendOnlyLog, max_age_days: int = 90) -> dict:
    cutoff = time.time() - max_age_days * 86400
    archive = [e for e in log.entries if e['timestamp'] < cutoff]
    remaining = [e for e in log.entries if e['timestamp'] >= cutoff]

    if not archive:
        return {'archived': 0, 'remaining': len(remaining)}

    # Compress archive
    archive_bytes = gzip.compress(json.dumps(archive).encode())
    archive_file  = f'/tmp/audit_archive_{int(time.time())}.json.gz'
    with open(archive_file, 'wb') as f:
        f.write(archive_bytes)

    # Update in-memory log
    log._entries = remaining

    return {'archived': len(archive), 'remaining': len(remaining), 'file': archive_file}

针对可疑模式发出警报

实时监控审计日志中的异常模式：短时间内大量执行高风险操作、在工作时间之外执行操作，或在较短时间窗口内重复执行同一操作超过 N 次。

import time
from collections import defaultdict

HIGH_RISK_ACTIONS = {'delete_user', 'send_mass_email',
                     'transfer_funds', 'export_all_data'}
BURST_LIMIT = 5
BURST_WINDOW = 60  # seconds

action_timestamps: dict[str, list] = defaultdict(list)

def check_suspicious(entry: dict) -> list[str]:
    alerts = []
    action = entry['action_type']

    if action in HIGH_RISK_ACTIONS:
        alerts.append(f'HIGH_RISK_ACTION: {action} by {entry["agent_id"]}')

    now = time.time()
    action_timestamps[action].append(now)
    recent = [t for t in action_timestamps[action] if now - t < BURST_WINDOW]
    action_timestamps[action] = recent

    if len(recent) > BURST_LIMIT:
        alerts.append(f'ACTION_BURST: {action} called {len(recent)}x in {BURST_WINDOW}s')

    return alerts

if __name__ == '__main__':
    entry = {'action_type': 'delete_user', 'agent_id': 'agent-9'}
    for _ in range(6):
        alerts = check_suspicious(entry)
    print('Alerts on 6th call:', alerts)

按需验证日志完整性

将完整性检查作为监控流水线的一部分。如果检测到任何验证失败，请立即发出警报——这表明日志代码中可能存在错误，或有人正在主动尝试篡改日志。

def verify_and_alert(log: AppendOnlyLog) -> dict:
    is_valid = log.verify_integrity()
    total    = len(log.entries)
    result   = {
        'total_entries': total,
        'integrity_ok':  is_valid
    }
    if not is_valid:
        import logging
        alert_logger = logging.getLogger('agent.integrity_alert')
        alert_logger.critical(
            'AUDIT LOG INTEGRITY FAILURE: tampered or corrupted entries detected. '
            'Total entries: %d. Initiating incident response.', total
        )
        result['alert_sent'] = True
    return result

# Run periodically via cron or monitoring hook
# verify_and_alert(global_audit_log)

审计日志中的哈希链能检测什么

哈希链是一种密码学机制，使审计日志具备可发现篡改的特性。理解它能检测什么，是设计审计日志的基础。

不可变操作日志回顾

不可变代理日志使用：通过哈希链检测篡改；使用仅追加存储（在数据库层面撤销 UPDATE/DELETE 权限）；使用WORM 对象存储（S3 对象锁）满足法规规定的保留要求；在写入前进行PII 脱敏；以及针对可疑操作模式进行实时异常警报。

常见问题解答

「智能体的不可变操作日志」课时是免费的吗？

是的 — 「智能体的不可变操作日志」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「智能体的不可变操作日志」这节课中我会学到什么？

仅追加日志、加密签名和可检测篡改的审计轨迹 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 1 节课，共 4 节。

「智能体的不可变操作日志」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 智能体的不可变操作日志
  2. 执行智能体操作策略
  3. 法规合规：GDPR 与 SOC2
  4. 人在回路中的审批关卡
← 返回 AI Agents