返回博客列表
最佳实践2026-06-2022 分钟阅读

熵管理：AI系统的垃圾回收机制

Entropy Manager · 后台清理 · 文档一致性 · 死代码扫描 · 技能花园维护

WeClaw_65_熵管理：AI系统的垃圾回收机制

第四季系列文章第 4 篇（总第 65 篇） - Entropy Manager · 后台清理 · 文档一致性 · 死代码扫描 · 技能花园维护

📚 专栏信息

《从零到一构建跨平台 AI 助手：WeClaw 实战指南》专栏 · 第四季

专栏定位：面向开发者和技术决策者的实战专栏，用真实案例和完整代码带你理解如何构建生产级 AI 应用

本文深入探讨 Harness Engineering 七大差距中的 G3——熵管理（Garbage Collection）。软件系统在使用过程中不可避免地产生"熵增"：过时的文档、冗余的技能、累积的日志、未引用的代码。本文设计 EntropyManager——一个后台低优先级运行的 Agent，在系统空闲时自动执行四类清理任务：死代码扫描、文档一致性检查、技能花园维护、旧轨迹清理。

👨‍💻 作者与项目

作者简介：翁勇刚 WENG YONGGANG 新概念龙虾-WeClaw 开发团队负责人，一群专注于跨平台 AI 应用的实践者 理念："再复杂的技术，也能用代码讲清楚"

📝 摘要

本文结构概览： 从软件系统的"熵增定律"出发——任何运行中的系统都会不可逆地产生信息熵增：文档与代码不一致、技能库膨胀、日志文件堆积、死代码残留。然后设计 EntropyManager：一个以后台 asyncio.Task 运行的低优先级 Agent，监听系统空闲信号（最近 5 分钟无用户请求），执行四类清理任务。重点讲解调度策略（空闲触发 + 优雅停止）、每类检查的实现思路，以及如何复用 WeClaw 现有的 AgentPool._cleanup_loop 调度模式。

核心问题： WeClaw 的 Curator 已经实现了技能自动沉淀和维护（813行），但系统的其他维度——文档一致性、死代码扫描、旧轨迹清理——尚无自动化机制。随着系统运行时间增长，这些"熵"会逐渐降低 Agent 的执行质量。

关键成果：

  • 设计了 EntropyManager 的四类检查任务
  • 空闲触发 + 优雅停止的调度策略
  • 每个检查项 <500ms，对主对话零干扰

适合读者：关注系统长期维护和自动化的开发者、运维工程师

阅读时长：约 12 分钟

关键词：Entropy Manager、垃圾回收、后台调度、文档一致性、死代码扫描、asyncio.Task

一、软件熵增定律

1.1 什么是"熵"？

在信息论中，"熵"衡量系统的不确定性。在软件工程中，我们借用这个概念来描述系统运行过程中产生的信息混乱：

熵增类型表现影响
文档漂移README 说的接口和代码实际不一致Agent 依赖过时文档做决策
技能膨胀Curator 自动沉淀的技能越来越多，大量重复或过时检索效率下降，噪音增加
轨迹堆积TaskTrace 日志文件持续增长磁盘空间、检索速度
死代码残留重构后旧函数未被清理Agent 可能引用不存在的代码

1.2 为什么 Agent 系统特别需要熵管理？

传统软件的熵增主要影响可维护性——开发者读代码时感到困惑。但 Agent 系统的熵增直接影响运行时质量：

Agent 读取 README → README 说的接口已过时 → Agent 生成了错误代码 → 用户困惑

Agent 不像人类开发者能"意识到文档可能过时"。它信任上下文中的所有信息。如果上下文中有 30% 是过时的，Agent 的输出质量就会下降 30%。

1.3 现状：Curator 的覆盖范围

WeClaw 的 Curator（813行）已经实现了技能层面的熵管理：

  • 对话→技能自动沉淀
  • 周期性维护（相似技能合并、过时技能清理）
  • 技能使用频率统计

但其他三个维度——文档一致性、死代码扫描、轨迹清理——尚无自动化机制。

二、EntropyManager 设计

2.1 核心架构

class EntropyManager:
    """后台熵管理 — 定期清理系统熵增。"""
    
    CHECKS = {
        "dead_code_scan": _scan_dead_code,
        "doc_consistency": _check_doc_consistency,
        "skill_garden": _tend_skill_garden,
        "trace_cleanup": _cleanup_old_traces,
    }
    
    def __init__(self, event_bus: EventBus):
        self._event_bus = event_bus
        self._stop_event = asyncio.Event()
        self._running_task: asyncio.Task | None = None
        self._last_user_activity = time.time()
    
    async def start(self):
        """启动后台调度循环。"""
        self._running_task = asyncio.create_task(self._schedule_loop())
    
    async def stop(self):
        """优雅停止。设置停止信号并等待当前周期完成。"""
        self._stop_event.set()
        if self._running_task:
            await self._running_task
    
    async def run_cycle(self):
        """执行一轮熵管理（在低负载时触发）。"""

