WeClaw_40_系统监控与日志体系：多层次日志架构与Trace追踪

👋 Hi，带娃的我热爱 AI 大模型应用落地、意识解码与 AI 开发工具链 。 💡 创业路上，用技术换时间，一起把 AI 变成生产力 🚀 >

WeClaw_40_系统监控与日志体系：多层次日志架构与Trace追踪

作者: WeClaw 开发团队
日期: 2026-03-25
版本: v1.0
标签: 日志系统、Trace 追踪、监控、可观测性、敏感信息脱敏

📖 摘要

本文深入剖析 WeClaw 系统监控与日志体系的完整设计与实现。针对 AI Agent 系统的可观测性需求，我们展示了如何构建多层次、全链路的监控系统。文章涵盖 Python logging 模块最佳实践、TaskTraceCollector 全链路追踪、敏感信息脱敏、JSONL 日志持久化等核心技术实践。

核心收获：

• 📝 掌握 Python logging 模块最佳配置
• 🔍 理解全链路 Trace 追踪架构
• 🛡️ 学会敏感信息脱敏技术
• 💾 掌握 JSONL 日志文件管理
• 📊 获得性能分析与问题诊断方法

🎯 需求背景：为什么需要监控与日志？

AI Agent 系统的可观测性挑战

在 WeClaw 系统中，每次用户请求都涉及多个复杂环节：

用户输入 → 意图识别 → 工具暴露 → 模型调用 → 工具执行 → 响应生成
↓ ↓ ↓ ↓ ↓ ↓
记录日志 记录日志 记录日志 记录日志 记录日志 记录日志

问题：

• ❌ 请求链路长，定位问题困难
• ❌ 工具调用失败原因不明
• ❌ 性能瓶颈无法量化
• ❌ 敏感信息泄露风险
• ❌ 日志分散难以分析

我们的解决方案

三层监控体系：

┌─────────────────────────────────────────────────────┐
│ 监控层 │
│ ┌─────────────────┐ ┌─────────────────────────┐ │
│ │ TaskTrace │ │ Event Bus │ │
│ │ 全链路追踪 │ │ 实时事件监控 │ │
│ └────────┬────────┘ └───────────┬─────────────┘ │
│ │ │ │
│ └───────────┬───────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────┐ │
│ │ 日志层（Python logging） │ │
│ └─────────────────────────────────────────────┘ │
│ │ │
│ ┌─────────────────────────────────────────────┐ │
│ │ 持久化层（JSONL 文件） │ │
│ └─────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────┘

📝 核心模块一：Python Logging 最佳实践

日志配置

import logging
import sys
from pathlib import Path

def setup_logging(
level: int = logging.INFO,
log_file: str = None,
format_string: str = None,
) –> None:
"""配置日志系统。

Args:
level: 日志级别
log_file: 日志文件路径（可选）
format_string: 日志格式
"""
if format_string is None:
format_string = "%(asctime)s │ %(levelname)-8s │ %(name)s │ %(message)s"

# 创建 formatter
formatter = logging.Formatter(format_string)

# 配置根 logger
root_logger = logging.getLogger()
root_logger.setLevel(level)

# 清除现有 handlers
root_logger.handlers.clear()

# 控制台 handler
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(level)
console_handler.setFormatter(formatter)
root_logger.addHandler(console_handler)

# 文件 handler（可选）
if log_file:
log_path = Path(log_file)
log_path.parent.mkdir(parents=True, exist_ok=True)

file_handler = logging.FileHandler(log_path, encoding="utf-8")
file_handler.setLevel(level)
file_handler.setFormatter(formatter)
root_logger.addHandler(file_handler)

# 初始化
setup_logging(
level=logging.DEBUG if os.getenv("DEBUG") else logging.INFO,
log_file="~/.winclaw/logs/weclaw.log",
)

模块级 Logger

# 在每个模块中创建 logger
logger = logging.getLogger(__name__)

class MyClass:
"""示例类。"""

def __init__(self):
# 使用类名作为 logger 名
self._logger = logging.getLogger(self.__class__.__name__)

def do_something(self):
# 使用 logger 记录日志
self._logger.info("开始执行操作")

