Claude Code人机交互日志追踪实战：从日志采集到行为分析全链路解析

1次阅读

共计 2935 个字符，预计需要花费 8 分钟才能阅读完成。

开发基于 Claude Code 的人机交互系统时，日志分析常遇到三个典型问题：

多源异构日志：前端点击流、API 请求日志、服务端处理日志分散在不同系统，时间戳和格式不统一
实时性要求：对话式交互需要秒级延迟的日志反馈，传统批处理模式无法满足
敏感信息泄露风险 ：用户输入可能包含身份证号、手机号等 PII(Personally Identifiable Information) 数据

通过装饰器自动记录函数调用信息，以下是核心实现（符合 PEP8 规范）：

import json
import time
from functools import wraps

def log_interaction(action_name: str):
    """
    人机交互日志记录装饰器
    时间复杂度：O(1) 不影响原函数性能
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start_time = time.perf_counter()
            result = func(*args, **kwargs)
            duration = (time.perf_counter() - start_time) * 1000  # 毫秒

            log_data = {"timestamp": int(time.time() * 1000),
                "action": action_name,
                "params": kwargs,
                "duration_ms": round(duration, 2),
                "status": "SUCCESS" if result else "FAILED"
            }

            # 异步写入日志（具体实现见避坑指南章节）log_queue.put(json.dumps(log_data))
            return result
        return wrapper
    return decorator

在微服务场景下，通过 TraceID 实现调用链追踪：

from opentelemetry import trace

tracer = trace.get_tracer(__name__)

@log_interaction("process_user_query")
def handle_query(user_input: str):
    with tracer.start_as_current_span("query_processing") as span:
        span.set_attribute("user_input", sanitize_input(user_input))
        # 业务逻辑处理...

使用 docker-compose 搭建日志分析平台（含健康检查）：

version: '3.8'
services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.7.0
    environment:
      - discovery.type=single-node
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9200"]
    ports:
      - "9200:9200"

  logstash:
    image: docker.elastic.co/logstash/logstash:8.7.0
    volumes:
      - ./logstash.conf:/usr/share/logstash/pipeline/logstash.conf
    depends_on:
      elasticsearch:
        condition: service_healthy

  kibana:
    image: docker.elastic.co/kibana/kibana:8.7.0
    ports:
      - "5601:5601"
    depends_on:
      elasticsearch:
        condition: service_healthy

使用正则表达式匹配并替换敏感内容：

import re

def sanitize_input(text: str) -> str:
    # 手机号脱敏
    text = re.sub(r'(1[3-9])\d{9}', r'\1******', text)
    # 身份证号脱敏
    text = re.sub(r'([1-9])\d{5}(19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[0-9Xx]', 
                 r'\1******\2***\3****', text)
    return text

采用异步队列缓解 I / O 压力：

import threading
import queue

log_queue = queue.Queue(maxsize=10000)

def log_worker():
    while True:
        try:
            log_data = log_queue.get()
            # 写入文件或发送到 Logstash
            with open('interaction.log', 'a') as f:
                f.write(log_data + '\n')
        except Exception as e:
            print(f"Log write failed: {str(e)}")

# 启动后台线程
threading.Thread(target=log_worker, daemon=True).start()

根据日志级别设置不同的 TTL(Time To Live)：

DEBUG 日志：保留 7 天
INFO 日志：保留 30 天
WARNING/ERROR 日志：保留 180 天

在 Elasticsearch 中配置 ILM(Index Lifecycle Management)策略实现自动滚动删除。

创建 Index Pattern 匹配claude-logs-*
添加可视化组件：
按 action 分组的柱状图
错误率变化的折线图
耗时百分位的热力图
设置过滤器排除测试环境数据

关键指标监控 SQL 示例：

SELECT
  action,
  avg(duration_ms) as avg_time,
  count(*) as total_requests
FROM interaction_logs
WHERE time > NOW() - INTERVAL '1 HOUR'
GROUP BY action
ORDER BY avg_time DESC