2.2 调度策略

EntropyManager 的调度遵循三个原则：

原则1：空闲触发

async def _schedule_loop(self):
    while not self._stop_event.is_set():
        # 等待 60 秒
        try:
            await asyncio.wait_for(
                self._stop_event.wait(), timeout=60
            )
            break  # 收到停止信号
        except asyncio.TimeoutError:
            pass
        
        # 检查是否空闲（最近 5 分钟无用户请求）
        idle_time = time.time() - self._last_user_activity
        if idle_time < 300:  # 5 分钟
            continue
        
        # 执行一轮清理
        await self.run_cycle()

原则2：低优先级 EntropyManager 的所有操作都在 asyncio.Task 中运行，不会阻塞主事件循环。即使清理任务耗时较长，用户对话也不会受到影响。

原则3：优雅停止 当应用关闭时（gui_app.py 的 closeEvent），调用 await entropy_manager.stop()，等待当前清理周期完成后才退出——避免清理到一半被中断导致数据不一致。

2.3 空闲检测

# 监听 EventBus 的 AGENT_RESPONSE 事件，更新最后活动时间
class EntropyManager:
    def __init__(self, event_bus: EventBus):
        # ...
        # 订阅用户活动事件
        event_bus.subscribe(EventType.AGENT_RESPONSE, self._on_agent_response)
    
    async def _on_agent_response(self, event_type, data):
        """更新最后用户活动时间。"""
        self._last_user_activity = time.time()

三、四类清理任务

3.1 死代码扫描

async def _scan_dead_code(self) -> list[DeadCodeReport]:
    """扫描未被引用的函数和类。
    
    策略：
    1. 遍历 src/ 目录下所有 .py 文件
    2. 提取所有 public 函数/类的定义
    3. 检查是否在测试文件或配置中被引用
    4. 报告疑似死代码（未被任何文件 import 或调用）
    """
    dead_codes = []
    for py_file in Path("src").rglob("*.py"):
        # 使用 AST 解析提取函数/类定义
        tree = ast.parse(py_file.read_text())
        for node in ast.walk(tree):
            if isinstance(node, (ast.FunctionDef, ast.ClassDef)):
                if node.name.startswith("_"):
                    continue  # 跳过私有方法
                # 检查是否被引用
                if not _is_referenced(node.name, py_file):
                    dead_codes.append(DeadCodeReport(
                        file=str(py_file),
                        name=node.name,
                        line=node.lineno,
                        type=type(node).__name__,
                    ))
    return dead_codes

3.2 文档一致性检查

async def _check_doc_consistency(self) -> list[InconsistencyReport]:
    """检查文档与代码的一致性。
    
    策略：
    1. 扫描 README.md 中的 API 接口描述
    2. 对比 src/ 中的实际函数签名
    3. 检查 config/ 中的配置项文档
    4. 报告不一致项
    """
    issues = []
    
    # 检查 README 中的函数签名
    readme = Path("README.md").read_text()
    # 提取文档中提到的函数名和参数
    doc_functions = _extract_documented_functions(readme)
    
    for func_name, doc_params in doc_functions.items():
        actual_params = _get_actual_signature(func_name)
        if actual_params is None:
            issues.append(InconsistencyReport(
                type="function_not_found",
                doc_reference=func_name,
                message=f"文档引用的函数 {func_name} 在代码中不存在",
            ))
        elif set(doc_params) != set(actual_params):
            issues.append(InconsistencyReport(
                type="parameter_mismatch",
                doc_reference=func_name,
                message=f"参数不一致: 文档={doc_params}, 实际={actual_params}",
            ))
    
    return issues

3.3 技能花园维护

async def _tend_skill_garden(self) -> GardenReport:
    """维护技能花园。
    
    与 Curator 的区别：
    - Curator 负责技能的创建和合并（高频）
    - EntropyManager 负责技能的使用率审计（低频）
    
    策略：
    1. 统计每个技能过去 30 天的使用次数
    2. 标记使用率 <10% 的技能为"休眠"
    3. 标记与其他技能相似度 >0.9 的技能为"待合并"
    """
    garden = GardenReport()
    
    for skill in skill_manager.list_all():
        usage_count = skill.usage_count_30days
        if usage_count == 0:
            garden.dormant.append(skill.name)
        elif usage_count < 3:
            garden.low_usage.append(skill.name)
        
        # 检查相似度
        for other in skill_manager.list_all():
            if other.id != skill.id:
                similarity = _compute_similarity(skill, other)
                if similarity > 0.9:
                    garden.merge_candidates.append(
                        (skill.name, other.name, similarity)
                    )
    
    return garden

3.4 旧轨迹清理