try:
result = self._process()
self._logger.debug("处理结果：%s", result)
return result
except Exception as e:
self._logger.error("操作失败：%s", e, exc_info=True)
raise

日志级别规范

# 日志级别使用规范

# DEBUG: 详细调试信息
logger.debug("请求参数：%s", params)
logger.debug("函数入口，变量值：x=%d, y=%s", x, y)

# INFO: 正常流程信息
logger.info("用户登录成功：%s", username)
logger.info("工具调用：%s.%s", tool_name, action)

# WARNING: 警告信息（不影响流程）
logger.warning("配置缺失，使用默认值：%s", default_value)
logger.warning("API 限流，降低请求频率")

# ERROR: 错误信息（影响功能）
logger.error("数据库连接失败：%s", e)
logger.error("工具执行超时：%s", tool_name)

# CRITICAL: 严重错误（可能导致崩溃）
logger.critical("内存耗尽，程序即将退出")
logger.critical("无法连接到关键服务")

🔍 核心模块二：TaskTraceCollector 全链路追踪

追踪数据模型

from dataclasses import dataclass, field, asdict
from typing import Any

@dataclass
class ToolCallRecord:
"""单次工具调用记录。"""

step: int # 第几步
function_name: str # 如 "browser_use_run_task"
arguments: dict[str, Any] # 调用参数
status: str # success / error / timeout / denied
duration_ms: float # 执行时长
error: str = "" # 错误信息
output_preview: str = "" # 输出前 N 字符

@dataclass
class TaskTrace:
"""一次用户请求的完整追踪记录。"""

trace_id: str # UUID
session_id: str # 会话 ID
timestamp: str # ISO 时间
user_input: str # 原始用户输入

# — 意图识别阶段 —
intent_primary: str = "" # 主要意图
intent_all: list[str] = field(default_factory=list) # 所有匹配意图
intent_confidence: float = 0.0 # 置信度
matched_keywords: dict[str, list[str]] = field(default_factory=dict)

# — 工具暴露阶段 —
tool_tier: str = "" # recommended / extended / full
tools_exposed: list[str] = field(default_factory=list)
tools_exposed_count: int = 0

# — 执行阶段 —
total_steps: int = 0
tool_calls: list[ToolCallRecord] = field(default_factory=list)
consecutive_failures_max: int = 0
tier_upgrades: list[str] = field(default_factory=list)

# — 结果阶段 —
final_status: str = "" # completed / max_steps / error / cancelled
total_tokens: int = 0
total_duration_ms: float = 0.0
assistant_response_preview: str = ""

追踪采集器实现

import uuid
import time
from datetime import datetime
from pathlib import Path

class TaskTraceCollector:
"""追踪采集器 — 绑定一次 chat/chat_stream 调用的完整生命周期。

用法::
collector = TaskTraceCollector(session_id, user_input)
collector.set_intent(intent_result, tier, exposed_tools)

# 每次工具调用
collector.add_tool_call(step, func_name, args, result)

# 层级升级（如有）
collector.add_tier_upgrade("recommended", "extended")

# 结束
collector.finalize(status="completed", tokens=500, response_preview="…")
collector.flush() # 写入文件
"""

def __init__(
self,
session_id: str,
user_input: str,
trace_dir: Path | None = None,
max_output_preview: int = 200,
max_trace_days: int = 30,
enabled: bool = True,
):
self._trace = TaskTrace(
trace_id=str(uuid.uuid4())[:8],
session_id=session_id,
timestamp=datetime.now().isoformat(),
user_input=user_input,
)
self._trace_dir = trace_dir or DEFAULT_TRACE_DIR
self._max_output_preview = max_output_preview
self._max_trace_days = max_trace_days
self._enabled = enabled
self._start_time = time.perf_counter()
self._consecutive_failures = 0
self._finalized = False

追踪数据采集

def set_intent(
self,
intent_result: Any,
tier: str,
exposed_tools: list[str],
) –> None:
"""设置意图识别结果。"""
self._trace.intent_primary = getattr(intent_result, "primary_intent", "")
self._trace.intent_all = list(getattr(intent_result, "intents", set()))
self._trace.intent_confidence = getattr(intent_result, "confidence", 0.0)
self._trace.matched_keywords = {
k: v for k, v in getattr(intent_result, "matched_keywords", {}).items()
}
self._trace.tool_tier = tier
self._trace.tools_exposed = list(exposed_tools)
self._trace.tools_exposed_count = len(exposed_tools)

