WeClaw_66_自适应Harness：让系统从历史成功模式中学习
第四季系列文章第 5 篇（总第 66 篇） - Adaptive Harness · 模式挖掘 · 工具暴露优化 · 数据驱动 · 渐进调整
📚 专栏信息
《从零到一构建跨平台 AI 助手：WeClaw 实战指南》专栏 · 第四季
专栏定位：面向开发者和技术决策者的实战专栏，用真实案例和完整代码带你理解如何构建生产级 AI 应用
本文深入探讨 Harness Engineering 七大差距中的 G4——自适应 Harness。WeClaw 当前的工具暴露策略是"配置驱动"的——confidence ≥ 0.8 时推荐集、≥ 0.5 时扩展集、否则全量集。但这个阈值是人工设定的，不会根据实际使用效果自动调整。本文设计 AdaptiveHarness：一个从 TaskTrace 历史数据中挖掘成功模式，自动调整工具暴露策略和工具排序的引擎。
👨💻 作者与项目
作者简介：翁勇刚 WENG YONGGANG 新概念龙虾-WeClaw 开发团队负责人，一群专注于跨平台 AI 应用的实践者 理念："再复杂的技术，也能用代码讲清楚"
- 💻 项目地址：https://github.com/wyg5208/weclaw.git
- 🌐 官网地址：https://weclaw.link
- 📝 作者 CSDN：https://blog.csdn.net/yweng18
- ⭐ 欢迎 Star⭐、Fork🍴、贡献代码🤝
📝 摘要
本文结构概览： 从"静态阈值 vs 动态适应"的讨论出发——当所有意图都用同一个 confidence 阈值（0.8/0.5）来决定工具暴露层级时，某些意图可能需要更精确的工具子集，而另一些意图可能需要更宽泛的探索。然后设计 AdaptiveHarness：利用 TaskTrace 中记录的意图→工具序列→成功/失败数据，挖掘"成功模式"（如"code_modify 意图 + file+shell 工具序列 = 高成功率"），自动调整 ToolExposureEngine 的层级阈值和工具排序。重点讲解数据积累要求（min_samples=10）、安全边界（调整幅度 ±10%、每日重置）和与 ToolExposureEngine 的集成方式。
核心问题： WeClaw 的渐进式工具暴露引擎是一个创新设计，但它的分层阈值是静态的。如何让它"越用越聪明"——根据历史成功模式自动优化工具选择？
关键成果：
- 设计了 AdaptiveHarness 的模式挖掘和策略生成流程
- 与 ToolExposureEngine 的 8 行集成代码
- 安全边界设计（±10% 调整幅度 + 每日重置 + min_samples 门槛）
适合读者：对数据驱动 Agent 优化感兴趣的开发者、机器学习工程师
阅读时长：约 14 分钟
关键词：Adaptive Harness、模式挖掘、工具暴露优化、TaskTrace、数据驱动、ExperienceStore
一、静态阈值 vs 动态适应
1.1 当前系统：一刀切的阈值
WeClaw 的 ToolExposureEngine._determine_tier() 方法：
def _determine_tier(self, intent_result):
if _forced_tier:
return _forced_tier
if intent_result.confidence >= 0.8:
return "recommended" # 高置信度 → 只看推荐工具
elif intent_result.confidence >= 0.5:
return "extended" # 中置信度 → 扩展工具集
else:
return "full" # 低置信度 → 全量工具
这个设计已经很优秀了——它根据意图置信度分层暴露工具，减少了模型的选择空间。但问题是：0.8 和 0.5 这两个阈值是拍脑袋定的。
1.2 不同意图需要不同阈值
考虑两种意图：
意图 A：casual_chat（闲聊）
- 成功率高，工具需求少
- confidence ≥ 0.7 时就可以只暴露推荐集
- 当前阈值 0.8 太保守，导致闲聊时也暴露了过多工具
意图 B：code_modify（代码修改）
- 需要更精确的工具选择
- confidence ≥ 0.85 时才应该收窄到推荐集
- 当前阈值 0.8 太激进，导致代码修改时过早收窄工具
核心洞察：不同意图的"最佳阈值"不同，而且这个最佳值会随着使用模式的变化而变化。
1.3 数据就在那里
WeClaw 的 TaskTrace（367行）已经在记录每次任务的完整轨迹：
@dataclass
class TaskTrace:
trace_id: str
session_id: str
user_input: str
intent_primary: str # 意图
intent_confidence: float # 置信度
tool_tier: str # 使用的工具层级
tools_exposed: list[str] # 暴露的工具列表
total_steps: int # 总步数
tool_calls: list # 工具调用序列
final_status: str # "success" | "failure"
total_tokens: int
total_duration_ms: int
这些数据就是"经验"——但我们一直没有利用它来优化工具暴露策略。
二、AdaptiveHarness 设计
2.1 核心架构
class AdaptiveHarness:
"""根据历史成功模式自动调整工具暴露和上下文策略。"""
def __init__(self, trace_dir: Path, experience_store: ExperienceStore):
self._trace_dir = trace_dir
self._experience_store = experience_store
self._pattern_cache: dict[str, SuccessPattern] = {}
self._refresh_interval = 3600 # 每小时刷新
self._last_refresh = 0
def analyze_patterns(self, traces: list[TaskTrace]) -> HarnessStrategy:
"""从历史轨迹中挖掘成功模式。"""
...
def suggest_tool_order(self, intent: str) -> list[str]:
"""为特定意图推荐工具排序。"""
...
def get_tier_override(self, intent: str) -> str | None:
"""为特定意图返回层级覆盖。"""
...
2.2 模式挖掘
@dataclass
class SuccessPattern:
"""一个意图的成功模式。"""
intent: str
optimal_tier: str # 最佳工具层级
optimal_confidence_threshold: float # 最佳置信度阈值
tool_sequence: list[str] # 成功任务的工具序列
avg_steps: int # 平均步数
success_rate: float # 成功率
sample_count: int # 样本数
def analyze_patterns(self, traces: list[TaskTrace]) -> dict[str, SuccessPattern]:
"""按意图分组，挖掘每种意图的成功模式。"""
patterns = {}
# 按意图分组
intent_groups = {}
for trace in traces:
intent = trace.intent_primary
if intent not in intent_groups:
intent_groups[intent] = []
intent_groups[intent].append(trace)
for intent, group_traces in intent_groups.items():
# 数据量不足时跳过
if len(group_traces) < self._min_samples:
continue
# 分离成功和失败
success_traces = [t for t in group_traces if t.final_status == "success"]
failure_traces = [t for t in group_traces if t.final_status == "failure"]
if not success_traces:
continue
# 挖掘成功模式
success_tiers = [t.tool_tier for t in success_traces]
optimal_tier = max(set(success_tiers), key=success_tiers.count)
# 计算最佳置信度阈值
success_confidences = [t.intent_confidence for t in success_traces]
optimal_threshold = sum(success_confidences) / len(success_confidences)
# 提取工具序列模式
tool_sequences = [tuple(t.tools_exposed[:5]) for t in success_traces]
most_common_seq = max(set(tool_sequences), key=tool_sequences.count)
patterns[intent] = SuccessPattern(
intent=intent,
optimal_tier=optimal_tier,
optimal_confidence_threshold=optimal_threshold,
tool_sequence=list(most_common_seq),
avg_steps=sum(t.total_steps for t in success_traces) // len(success_traces),
success_rate=len(success_traces) / len(group_traces),
sample_count=len(group_traces),
)
return patterns
2.3 策略生成
def get_tier_override(self, intent: str) -> dict | None:
"""为特定意图返回层级覆盖策略。"""
pattern = self._pattern_cache.get(intent)
if not pattern or pattern.sample_count < self._min_samples:
return None # 数据不足，不覆盖
# 安全边界：调整幅度不超过 ±10%
base_threshold = 0.8 # 默认阈值
adjusted = pattern.optimal_confidence_threshold
clamped = max(base_threshold * 0.9, min(base_threshold * 1.1, adjusted))
return {
"forced_tier": pattern.optimal_tier,
"adjusted_threshold": clamped,
"confidence": pattern.success_rate,
}
三、与 ToolExposureEngine 的集成
3.1 8 行集成代码
在 tool_exposure.py 的 _determine_tier() 方法末尾追加：
def _determine_tier(self, intent_result):
# ... 现有逻辑 ...
# Adaptive harness override (G4)
if self._adaptive_overrides:
override = self._adaptive_overrides.get(intent_result.primary_intent)
if override:
forced_tier = override.get("forced_tier")
if forced_tier in ("recommended", "extended", "full"):
_tier_order = {"recommended": 0, "extended": 1, "full": 2}
# 只能放宽，不能收窄（安全约束）
if _tier_order.get(forced_tier, 0) >= _tier_order.get(tier, 0):
return forced_tier
return tier
3.2 关键安全约束：只能放宽，不能收窄
注意这行代码：
if _tier_order.get(forced_tier, 0) >= _tier_order.get(tier, 0):
这意味着 AdaptiveHarness 只能将"recommended"放宽到"extended"或"full"，不能将"full"收窄到"recommended"。
为什么？ 因为收窄工具集是有风险的——如果自适应策略判断错误，收窄会导致模型看不到必要的工具。放宽是安全的——即使判断错误，最多是多给了一些工具，不会导致功能缺失。
四、数据源与积累
4.1 双数据源
AdaptiveHarness 的数据来自两个现有系统：
数据源1：TaskTrace
# task_trace.py — 每次任务的完整轨迹
trace = TaskTrace(
intent_primary="code_modify",
intent_confidence=0.85,
tool_tier="recommended",
tools_exposed=["file", "shell", "codebase_search"],
total_steps=12,
final_status="success",
)
数据源2：ExperienceStore
# experience_store.py — SQLite + ChromaDB 双引擎
# 存储历史经验评分、意图分布统计
experience = experience_store.get(intent="code_modify")
# → {"avg_success_rate": 0.82, "common_tools": ["file", "shell"], ...}
4.2 数据积累要求
[agent.adaptive_harness]
enabled = false
analysis_window = 20 # 分析最近 20 次任务
auto_adjust_tier_thresholds = true
min_samples_per_intent = 10 # 每个意图至少 10 个样本才调整
min_samples=10 的含义：只有某个意图积累了至少 10 次任务记录后，AdaptiveHarness 才会为该意图生成覆盖策略。10 次以下时，继续使用默认阈值。
4.3 刷新策略
async def _refresh_patterns(self):
"""每小时刷新一次模式缓存。"""
while not self._stop_event.is_set():
await asyncio.sleep(self._refresh_interval) # 3600 秒
# 加载最近的轨迹
traces = self._load_recent_traces(limit=self._analysis_window)
# 挖掘模式
new_patterns = self.analyze_patterns(traces)
# 安全合并：只更新新发现的模式，不删除已有的
for intent, pattern in new_patterns.items():
old = self._pattern_cache.get(intent)
if old:
# 调整幅度不超过 ±10%
new_threshold = pattern.optimal_confidence_threshold
old_threshold = old.optimal_confidence_threshold
clamped = max(
old_threshold * 0.9,
min(old_threshold * 1.1, new_threshold)
)
pattern.optimal_confidence_threshold = clamped
self._pattern_cache[intent] = pattern
五、安全边界设计
5.1 三重安全约束
AdaptiveHarness 的自适应能力必须被严格限制——否则它可能"学偏了"：
| 约束 | 值 | 目的 |
|---|---|---|
| 调整幅度 | ±10% | 每次调整不超过当前阈值的 10% |
| 每日重置 | 每天凌晨 | 防止累积偏移 |
| 最小样本 | 10 | 数据不足时不调整 |
| 只放宽不收窄 | 硬编码 | 防止误判导致工具缺失 |
5.2 每日重置
def _check_daily_reset(self):
"""每天凌晨重置所有调整到基线值。"""
today = date.today()
if self._last_reset_date != today:
# 重置所有模式缓存
self._pattern_cache.clear()
self._last_reset_date = today
logger.info("AdaptiveHarness 每日重置完成")
为什么需要每日重置？ 因为自适应策略可能"过拟合"——某一天的数据不代表长期趋势。每日重置确保每次调整都是基于最新的、有限幅度的变化。
5.3 降级机制
当 AdaptiveHarness 的策略导致成功率下降时，自动降级：
def _check_performance_degradation(self):
"""检测自适应策略是否导致性能下降。"""
recent_traces = self._load_recent_traces(limit=20)
recent_success_rate = sum(
1 for t in recent_traces if t.final_status == "success"
) / len(recent_traces)
baseline_success_rate = 0.85 # 基线成功率
if recent_success_rate < baseline_success_rate * 0.9: # 下降超过 10%
logger.warning(
"AdaptiveHarness 性能下降: %.2f vs 基线 %.2f，暂停自适应",
recent_success_rate, baseline_success_rate
)
self._pattern_cache.clear() # 清空所有覆盖
return True # 触发降级
return False
六、扩展 TaskTrace 数据采集
6.1 新增字段
在 TaskTrace 数据类中新增 success_pattern 字段：
@dataclass
class TaskTrace:
# ... 现有字段 ...
# --- 成功模式记录（G4 自适应 Harness 数据源） ---
success_pattern: str = "" # 成功时的 intent→tool_sequence 摘要
6.2 模式摘要生成
# 在 TaskTraceCollector.finalize() 中
def finalize(self, final_status: str):
self._trace.final_status = final_status
if final_status == "success":
# 生成成功模式摘要
tool_seq = [tc.tool_name for tc in self._trace.tool_calls]
self._trace.success_pattern = (
f"{self._trace.intent_primary}→"
f"{'→'.join(tool_seq[:5])}"
)
这个摘要用于 ExperienceStore 的快速检索——当遇到类似意图时，可以快速找到历史成功模式。
七、AdaptiveHarness 的局限性
7.1 冷启动问题
系统刚上线时没有历史数据，AdaptiveHarness 无法工作。解决方案：
- 前 1-2 周使用默认阈值
- 积累 10 个样本后开始调整
- 第 3-4 周达到稳定状态
7.2 过拟合风险
如果某段时间用户的任务类型集中（如都在做代码修改），AdaptiveHarness 会偏向该类型。解决方案：
- 每日重置
- min_samples 门槛
- 只放宽不收窄
7.3 因果关系不明
AdaptiveHarness 发现"code_modify + recommended 层级 = 高成功率"，但无法确定是 recommended 层级导致了高成功率，还是高置信度的任务恰好用了 recommended 层级。这是一个相关性而非因果性的推断。
我们的立场：AdaptiveHarness 是一个辅助工具，不是决策者。它的调整幅度被限制在 ±10% 以内，即使判断错误，影响也有限。
八、核心教训
8.1 数据驱动需要先有数据
G4 是唯一一个"需要等待数据积累"的差距领域。其他 6 个差距都可以立即实施，但 G4 必须等 TaskTrace 积累 1-2 周后才能启用。这提醒我们：数据基础设施的建设应该尽早开始，即使暂时不用。
8.2 自适应的安全边界比算法更重要
AdaptiveHarness 的模式挖掘算法并不复杂——按意图分组、统计成功模式。真正复杂的是安全边界设计：调整幅度多少？何时重置？如何降级？这些"护栏"比算法本身更决定了系统的安全性。
8.3 只放宽不收窄是一个工程判断
从纯算法角度看，"只放宽不收窄"是次优的——有时候确实需要收窄。但从工程角度看，收窄的风险远大于放宽。这是一个典型的"安全优先于最优"的工程决策。
📖 相关文章
本文是 WeClaw 专栏第四季的第 5 篇（总第 66 篇）。如果这篇文章对你有帮助，欢迎给项目点个 Star ⭐