async def _cleanup_old_traces(self) -> CleanupReport:
    """清理过期的 TaskTrace 日志。
    
    策略：
    1. 扫描 data/traces/ 目录
    2. 删除 30 天前的轨迹文件
    3. 压缩 7-30 天的轨迹文件（只保留摘要）
    """
    report = CleanupReport()
    traces_dir = Path("data/traces")
    
    for trace_file in traces_dir.glob("*.jsonl"):
        age_days = (time.time() - trace_file.stat().st_mtime) / 86400
        
        if age_days > 30:
            trace_file.unlink()
            report.deleted += 1
        elif age_days > 7:
            # 压缩：只保留摘要字段
            _compress_trace(trace_file)
            report.compressed += 1
    
    return report

四、报告与通知

4.1 熵管理报告

每轮清理完成后，EntropyManager 生成一份报告并通过 EventBus 发布：

@dataclass
class EntropyReport:
    """熵管理周期报告。"""
    timestamp: str
    dead_codes_found: int
    doc_inconsistencies: int
    dormant_skills: int
    traces_cleaned: int
    duration_ms: int

# 发布到 EventBus
await self._event_bus.emit("entropy_cycle_complete", report)

4.2 告警机制

当发现严重问题时（如死代码数量 >50 或文档不一致 >10），主动通知用户：

if len(dead_codes) > 50:
    await self._event_bus.emit(
        EventType.SYSTEM_HEALTH_WARNING,
        {"type": "dead_code_accumulation", "count": len(dead_codes)}
    )

五、复用现有调度模式

5.1 学习 AgentPool._cleanup_loop

WeClaw 的 AgentPool 已经有一个后台清理循环（L354-381），EntropyManager 的调度模式直接复用：

# AgentPool._cleanup_loop 的现有模式（参考）
async def _cleanup_loop(self):
    """定期清理过期的 Agent 实例。"""
    while not self._stop_event.is_set():
        try:
            await asyncio.wait_for(
                self._stop_event.wait(), timeout=self._cleanup_interval
            )
            break
        except asyncio.TimeoutError:
            pass
        
        # 执行清理
        async with self._lock:
            expired = [
                sid for sid, (agent, ts) in self._instances.items()
                if time.time() - ts > self._session_timeout
            ]
            for sid in expired:
                await self._instances.pop(sid)[0].aclose()

EntropyManager 使用完全相同的模式，只是清理对象不同。

5.2 停止机制集成

在 gui_app.py 的关闭流程中：

# gui_app.py closeEvent / _shutdown
async def _shutdown(self):
    # ... 其他清理 ...
    if self._entropy_manager:
        await self._entropy_manager.stop()  # 优雅停止
    # ...

六、性能保障

6.1 每项检查的时间上限

检查项预估耗时频率
死代码扫描~300ms每天一次
文档一致性~200ms每天一次
技能花园~500ms每周一次
轨迹清理~100ms每天一次

所有检查项都在 500ms 以内，且只在系统空闲时运行。

6.2 资源隔离

async def run_cycle(self):
    """执行一轮熵管理。"""
    # 设置每个检查的超时
    for check_name, check_fn in self.CHECKS.items():
        try:
            result = await asyncio.wait_for(
                check_fn(), timeout=5.0  # 单项检查最多 5 秒
            )
            self._report(check_name, result)
        except asyncio.TimeoutError:
            logger.warning("熵管理检查超时: %s", check_name)
        except Exception as e:
            logger.error("熵管理检查失败: %s - %s", check_name, e)

七、与 Curator 的关系

EntropyManager 和 Curator 的职责划分：

维度CuratorEntropyManager
触发方式实时（对话后触发）空闲时（后台周期性）
技能管理创建、合并、评分使用率审计、休眠标记
文档检查❌✅
死代码扫描❌✅
轨迹清理❌✅
频率每次对话每天/每周
优先级高（影响用户体验）低（后台运行）

两者互补而非替代：Curator 负责"增量"（不断学习新技能），EntropyManager 负责"减量"（清理过时内容）。

八、核心教训

8.1 熵管理是"可选但必要"的

单个熵增项的影响很小——一个过时的文档、一个未使用的函数——但累积起来会显著降低 Agent 的执行质量。就像内存泄漏不会立即导致崩溃，但长期运行必然出问题。

8.2 空闲触发比定时触发更好

定时触发（如每天凌晨 3 点）的问题是：如果用户在凌晨 3 点正好在用系统，清理任务会和用户请求竞争资源。空闲触发（最近 5 分钟无请求）天然避免了这个问题。

8.3 报告比自动修复更重要

V1 方案曾设想 EntropyManager 自动修复文档不一致——但代码级评审认为这太危险（自动修改文档可能引入新的不一致）。V3 方案改为只报告不修复：生成报告供开发者审查，由人类决定是否修复。

📖 相关文章

本文是 WeClaw 专栏第四季的第 4 篇（总第 65 篇）。如果这篇文章对你有帮助，欢迎给项目点个 Star ⭐

上一篇角色化多Agent协作：从并行隔离到Explorer_Planner_Coder_Reviewer分工编排
下一篇自适应Harness：让系统从历史成功模式中学习