def add_tool_call(
self,
step: int,
function_name: str,
arguments: dict[str, Any],
status: str,
duration_ms: float,
error: str = "",
output: str = "",
) –> None:
"""添加一次工具调用记录。"""
# 更新连续失败计数
if status in ("error", "timeout", "denied"):
self._consecutive_failures += 1
if self._consecutive_failures > self._trace.consecutive_failures_max:
self._trace.consecutive_failures_max = self._consecutive_failures
else:
self._consecutive_failures = 0

record = ToolCallRecord(
step=step,
function_name=function_name,
arguments=arguments,
status=status,
duration_ms=duration_ms,
error=error,
output_preview=output[:self._max_output_preview] if output else "",
)
self._trace.tool_calls.append(record)
self._trace.total_steps = step

def add_tier_upgrade(self, from_tier: str, to_tier: str) –> None:
"""记录层级升级。"""
self._trace.tier_upgrades.append(f"{from_tier}->{to_tier}")
self._trace.tool_tier = to_tier

def finalize(
self,
status: str,
tokens: int = 0,
response_preview: str = "",
) –> None:
"""完成追踪记录。"""
if self._finalized:
return

self._trace.final_status = status
self._trace.total_tokens = tokens
self._trace.total_duration_ms = (time.perf_counter() – self._start_time) * 1000
self._trace.assistant_response_preview = response_preview[:300] if response_preview else ""
self._finalized = True

🛡️ 核心模块三：敏感信息脱敏

敏感信息定义

# 敏感参数名列表
SENSITIVE_PARAM_NAMES = {
"api_key", "apikey", "api-key",
"password", "passwd", "pwd",
"token", "access_token", "refresh_token", "auth_token",
"secret", "secret_key", "client_secret",
"credential", "credentials",
"private_key", "privatekey",
"authorization", "auth",
}

