构建 AI Agent 的持续学习闭环：如何从「能纠正」到「越用越强」

class FailureCapture:
    def __init__(self, memory_path):
        self.memory_path = memory_path
        self.current_failure = None

    def capture(self, task_type, manifestation, agent_trace):
        """在失败发生时立即调用"""
        self.current_failure = {
            "failure_id": generate_id(),
            "timestamp": now_iso(),
            "task_type": task_type,
            "manifestation": manifestation,
            "agent_trace": agent_trace,
            "status": "pending_analysis"
        }
        self._write_episodic(self.current_failure)

    def _write_episodic(self, failure_record):
        """写入事件记忆"""
        date = today()
        path = f"{self.memory_path}/{date}.md"
        append_to_file(path, self._format_failure(failure_record))

ROOT_CAUSE_PROMPT = """
任务类型：{task_type}
失败表现：{manifestation}
Agent 执行过程：
{agent_trace}

请分析以下四个问题：
1. Agent 的决策在哪里偏离了正确方向？
2. 当时的上下文信息是否充足？（足够/不足/误导性）
3. 这个失败是 Agent 能力问题还是信息问题？
4. 如果重新处理类似任务，Agent 应该在哪一步改变策略？

请用以下格式输出：
根因：[一句话描述]
根因类型：[信息不足 | 能力不足 | 误导性反馈 | 系统限制]
预防策略：[具体可操作的行为改变]
"""

def extract_pattern(failures):
    """
    从多个失败记录中提取共同模式
    触发条件：相同根因出现 ≥3 次
    """
    # 1. 按根因分组
    by_root_cause = group_by(failures, "root_cause_type")

    patterns = []
    for root_cause, group in by_root_cause.items():
        if len(group) >= 3:
            pattern = {
                "pattern_id": generate_id(),
                "trigger_conditions": extract_common_triggers(group),
                "root_cause": root_cause,
                "prevention_strategy": group[0]["prevention_strategy"],
                "frequency": len(group),
                "confidence": "high" if len(group) >= 5 else "medium",
                "first_seen": group[0]["timestamp"],
                "last_seen": group[-1]["timestamp"]
            }
            patterns.append(pattern)

    return patterns

def inject_task_context(task):
    """在任务开始时，检查相关模式并注入上下文"""
    relevant_patterns = search_patterns(task.type)

    if relevant_patterns:
        context = "[学习历史] 此任务类型过去出现过问题：\n"
        for p in relevant_patterns:
            context += f"- {p['pattern_id']}: {p['prevention_strategy']}\n"
        context += "请在执行时格外注意上述风险点。\n"
        return context
    return ""

def search_episodic(query, date_range=None):
    """检索相关事件记忆"""
    results = []
    for file in list_files("memory/episodic/", date_range):
        content = read_file(file)
        if keyword_match(content, query):
            results.append(parse_episodic(content))
    return results

PATTERN_INDEX = {
    "code_refactor": ["P-001", "P-002"],
    "regex_modification": ["P-001"],
    "unfamiliar_api": ["P-003"],
}

def get_relevant_patterns(task_type):
    """任务开始时获取相关模式"""
    pattern_ids = PATTERN_INDEX.get(task_type, [])
    return [read_pattern(f"memory/patterns/{pid}.yaml") for pid in pattern_ids]

STRATEGIC_MEMORY_INDEX = {
    "refactoring": "memory/strategic/refactoring.md",
    "unfamiliar_code": "memory/strategic/unfamiliar-code.md",
    "high_stakes": "memory/strategic/high-stakes-decisions.md"
}

# ~/.claude/commands/learning-loop.sh
#!/bin/bash

LEARNING_DIR="$HOME/.claude/learning"
mkdir -p "$LEARNING_DIR"/{episodic,patterns,strategic}

capture_failure() {
    local task_type="$1"
    local manifestation="$2"
    local tool_name="$3"

    local failure_id="fail-$(date +%Y%m%d-%H%M%S)"
    local record_file="$LEARNING_DIR/episodic/$(date +%Y-%m-%d).md"

    cat >> "$record_file" << EOF

## $failure_id
**时间**：$(date -Iseconds)
**任务类型**：$task_type
**失败工具**：$tool_name
**表现**：$manifestation
**根因分析**：pending
**状态**：待分析
EOF
    echo "✅ 失败记录已保存: $failure_id"
}

analyze_patterns() {
    # 定期运行（每周一次），从 episodic memory 中提炼模式
    # 使用 Claude Code 对 recent failures 做分析
    echo "🔍 分析近期失败记录，提炼模式..."
}

update_strategy() {
    # 将高置信度模式转为策略记忆
    local pattern_id="$1"
    echo "📝 更新策略记忆: $pattern_id"
}