向 DSH 学什么：WeClaw 经验回写闭环的五个迭代方向
系列文章第 9 篇 - 从 DeepSeek Harness 的防御模式、事件溯源和插件架构中提炼 WeClaw 可落地的优化思考
前言
[上一篇文章](./WeClaw_98_当自进化遇上极致工程化：WeClaw与DeepSeek Harness架构深度对比.md)中，我们对比了 WeClaw 的"自进化 Agent"路线和 DeepSeek Harness（DSH）的"极致可组合"路线。结论是：两者代表了 Agent 技术栈的两条互补路径。
本文聚焦实操层面：WeClaw 可以从 DSH 借鉴什么？这些借鉴如何落地为经验回写闭环的具体迭代？
我们提炼出五个方向，按实施难度从低到高排列，每个方向都配有代码级的设计草案和预期收益分析。
一、异常容器化：给 EventBus 装上"保险丝"
1.1 问题
DSH 的防御模式文档（defensive-patterns.md）第 5 条规则：
Contain callback exceptions in the dispatcher. A user-supplied listener that throws must not reject the promise it runs inside or starve the listeners after it.
WeClaw 的经验回写引擎通过 EventBus 订阅事件，采用 fire-and-forget 模式。当前的异常处理是粗粒度的 try/except：
# 当前实现（experience_writeback.py）
async def _handle_event(self, event):
try:
# ... 所有逻辑
except Exception as e:
logger.debug(f"Experience writeback event handling error: {e}")
风险：如果某个经验条目触发了未预期的异常（如畸形 JSON、嵌入向量维度不匹配），整个事件处理链可能中断，导致后续事件丢失。
1.2 DSH 的做法
DSH 在 Cordis 事件分发层实现了严格的异常容器化：
// DSH Cordis 事件分发（概念简化）
for (const listener of listeners) {
try {
await listener(event)
} catch (err) {
// 单个监听器异常被容器化，不影响后续监听器
this.ctx.logger.warn(`listener failed: ${err.message}`)
}
}
1.3 优化方案
为 WeClaw 的经验回写引擎引入分段容器化，将事件处理拆分为独立的阶段，每个阶段独立捕获异常：
# 优化后：分段容器化
async def _handle_event(self, event):
"""分段处理，每段独立异常容器化。"""
# 阶段 1: 事件解析（不应因解析失败影响后续事件）
try:
candidate = self._extract_candidate(event)
if not candidate:
return
except Exception as e:
logger.debug(f"[Stage 1] Candidate extraction failed: {e}")
return # 本事件放弃，不影响后续
# 阶段 2: 评分门控（纯计算，异常概率低但需要防御）
try:
score = self._score_candidate(candidate)
except Exception as e:
logger.debug(f"[Stage 2] Scoring failed for candidate: {e}")
return
# 阶段 3: 存储写入（数据库异常需要隔离）
try:
experience_id = self._store_candidate(candidate, score)
except Exception as e:
logger.warning(f"[Stage 3] Store failed: {e}")
return
# 阶段 4: 变异生成（模板异常需要隔离）
if score >= self._mutate_threshold:
try:
self._generate_and_store_variants(candidate, experience_id)
except Exception as e:
logger.debug(f"[Stage 4] Mutation failed: {e}")
# 变异失败不影响已入库的原始经验
1.4 预期收益
| 指标 | 当前 | 优化后 |
|---|---|---|
| 单事件异常影响范围 | 整个处理链中断 | 仅本事件放弃 |
| 异常定位粒度 | 粗粒度（只知道"某处出错"） | 精确到阶段 |
| 数据完整性 | 变异失败可能回滚原始入库 | 各阶段独立，互不影响 |
实施难度：低。纯重构，不改变外部行为。
二、事件类型强化：给经验生命周期一个"类型身份证"
2.1 问题
当前 WeClaw 的经验相关事件通过 EventBus 的通用 EventType 传递，没有专用的事件类型体系。经验生命周期的关键转换（采集 → 评分 → 入库 → 变异 → 回滚）散落在方法调用日志中，缺乏结构化的审计追踪。
DSH 的做法截然不同——它有 40+ 种强类型事件，每种事件都有：
- 唯一的类型标识（
'goal/change'、'compaction/start'） - 类型安全的载荷声明
- Surface vs Log-only 的明确分类
- 单调序列号保证顺序
2.2 优化方案
为经验回写闭环定义专用的事件类型枚举：
# src/core/experience_events.py（新增）
from enum import Enum
from dataclasses import dataclass
from typing import Optional
class ExperienceEventType(Enum):
"""经验生命周期事件类型。"""
# 采集阶段
CANDIDATE_CAPTURED = "experience/candidate_captured"
CANDIDATE_REJECTED = "experience/candidate_rejected"
# 评分阶段
SCORED = "experience/scored"
# 门控决策
GATED_DISCARD = "experience/gated_discard"
GATED_STORE = "experience/gated_store"
GATED_MUTATE = "experience/gated_mutate"
# 变异阶段
VARIANT_GENERATED = "experience/variant_generated"
VARIANT_STORED = "experience/variant_stored"
# 观测阶段
HIT_COUNT_SAMPLED = "experience/hit_count_sampled"
# 回滚阶段
ROLLBACK_TRIGGERED = "experience/rollback_triggered"
ROLLBACK_COMPLETED = "experience/rollback_completed"
# 系统事件
ENGINE_ENABLED = "experience/engine_enabled"
ENGINE_DISABLED = "experience/engine_disabled"
DAILY_LIMIT_REACHED = "experience/daily_limit_reached"
@dataclass(frozen=True)
class ExperienceEvent:
"""经验生命周期事件（不可变）。"""
type: ExperienceEventType
seq: int # 单调递增序列号
timestamp: float # Unix 时间戳
experience_id: Optional[int] # 关联的经验条目 ID
data: dict # 类型特定的载荷
log_only: bool = True # 默认 log-only，不进入模型上下文
2.3 事件流转图
sequenceDiagram
participant EB as EventBus
participant WE as WritebackEngine
participant Store as ExperienceStore
participant Log as writeback_log
EB->>WE: TOOL_RESULT (success after failures)
WE->>Log: experience/candidate_captured {tool, failures_count}
WE->>WE: score = _score_candidate()
WE->>Log: experience/scored {score=0.87, components={...}}
alt score < 0.5
WE->>Log: experience/gated_discard {reason="low_score"}
else 0.5 <= score < 0.8
WE->>Store: store(candidate)
WE->>Log: experience/gated_store {id=142}
else score >= 0.8
WE->>Store: store(candidate)
WE->>Log: experience/gated_store {id=142}
WE->>WE: _generate_variants()
WE->>Store: store(variant_1)
WE->>Log: experience/variant_generated {parent=142, strategy="add_example"}
WE->>Store: store(variant_2)
WE->>Log: experience/variant_generated {parent=142, strategy="add_constraint"}
end
Note over WE,Log: 后台观测任务（定期执行）
WE->>Store: query hit_counts(window)
WE->>Log: experience/hit_count_sampled {parent=142, child_hits=3, parent_hits=8}
alt child.hit_rate < parent.hit_rate * 0.5
WE->>Store: rollback(variant_id)
WE->>Log: experience/rollback_triggered {variant=143, reason="hit_rate_degradation"}
WE->>Log: experience/rollback_completed {variant=143, status="deactivated"}
end
2.4 预期收益
| 收益 | 说明 |
|---|---|
| 审计完整性 | 每个生命周期转换都有独立事件记录，可重建完整历史 |
| 调试效率 | 异常定位从"方法调用栈"变为"事件流分析" |
| 可观测性 | 可以构建经验生命周期的仪表盘（如每日门控决策分布） |
| 论文支撑 | W1 论文消融实验的数据采集基础 |
实施难度：低-中。新增事件类型定义 + 在现有代码中插入 emit 调用。
三、乐观并发控制：给经验库加上"版本号"
3.1 问题
当前 ExperienceStore 使用 threading.RLock 保护写操作，读操作依赖 SQLite 的默认隔离级别。在经验回写引擎与检索查询并发执行时，存在以下场景：
时间线:
t1: 回写引擎读取经验 #42 的 hit_count = 10
t2: 检索查询命中经验 #42，hit_count += 1 → 11
t3: 回写引擎基于 hit_count=10 做回滚判定
t4: 回滚决策基于过期数据，可能误判
DSH 在多个模块中使用了乐观并发控制（Optimistic Concurrency Control, OCC）来解决类似问题。
3.2 DSH 的做法
DSH 的 MessageFeedbackService 和 GoalService 都使用版本号 CAS（Compare-And-Set）：
// DSH GoalService 的乐观并发（概念简化）
edit(agent, ref: GoalRef, request) {
// ref 包含期望的当前 revision
current = this.expectCurrent(cache, ref)
// 如果 revision 不匹配 → 抛 "stale ref" 错误
// 只有 revision 匹配时才执行更新
newGoal = { ...current, revision: current.revision + 1 }
}
3.3 优化方案
为经验条目引入版本号，回滚决策使用 CAS 语义：
# experience_store.py 新增字段
@dataclass
class Experience:
# ... 现有字段 ...
revision: int = 1 # 乐观并发版本号
# 回滚决策的 CAS 语义
def rollback_if_stale(self, experience_id: int, expected_revision: int) -> bool:
"""仅在版本号匹配时执行回滚。
Returns:
True if rollback succeeded (version matched)
False if version mismatched (concurrent modification detected)
"""
cursor = self._conn.execute(
"""UPDATE experiences
SET status = 'inactive', revision = revision + 1
WHERE id = ? AND revision = ? AND status = 'active'""",
(experience_id, expected_revision),
)
self._conn.commit()
return cursor.rowcount > 0
3.4 回滚决策的完整流程
# experience_writeback.py
def _check_and_rollback(self, parent_id: int):
"""命中率观测窗回滚（带 OCC 保护）。"""
# 1. 读取当前快照（含 revision）
parent = self._store.get_with_revision(parent_id)
variants = self._store.get_variants(parent_id)
# 2. 计算命中率
window_hits = sum(v.hit_count for v in variants)
window_total = sum(v.total_queries for v in variants)
if window_total < 5:
return # 样本不足
child_hit_rate = window_hits / window_total
parent_hit_rate = parent.hit_count / max(parent.total_queries, 1)
# 3. 回滚判定
if child_hit_rate < parent_hit_rate * 0.5:
for variant in variants:
# 4. CAS 回滚：仅在版本号匹配时执行
success = self._store.rollback_if_stale(
variant.id, variant.revision
)
if success:
self._log_event(ExperienceEventType.ROLLBACK_COMPLETED, {
"variant_id": variant.id,
"parent_revision": parent.revision,
"child_hit_rate": child_hit_rate,
"parent_hit_rate": parent_hit_rate,
})
else:
# 版本号不匹配 → 并发修改已发生，跳过本次回滚
logger.info(
f"Rollback skipped for variant {variant.id}: "
f"revision mismatch (expected {variant.revision})"
)
3.5 预期收益
| 场景 | 当前 | 优化后 |
|---|---|---|
| 并发读写 | RLock 保护写，读可能过期 | OCC 版本号检测过期 |
| 回滚误判 | 基于快照可能误判 | CAS 保证只有最新数据能触发回滚 |
| 竞态调试 | 难以重现 | 版本号不匹配时有明确日志 |
实施难度：中。需要在 experiences 表增加 revision 列并做 schema 迁移。
四、抽象服务层：为经验引擎预留"可插拔后端"
4.1 问题
当前 ExperienceWritebackEngine 是一个具体实现类，直接依赖 ExperienceStore 和 MutationEngine。如果未来需要：
- 替换评分算法（如从归一化加权改为 LLM 评估）
- 替换变异策略（如引入 LLM 驱动的语义变异）
- 替换存储后端（如从 SQLite 迁移到 PostgreSQL）
需要修改 ExperienceWritebackEngine 的核心代码，违反开闭原则。
DSH 的做法是定义抽象服务层（Service Definition），具体实现作为独立的后端插件：
// DSH CompactionEngine 的抽象层（简化）
abstract class CompactionEngine extends Service {
abstract compactIfNeeded(agent, trigger, signal): Promise<CompactionResult | null>
abstract compactNow(agent, signal): Promise<CompactionResult | null>
abstract compactRegion(start, end, agent, signal): Promise<CompactionResult>
}
// 具体实现
class CompactionBasic extends CompactionEngine {
// 具体的压力策略和摘要算法
}
4.2 优化方案
将经验回写引擎拆分为三层：
┌─────────────────────────────────────────────┐
│ WritebackOrchestrator（编排层） │
│ - EventBus 订阅 │
│ - 频率保护（每日上限） │
│ - 生命周期管理 │
└─────────────┬───────────────────────────────┘
│ 依赖抽象接口
┌─────────┼─────────┬──────────────┐
▼ ▼ ▼ ▼
┌────────┐ ┌────────┐ ┌──────────┐ ┌────────┐
│ Scorer │ │ Gater │ │ Mutator │ │Observer│
│(评分器)│ │(门控器)│ │(变异器) │ │(观测器)│
│ │ │ │ │ │ │ │
│ 接口 │ │ 接口 │ │ 接口 │ │ 接口 │
│ ────── │ │ ────── │ │ ──────── │ │ ────── │
│ 默认 │ │ 三档 │ │ 4 算子 │ │hit_cnt │
│ 实现 │ │ 实现 │ │ 实现 │ │ 实现 │
│ │ │ │ │ │ │ │
│ LLM │ │ 二元 │ │ LLM │ │ 任务批 │
│ 实现 │ │ 实现 │ │ 实现 │ │ 成功率 │
│(未来) │ │(未来) │ │(未来) │ │(未来) │
└────────┘ └────────┘ └──────────┘ └────────┘
# src/core/writeback/interfaces.py（新增）
from abc import ABC, abstractmethod
from dataclasses import dataclass
@dataclass
class ScoreResult:
score: float # [0, 1]
components: dict[str, float] # 各分量明细
method: str # "weighted_v1" | "llm_judge" | ...
class ExperienceScorer(ABC):
"""经验评分器接口。"""
@abstractmethod
def score(self, candidate: dict) -> ScoreResult:
"""评估候选经验的质量分数。"""
...
class ExperienceGater(ABC):
"""门控决策器接口。"""
@abstractmethod
def decide(self, score_result: ScoreResult) -> str:
"""返回门控决策：'discard' | 'store' | 'mutate'。"""
...
class ExperienceMutator(ABC):
"""变异策略器接口。"""
@abstractmethod
def generate_variants(
self, parent: Any, max_variants: int = 2, **context
) -> list[dict]:
"""为高质量经验生成变体。"""
...
class ExperienceObserver(ABC):
"""效果观测器接口。"""
@abstractmethod
def should_rollback(self, parent_id: int, variants: list) -> bool:
"""基于观测数据判断是否需要回滚。"""
...
4.3 当前默认实现（向后兼容）
# src/core/writeback/defaults.py（新增）
class WeightedScorer(ExperienceScorer):
"""默认的归一化加权评分器（现有逻辑提取）。"""
WEIGHTS = {
"fail_recovery": 0.3,
"field_completeness": 0.3,
"keyword_hit": 0.2,
"novelty_bonus": 0.2,
}
def score(self, candidate: dict) -> ScoreResult:
# 现有评分逻辑原样迁移
...
class ThreeTierGater(ExperienceGater):
"""默认的三档门控器。"""
def decide(self, score_result: ScoreResult) -> str:
s = score_result.score
if s < 0.5:
return "discard"
elif s < 0.8:
return "store"
else:
return "mutate"
class TemplateMutator(ExperienceMutator):
"""默认的模板变异器（现有 4 算子原样迁移）。"""
STRATEGIES = ["add_example", "add_constraint", "restructure", "add_edge_case"]
...
4.4 预期收益
| 场景 | 当前 | 优化后 |
|---|---|---|
| 替换评分算法 | 修改 experience_writeback.py | 实现新的 ExperienceScorer |
| W1 消融实验 | 需要大量代码分支 | 运行时替换 Scorer/Gater 实例 |
| LLM 评估对照 | 硬编码条件分支 | 注入 LLMScorer 实例 |
| 社区扩展 | 不支持 | 第三方可实现自定义后端 |
实施难度：中-高。需要将现有 655 行的 ExperienceWritebackEngine 拆分为 4 个接口 + 4 个默认实现 + 1 个编排器。
五、目标状态机：给经验库一个"使命"
5.1 问题
当前 WeClaw 的经验库是一个无目标的被动存储——它不知道"为什么积累经验"、"当前阶段的经验重点是什么"、"哪些经验已经完成了使命"。
DSH 的 GoalService 提供了形式化的目标管理：
// DSH GoalService 状态机
interface GoalSnapshot {
id: GoalId
revision: number
objective: string // 目标描述
phase: GoalPhase // active | paused | blocked | complete
maxGoalRounds: number // 轮次预算
}
5.2 优化方案：经验库的"使命周期"
为经验库引入使命（Mission）概念——一个有时限和成功标准的阶段性目标：
# src/core/experience_mission.py（新增概念设计）
@dataclass
class ExperienceMission:
"""经验库的阶段使命。"""
id: str
objective: str # "提升文件操作类工具的故障恢复率"
phase: str # active | paused | complete | archived
target_tool_category: str # "file_operations"
success_threshold: float # 0.8 (目标命中率)
current_hit_rate: float # 当前命中率
round_count: int # 已执行轮次
max_rounds: int # 最大轮次
created_at: str
completed_at: Optional[str]
def is_satisfied(self) -> bool:
"""使命是否达成？"""
return (
self.current_hit_rate >= self.success_threshold
and self.round_count >= 3 # 至少 3 个观测周期
)
stateDiagram-v2
[*] --> active : 创建使命
active --> active : 执行回写轮次
active --> paused : 人工暂停
paused --> active : 恢复
active --> complete : 命中率达标 + 周期足够
active --> archived : 超过最大轮次
complete --> [*]
archived --> [*]
5.3 使命驱动的经验采集
有了使命后，评分门控可以引入使命相关性加分：
class MissionAwareScorer(ExperienceScorer):
"""使命感知的评分器。"""
def score(self, candidate: dict) -> ScoreResult:
base_score = self._base_scorer.score(candidate)
# 使命相关性加分
mission = self._get_active_mission()
if mission and self._is_relevant(candidate, mission):
bonus = 0.1 # 使命相关经验 +10% 加分
adjusted = min(1.0, base_score.score + bonus)
return ScoreResult(
score=adjusted,
components={**base_score.components, "mission_bonus": bonus},
method="weighted_v1_mission_aware",
)
return base_score
5.4 预期收益
| 维度 | 无使命（当前） | 有使命（优化后） |
|---|---|---|
| 经验采集策略 | 全量（所有工具失败都采集） | 可聚焦（优先采集使命相关工具） |
| 质量评估 | 通用评分 | 使命相关性加分 |
| 成功标准 | 无（经验库永远增长） | 有（命中率达标即使命完成） |
| 库漂移治理 | 被动（回滚检测退化） | 主动（使命完成后可归档整批经验） |
实施难度：高。需要设计使命生命周期、与现有评分系统集成、构建使命管理 UI。
六、五个方向的实施优先级
| 优先级 | 方向 | 实施难度 | 收益 | 前置条件 |
|---|---|---|---|---|
| P0 | 异常容器化 | 低 | 高（稳定性） | 无 |
| P1 | 事件类型强化 | 低-中 | 高（可观测性 + 论文数据） | 无 |
| P2 | 乐观并发控制 | 中 | 中（回滚精度） | schema 迁移 |
| P3 | 抽象服务层 | 中-高 | 高（可扩展性 + 消融实验） | P1 完成 |
| P4 | 目标状态机 | 高 | 高（使命驱动） | P3 完成 |
gantt
title WeClaw 经验回写闭环迭代路线图
dateFormat YYYY-MM
axisFormat %Y-%m
section P0 异常容器化
代码重构 :p0, 2026-09, 1M
测试验证 :p0t, after p0, 1M
section P1 事件类型
事件类型定义 :p1, 2026-09, 1M
插入 emit 调用 :p1e, after p1, 1M
仪表盘原型 :p1d, after p1e, 1M
section P2 OCC
Schema 迁移 :p2, 2026-10, 1M
CAS 回滚逻辑 :p2c, after p2, 1M
section P3 抽象层
接口设计 :p3, 2026-11, 1M
默认实现迁移 :p3m, after p3, 2M
section P4 使命
概念设计 :p4, 2026-12, 1M
核心实现 :p4i, after p4, 2M
七、总结：从"能用"到"好用"的五个台阶
本文提出的五个优化方向，本质上是把 DSH 在工程化维度的最佳实践，适配到 WeClaw 的自进化场景中：
| 台阶 | 借鉴来源 | WeClaw 落地 |
|---|---|---|
| 异常容器化 | DSH 防御模式 #5 | EventBus 分段 try/except |
| 事件类型强化 | DSH Session Event Log | 经验生命周期专用事件 |
| 乐观并发控制 | DSH GoalService CAS | 经验版本号 + 回滚 CAS |
| 抽象服务层 | DSH CompactionEngine 抽象类 | Scorer/Gater/Mutator/Observer 接口 |
| 目标状态机 | DSH GoalService 状态机 | 经验库使命周期管理 |
核心观点：
-
自进化不等于粗糙。WeClaw 的经验回写闭环在概念层面领先全球（三档门控 + 遗传变异 + 命中率回滚的组合无完整先例），但在工程实现上仍有显著提升空间。DSH 的防御模式、类型安全、可插拔架构是值得学习的工程标杆。
-
事件溯源是论文数据采集的基础设施。P1（事件类型强化）的实施将直接为 W1 论文提供消融实验的数据采集管道——每一次门控决策、每一次变异、每一次回滚都有结构化事件记录。
-
抽象服务层是消融实验的前提。W1 论文需要对比"无门控 vs 二元门控 vs 三档门控"、"LLM 评分 vs 客观评分"等消融条件。P3（抽象服务层）使得这些对照可以通过运行时替换组件实例来实现，而不需要代码分支。
-
使命驱动是经验库治理的下一个前沿。当经验库从"被动积累"进化到"主动聚焦"，我们不仅解决了漂移问题，还让 Agent 具备了"有目标地学习"的能力——这可能是 WeClaw 经验系统的下一个论文主题。
系列文章导航：
| 篇序 | 标题 | 核心内容 |
|---|---|---|
| 01–05 | WinClaw 工具优化实战专栏 | 从工具滥用灾难到渐进式暴露 |
| 06 | 双通道密钥管理 | 开发者便利与用户安全的平衡 |
| 07 | 双通道实时通信 | PWA 多实例绑定路由 |
| 08 | 当自进化遇上极致工程化 | WeClaw vs DSH 架构对比 |
| 09 | 本文 | DSH 借鉴思考与迭代方向 |
参考资料：
- DeepSeek Harness 防御模式文档：defensive-patterns.md
- DeepSeek Harness 持久化事件目录：persistence-catalog.md
- DeepSeek Harness 能力接缝图：capability-seams.md
- WeClaw 经验回写闭环论文规划：weclaw_series_papers_经验回写闭环新论文主题规划_20260826.md
上一篇文章：[当自进化遇上极致工程化：WeClaw 与 DeepSeek Harness 架构深度对比](./WeClaw_98_当自进化遇上极致工程化：WeClaw与DeepSeek Harness架构深度对比.md)