# 敏感值模式（正则）
SENSITIVE_VALUE_PATTERNS = [
# API Key 格式（超过 20 字符的字母数字混合）
re.compile(r'\\b[a-zA-Z0-9_-]{20,}\\b'),
# 明显的 key=xxx 模式
re.compile(r'(api_key|apikey|api-key|token|secret|password)\\s*[=:]\\s*["\\']?[\\w-]{10,}["\\']?', re.I),
]

脱敏函数实现

def _sanitize_dict(d: dict[str, Any], max_preview: int = 200) –> dict[str, Any]:
"""对字典中的敏感参数值进行脱敏。"""
result = {}
for key, value in d.items():
key_lower = key.lower().replace("-", "_")

# 检查是否是敏感参数名
if key_lower in SENSITIVE_PARAM_NAMES:
result[key] = "***"
elif isinstance(value, dict):
# 递归处理嵌套字典
result[key] = _sanitize_dict(value, max_preview)
elif isinstance(value, str):
result[key] = _sanitize_string(value, max_preview)
else:
result[key] = value

return result

def _sanitize_string(s: str, max_len: int = 200) –> str:
"""对字符串中的敏感信息进行脱敏。"""
if not s:
return s

# 截断过长的字符串
if len(s) > max_len:
s = s[:max_len] + "…"

# 替换敏感模式
for pattern in SENSITIVE_VALUE_PATTERNS:
s = pattern.sub("***", s)

return s

def _sanitize_trace(self) –> dict[str, Any]:
"""对追踪记录进行脱敏处理。"""
trace_dict = self._trace.to_dict()

# 脱敏 user_input
trace_dict["user_input"] = _sanitize_string(
trace_dict.get("user_input", ""), max_len=1000
)

# 脱敏 tool_calls 中的 arguments 和 output_preview
for tc in trace_dict.get("tool_calls", []):
if isinstance(tc, dict):
if tc.get("arguments"):
tc["arguments"] = _sanitize_dict(tc["arguments"])
if tc.get("output_preview"):
tc["output_preview"] = _sanitize_string(tc["output_preview"])

# 脱敏 assistant_response_preview
if trace_dict.get("assistant_response_preview"):
trace_dict["assistant_response_preview"] = _sanitize_string(
trace_dict["assistant_response_preview"], max_len=300
)

return trace_dict

脱敏效果示例

# 输入（原始参数）
{
"api_key": "sk-abc123xyz456def789",
"password": "MySecretPass123",
"query": "今天天气怎么样",
"token": "eyJhbGciOiJIUzI1NiJ9…",
"user_id": "12345"
}

# 输出（脱敏后）
{
"api_key": "***",
"password": "***",
"query": "今天天气怎么样",
"token": "***",
"user_id": "12345"
}

💾 核心模块四：JSONL 日志持久化

JSONL 格式说明

JSONL（JSON Lines）是一种适合流式写入的日志格式：

{"trace_id": "a1b2c3d4", "session_id": "sess123", "timestamp": "2026-03-25T10:30:00", …}
{"trace_id": "e5f6g7h8", "session_id": "sess124", "timestamp": "2026-03-25T10:31:00", …}
{"trace_id": "i9j0k1l2", "session_id": "sess125", "timestamp": "2026-03-25T10:32:00", …}

优势：

• ✅ 每行独立，可流式追加
• ✅ 支持 append 模式
• ✅ 便于分批读取
• ✅ 比 JSON 数组更节省内存

持久化实现

def flush(self) –> bool:
"""将追踪记录写入 JSONL 文件。"""
if not self._enabled or not self._finalized:
return False

try:
# 确保目录存在
self._trace_dir.mkdir(parents=True, exist_ok=True)

# 清理过期文件
self._cleanup_old_traces()

# 脱敏处理
sanitized = self._sanitize_trace()

# 写入 JSONL
today = datetime.now().strftime("%Y-%m-%d")
trace_file = self._trace_dir / f"trace-{today}.jsonl"

with open(trace_file, "a", encoding="utf-8") as f:
f.write(json.dumps(sanitized, ensure_ascii=False) + "\\n")

logger.debug("TaskTrace 已写入: %s (id=%s)", trace_file, self._trace.trace_id)
return True

except Exception as e:
logger.error("写入 TaskTrace 失败: %s", e)
return False

def _cleanup_old_traces(self) –> None:
"""清理过期的 trace 文件。"""
if self._max_trace_days <= 0:
return

try:
cutoff = datetime.now() – timedelta(days=self._max_trace_days)

for f in self._trace_dir.glob("trace-*.jsonl"):
# 从文件名解析日期
try:
date_str = f.stem.replace("trace-", "")
file_date = datetime.strptime(date_str, "%Y-%m-%d")

if file_date < cutoff:
f.unlink()
logger.debug("已清理过期 trace 文件: %s", f)
except ValueError:
pass # 文件名格式不匹配，跳过
except Exception as e:
logger.warning("清理过期 trace 文件失败: %s", e)

日志文件结构

~/.winclaw/traces/
├── trace-2026-03-20.jsonl # 3 月 20 日的追踪记录
├── trace-2026-03-21.jsonl # 3 月 21 日的追踪记录
├── trace-2026-03-22.jsonl # 3 月 22 日的追踪记录
├── trace-2026-03-23.jsonl # 3 月 23 日的追踪记录
├── trace-2026-03-24.jsonl # 3 月 24 日的追踪记录
└── trace-2026-03-25.jsonl # 3 月 25 日的追踪记录（今天）

📊 核心模块五：Trace 日志分析

读取和分析工具

from pathlib import Path
import json
from collections import Counter, defaultdict

class TraceAnalyzer:
"""Trace 日志分析器。"""

def __init__(self, trace_dir: Path):
self._trace_dir = trace_dir

def load_traces(self, date: str = None) –> list[dict]:
"""加载指定日期的追踪记录。

Args:
date: 日期字符串（YYYY-MM-DD），None 表示今天

Returns:
追踪记录列表
"""
if date is None:
date = datetime.now().strftime("%Y-%m-%d")

trace_file = self._trace_dir / f"trace-{date}.jsonl"

if not trace_file.exists():
return []

traces = []
with open(trace_file, "r", encoding="utf-8") as f:
for line in f:
if line.strip():
traces.append(json.loads(line))

return traces

def analyze_tool_usage(self, traces: list[dict]) –> dict:
"""分析工具使用情况。"""
tool_counts = Counter()
tool_errors = Counter()
tool_durations = defaultdict(list)

for trace in traces:
for tc in trace.get("tool_calls", []):
func_name = tc.get("function_name", "unknown")
tool_counts[func_name] += 1

if tc.get("status") != "success":
tool_errors[func_name] += 1

tool_durations[func_name].append(tc.get("duration_ms", 0))

# 计算平均时长
avg_durations = {
name: sum(durations) / len(durations)
for name, durations in tool_durations.items()
}

return {
"tool_counts": dict(tool_counts.most_common(10)),
"tool_errors": dict(tool_errors.most_common(10)),
"avg_durations": avg_durations,
}

def analyze_intent_accuracy(self, traces: list[dict]) –> dict:
"""分析意图识别准确率。"""
total = len(traces)
high_confidence = sum(1 for t in traces if t.get("intent_confidence", 0) >= 0.8)
tier_upgrades = sum(1 for t in traces if t.get("tier_upgrades"))

return {
"total_traces": total,
"high_confidence_count": high_confidence,
"high_confidence_rate": high_confidence / total if total > 0 else 0,
"tier_upgrade_count": tier_upgrades,
"tier_upgrade_rate": tier_upgrades / total if total > 0 else 0,
}

def analyze_performance(self, traces: list[dict]) –> dict:
"""分析性能指标。"""
durations = [t.get("total_duration_ms", 0) for t in traces]
tokens = [t.get("total_tokens", 0) for t in traces]

return {
"avg_duration_ms": sum(durations) / len(durations) if durations else 0,
"max_duration_ms": max(durations) if durations else 0,
"min_duration_ms": min(durations) if durations else 0,
"avg_tokens": sum(tokens) / len(tokens) if tokens else 0,
"total_tokens": sum(tokens),
}

分析报告生成

def generate_report(date: str = None) –> str:
"""生成追踪分析报告。"""
analyzer = TraceAnalyzer(Path.home() / ".winclaw" / "traces")
traces = analyzer.load_traces(date)

if not traces:
return f"没有找到 {date} 的追踪记录"

# 收集统计
tool_stats = analyzer.analyze_tool_usage(traces)
intent_stats = analyzer.analyze_intent_accuracy(traces)
perf_stats = analyzer.analyze_performance(traces)

# 生成报告
report_lines = [
f"# Trace 分析报告 – {date or '今天'}",
"",
"## 概览",
f"- 总追踪数：{intent_stats['total_traces']}",
f"- 高置信度占比：{intent_stats['high_confidence_rate']:.1%}",
f"- 层级升级占比：{intent_stats['tier_upgrade_rate']:.1%}",
"",
"## 性能指标",
f"- 平均响应时间：{perf_stats['avg_duration_ms']:.0f}ms",
f"- 最大响应时间：{perf_stats['max_duration_ms']:.0f}ms",
f"- 平均 Token 数：{perf_stats['avg_tokens']:.0f}",
"",
"## 工具使用 TOP 10",
]

for tool, count in tool_stats["tool_counts"].items():
error_count = tool_stats["tool_errors"].get(tool, 0)
avg_dur = tool_stats["avg_durations"].get(tool, 0)
report_lines.append(
f"- {tool}: {count}次 (错误:{error_count}, 平均:{avg_dur:.0f}ms)"
)

return "\\n".join(report_lines)

📈 核心模块六：Event Bus 实时监控

事件类型定义

class EventType:
"""事件类型枚举。"""

# 模型调用
MODEL_CALL = "model_call"
MODEL_RESPONSE = "model_response"
MODEL_REASONING = "model_reasoning"
MODEL_USAGE = "model_usage"

# 工具调用
TOOL_CALL = "tool_call"
TOOL_RESULT = "tool_result"

# CFTA: 异步工具执行
DEFERRED_TOOL_STARTED = "deferred_tool_started"
DEFERRED_TOOL_RESULT = "deferred_tool_result"

@dataclass
class ToolCallEvent:
"""工具调用事件数据。"""
tool_name: str
action_name: str
arguments: dict[str, Any]
function_name: str
session_id: str = ""

@dataclass
class ToolResultEvent:
"""工具结果事件数据。"""
tool_name: str
action_name: str
status: str
output: str
error: Optional[str]
duration_ms: float
session_id: str = ""

@dataclass
class ModelUsageEvent:
"""模型用量事件。"""
model_key: str
prompt_tokens: int
completion_tokens: int
total_tokens: int
cost: float
session_id: str = ""

事件总线监控

class EventMonitor:
"""事件监控器。"""

def __init__(self, event_bus: EventBus):
self._event_bus = event_bus
self._stats = defaultdict(int)
self._callbacks = {}

def start(self) –> None:
"""开始监控。"""
# 订阅所有事件
for event_type in [
EventType.MODEL_CALL,
EventType.MODEL_RESPONSE,
EventType.TOOL_CALL,
EventType.TOOL_RESULT,
]:
sub_id = self._event_bus.on(event_type, self._on_event)
self._callbacks[event_type] = sub_id

logger.info("事件监控已启动")

def stop(self) –> None:
"""停止监控。"""
for event_type, sub_id in self._callbacks.items():
self._event_bus.off(event_type, sub_id)
self._callbacks.clear()
logger.info("事件监控已停止")

async def _on_event(self, data) –> None:
"""事件处理。"""
event_type = type(data).__name__
self._stats[event_type] += 1

# 打印关键事件
if isinstance(data, ToolCallEvent):
logger.info(
"🔧 工具调用: %s.%s",
data.tool_name, data.action_name
)
elif isinstance(data, ToolResultEvent):
if data.status == "success":
logger.info(
"✅ 工具结果: %s.%s (%.0fms)",
data.tool_name, data.action_name, data.duration_ms
)
else:
logger.warning(
"❌ 工具失败: %s.%s – %s",
data.tool_name, data.action_name, data.error
)

def get_stats(self) –> dict:
"""获取统计信息。"""
return dict(self._stats)

📊 测试验证

日志测试

Trace 追踪测试

性能测试

💡 经验教训

1. 日志级别的选择

教训：初期全部使用 INFO，导致日志过多难以排查问题。

解决方案：严格区分日志级别

# DEBUG: 仅开发调试使用
# INFO: 正常流程关键节点
# WARNING: 非关键异常
# ERROR: 影响功能的错误

2. 敏感信息泄露

教训：日志中意外暴露了 API Key。

解决方案：多层脱敏

# 1. 参数名检查
if key_lower in SENSITIVE_PARAM_NAMES:
result[key] = "***"

# 2. 参数值正则检查
for pattern in SENSITIVE_VALUE_PATTERNS:
s = pattern.sub("***", s)

3. JSONL 文件过大

教训：日志文件无限制增长，占用大量磁盘。

解决方案：过期自动清理

cutoff = datetime.now() – timedelta(days=self._max_trace_days)
for f in self._trace_dir.glob("trace-*.jsonl"):
if file_date < cutoff:
f.unlink()

4. Trace 采集的性能影响

教训：每个工具调用都记录 Trace，影响性能。

解决方案：异步写入 + 批量处理

# 采集在主线程
collector.add_tool_call(...)

# 写入在后台线程
asyncio.create_task(self._flush_async())

📊 架构总结

完整监控数据流

用户请求
↓
Agent 处理
├─ 记录 Trace（意图/工具/性能）
└─ 发布 Event（事件总线）
↓
┌───────┴───────┐
↓ ↓
TaskTraceCollector EventMonitor
↓ ↓
JSONL 文件 实时统计
↓
TraceAnalyzer
↓
分析报告

关键技术点

字数统计: 约 5,800 字
阅读时间: 约 15 分钟
代码行数: 约 450 行

🎉 里程碑达成：WeClaw 技术博客系列已完成 40 篇！

系列总结：

• 总字数：约 72,500 字
• 总代码行数：约 5,000 行
• 覆盖领域：AI 对话、语音交互、文档处理、食谱推荐、家庭管理、音乐播放、异步优化、远程桥接、监控系统

技术栈全景：

• 🔧 Python: asyncio、SQLite、JSON、logging
• 🎨 Qt: PySide6、QtMultimedia
• 🤖 AI: GLM-4.6V、Whisper、TTS
• 🌐 协议: MCP、iCalendar
• 📊 监控: Trace、EventBus、日志分析