Agent/agent/tools/builtin/knowledge.py.backup

"""
原子知识保存工具

提供便捷的 API 让 Agent 快速保存结构化的原子知识
"""

import os
import re
import json
import yaml
import logging
from datetime import datetime
from pathlib import Path
from typing import List, Dict, Optional, Any
from agent.tools import tool, ToolResult, ToolContext
from ...llm.openrouter import openrouter_llm_call

logger = logging.getLogger(__name__)


def _generate_knowledge_id() -> str:
    """生成知识原子 ID（带微秒和随机后缀避免冲突）"""
    import uuid
    timestamp = datetime.now().strftime('%Y%m%d-%H%M%S')
    random_suffix = uuid.uuid4().hex[:4]
    return f"knowledge-{timestamp}-{random_suffix}"


def _format_yaml_list(items: List[str], indent: int = 2) -> str:
    """格式化 YAML 列表"""
    if not items:
        return "[]"
    indent_str = " " * indent
    return "\n" + "\n".join(f"{indent_str}- {item}" for item in items)


@tool()
async def save_knowledge(
    scenario: str,
    content: str,
    tags_type: List[str],
    urls: List[str] = None,
    agent_id: str = "research_agent",
    score: int = 3,
    trace_id: str = "",
) -> ToolResult:
    """
    保存原子知识到本地文件（JSON 格式）

    Args:
        scenario: 任务描述（在什么情景下 + 要完成什么目标 + 得到能达成一个什么结果）
        content: 核心内容
        tags_type: 知识类型标签，可选：tool, usercase, definition, plan, strategy
        urls: 参考来源链接列表（论文/GitHub/博客等）
        agent_id: 执行此调研的 agent ID
        score: 初始评分 1-5（默认 3）
        trace_id: 当前 trace ID（可选）

    Returns:
        保存结果
    """
    try:
        # 生成 ID
        knowledge_id = _generate_knowledge_id()

        # 准备目录
        knowledge_dir = Path(".cache/knowledge_atoms")
        knowledge_dir.mkdir(parents=True, exist_ok=True)

        # 构建文件路径（使用 .json 扩展名）
        file_path = knowledge_dir / f"{knowledge_id}.json"

        # 构建 JSON 数据结构
        knowledge_data = {
            "id": knowledge_id,
            "trace_id": trace_id or "N/A",
            "tags": {
                "type": tags_type
            },
            "scenario": scenario,
            "content": content,
            "trace": {
                "urls": urls or [],
                "agent_id": agent_id,
                "timestamp": datetime.now().isoformat()
            },
            "eval": {
                "score": score,
                "helpful": 0,
                "harmful": 0,
                "helpful_history": [],
                "harmful_history": []
            },
            "metrics": {
                "helpful": 1,
                "harmful": 0
            },
            "created_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        }

        # 保存为 JSON 文件
        with open(file_path, "w", encoding="utf-8") as f:
            json.dump(knowledge_data, f, ensure_ascii=False, indent=2)

        return ToolResult(
            title="✅ 原子知识已保存",
            output=f"知识 ID: {knowledge_id}\n文件路径: {file_path}\n\n场景:\n{scenario[:100]}...",
            long_term_memory=f"保存原子知识: {knowledge_id} - {scenario[:50]}",
            metadata={"knowledge_id": knowledge_id, "file_path": str(file_path)}
        )

    except Exception as e:
        return ToolResult(
            title="❌ 保存失败",
            output=f"错误: {str(e)}",
            error=str(e)
        )


@tool()
async def update_knowledge(
    knowledge_id: str,
    add_helpful_case: Optional[Dict[str, str]] = None,
    add_harmful_case: Optional[Dict[str, str]] = None,
    update_score: Optional[int] = None,
    evolve_feedback: Optional[str] = None,
) -> ToolResult:
    """
    更新已有的原子知识的评估反馈

    Args:
        knowledge_id: 知识 ID（如 research-20260302-001）
        add_helpful_case: 添加好用的案例 {"case_id": "...", "scenario": "...", "result": "...", "timestamp": "..."}
        add_harmful_case: 添加不好用的案例 {"case_id": "...", "scenario": "...", "result": "...", "timestamp": "..."}
        update_score: 更新评分（1-5）
        evolve_feedback: 经验进化反馈（当提供时，会使用 LLM 重写知识内容）

    Returns:
        更新结果
    """
    try:
        # 查找文件（支持 JSON 和 MD 格式）
        knowledge_dir = Path(".cache/knowledge_atoms")
        json_path = knowledge_dir / f"{knowledge_id}.json"
        md_path = knowledge_dir / f"{knowledge_id}.md"

        file_path = None
        if json_path.exists():
            file_path = json_path
            is_json = True
        elif md_path.exists():
            file_path = md_path
            is_json = False
        else:
            return ToolResult(
                title="❌ 文件不存在",
                output=f"未找到知识文件: {knowledge_id}",
                error="文件不存在"
            )

        # 读取现有内容
        with open(file_path, "r", encoding="utf-8") as f:
            content = f.read()

        # 解析数据
        if is_json:
            data = json.loads(content)
        else:
            # 解析 YAML frontmatter
            yaml_match = re.search(r'^---\n(.*?)\n---', content, re.DOTALL)
            if not yaml_match:
                return ToolResult(
                    title="❌ 格式错误",
                    output=f"无法解析知识文件格式: {file_path}",
                    error="格式错误"
                )
            data = yaml.safe_load(yaml_match.group(1))

        # 更新内容
        updated = False
        summary = []

        if add_helpful_case:
            data["eval"]["helpful"] += 1
            data["eval"]["helpful_history"].append(add_helpful_case)
            data["metrics"]["helpful"] += 1
            summary.append(f"添加 helpful 案例: {add_helpful_case.get('case_id')}")
            updated = True

        if add_harmful_case:
            data["eval"]["harmful"] += 1
            data["eval"]["harmful_history"].append(add_harmful_case)
            data["metrics"]["harmful"] += 1
            summary.append(f"添加 harmful 案例: {add_harmful_case.get('case_id')}")
            updated = True

        if update_score is not None:
            data["eval"]["score"] = update_score
            summary.append(f"更新评分: {update_score}")
            updated = True

        # 经验进化机制
        if evolve_feedback:
            old_content = data.get("content", "")
            evolved_content = await _evolve_knowledge_with_llm(old_content, evolve_feedback)
            data["content"] = evolved_content
            data["metrics"]["helpful"] += 1
            summary.append(f"知识进化: 基于反馈重写内容")
            updated = True

        if not updated:
            return ToolResult(
                title="⚠️ 无更新",
                output="未指定任何更新内容",
                long_term_memory="尝试更新原子知识但未指定更新内容"
            )

        # 更新时间戳
        data["updated_at"] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

        # 保存更新
        if is_json:
            with open(file_path, "w", encoding="utf-8") as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
        else:
            # 重新生成 YAML frontmatter
            meta_str = yaml.dump(data, allow_unicode=True).strip()
            with open(file_path, "w", encoding="utf-8") as f:
                f.write(f"---\n{meta_str}\n---\n")

        return ToolResult(
            title="✅ 原子知识已更新",
            output=f"知识 ID: {knowledge_id}\n文件路径: {file_path}\n\n更新内容:\n" + "\n".join(f"- {s}" for s in summary),
            long_term_memory=f"更新原子知识: {knowledge_id}"
        )

    except Exception as e:
        return ToolResult(
            title="❌ 更新失败",
            output=f"错误: {str(e)}",
            error=str(e)
        )


@tool()
async def list_knowledge(
    limit: int = 10,
    tags_type: Optional[List[str]] = None,
) -> ToolResult:
    """
    列出已保存的原子知识

    Args:
        limit: 返回数量限制（默认 10）
        tags_type: 按类型过滤（可选）

    Returns:
        知识列表
    """
    try:
        knowledge_dir = Path(".cache/knowledge_atoms")

        if not knowledge_dir.exists():
            return ToolResult(
                title="📂 知识库为空",
                output="还没有保存任何原子知识",
                long_term_memory="知识库为空"
            )

        # 获取所有文件
        files = sorted(knowledge_dir.glob("*.md"), key=lambda x: x.stat().st_mtime, reverse=True)

        if not files:
            return ToolResult(
                title="📂 知识库为空",
                output="还没有保存任何原子知识",
                long_term_memory="知识库为空"
            )

        # 读取并过滤
        results = []
        for file_path in files[:limit]:
            with open(file_path, "r", encoding="utf-8") as f:
                content = f.read()

            # 提取关键信息
            import re
            id_match = re.search(r"id: (.+)", content)
            scenario_match = re.search(r"scenario: \|\n  (.+)", content)
            score_match = re.search(r"score: (\d+)", content)

            knowledge_id = id_match.group(1) if id_match else "unknown"
            scenario = scenario_match.group(1) if scenario_match else "N/A"
            score = score_match.group(1) if score_match else "N/A"

            results.append(f"- [{knowledge_id}] (⭐{score}) {scenario[:60]}...")

        output = f"共找到 {len(files)} 条原子知识，显示最近 {len(results)} 条:\n\n" + "\n".join(results)

        return ToolResult(
            title="📚 原子知识列表",
            output=output,
            long_term_memory=f"列出 {len(results)} 条原子知识"
        )

    except Exception as e:
        return ToolResult(
            title="❌ 列表失败",
            output=f"错误: {str(e)}",
            error=str(e)
        )


# ===== 语义检索功能 =====

async def _route_knowledge_by_llm(query_text: str, metadata_list: List[Dict], k: int = 5) -> List[str]:
    """
    第一阶段：语义路由。
    让 LLM 挑选出 2*k 个语义相关的 ID。
    """
    if not metadata_list:
        return []

    # 扩大筛选范围到 2*k
    routing_k = k * 2

    routing_data = [
        {
            "id": m["id"],
            "tags": m["tags"],
            "scenario": m["scenario"][:100]  # 只取前100字符
        } for m in metadata_list
    ]

    prompt = f"""
你是一个知识检索专家。根据用户的当前任务需求，从下列原子知识元数据中挑选出最相关的最多 {routing_k} 个知识 ID。
任务需求："{query_text}"

可选知识列表：
{json.dumps(routing_data, ensure_ascii=False, indent=1)}

请直接输出 ID 列表，用逗号分隔（例如: knowledge-20260302-001, research-20260302-002）。若无相关项请输出 "None"。
"""

    try:
        print(f"\n[Step 1: 知识语义路由] 任务: '{query_text}' | 候选总数: {len(metadata_list)} | 目标提取数: {routing_k}")

        response = await openrouter_llm_call(
            messages=[{"role": "user", "content": prompt}],
            model="google/gemini-2.0-flash-001"
        )

        content = response.get("content", "").strip()
        selected_ids = [idx.strip() for idx in re.split(r'[,\s]+', content) if idx.strip().startswith(("knowledge-", "research-"))]

        print(f"[Step 1: 知识语义路由] LLM 初选 ID ({len(selected_ids)}个): {selected_ids}")
        return selected_ids
    except Exception as e:
        logger.error(f"LLM 知识路由失败: {e}")
        return []


async def _evolve_knowledge_with_llm(old_content: str, feedback: str) -> str:
    """
    使用 LLM 进行知识进化重写（类似经验进化机制）
    """
    prompt = f"""你是一个 AI Agent 知识库管理员。请根据反馈建议，对现有的知识内容进行重写进化。

【原知识内容】:
{old_content}

【实战反馈建议】:
{feedback}

【重写要求】:
1. 融合知识：将反馈中的避坑指南、新参数或修正后的选择逻辑融入原知识，使其更具通用性和准确性。
2. 保持结构：如果原内容有特定格式（如 Markdown、代码示例等），请保持该格式。
3. 语言：简洁直接，使用中文。
4. 禁止：严禁输出任何开场白、解释语或额外的 Markdown 标题，直接返回重写后的正文。
"""
    try:
        response = await openrouter_llm_call(
            messages=[{"role": "user", "content": prompt}],
            model="google/gemini-2.0-flash-001"
        )

        evolved_content = response.get("content", "").strip()

        # 简单安全校验：如果 LLM 返回太短或为空，回退到原内容+追加
        if len(evolved_content) < 5:
            raise ValueError("LLM output too short")

        return evolved_content

    except Exception as e:
        logger.warning(f"知识进化失败，采用追加模式回退: {e}")
        timestamp = datetime.now().strftime('%Y-%m-%d')
        return f"{old_content}\n\n---\n[Update {timestamp}]: {feedback}"


async def _route_knowledge_by_llm(query_text: str, metadata_list: List[Dict], k: int = 5) -> List[str]:
    """
    第一阶段：语义路由。
    让 LLM 挑选出 2*k 个语义相关的 ID。
    """
    if not metadata_list:
        return []

    # 扩大筛选范围到 2*k
    routing_k = k * 2

    routing_data = [
        {
            "id": m["id"],
            "tags": m["tags"],
            "scenario": m["scenario"][:100]  # 只取前100字符
        } for m in metadata_list
    ]

    prompt = f"""
你是一个知识检索专家。根据用户的当前任务需求，从下列原子知识元数据中挑选出最相关的最多 {routing_k} 个知识 ID。
任务需求："{query_text}"

可选知识列表：
{json.dumps(routing_data, ensure_ascii=False, indent=1)}

请直接输出 ID 列表，用逗号分隔（例如: knowledge-20260302-001, research-20260302-002）。若无相关项请输出 "None"。
"""

    try:
        print(f"\n[Step 1: 知识语义路由] 任务: '{query_text}' | 候选总数: {len(metadata_list)} | 目标提取数: {routing_k}")

        response = await openrouter_llm_call(
            messages=[{"role": "user", "content": prompt}],
            model="google/gemini-2.0-flash-001"
        )

        content = response.get("content", "").strip()
        selected_ids = [idx.strip() for idx in re.split(r'[,\s]+', content) if idx.strip().startswith(("knowledge-", "research-"))]

        print(f"[Step 1: 知识语义路由] LLM 初选 ID ({len(selected_ids)}个): {selected_ids}")
        return selected_ids
    except Exception as e:
        logger.error(f"LLM 知识路由失败: {e}")
        return []


async def _get_structured_knowledge(
    query_text: str,
    top_k: int = 5,
    min_score: int = 3,
    context: Optional[Any] = None,
    tags_filter: Optional[List[str]] = None
) -> List[Dict]:
    """
    语义检索原子知识（包括经验）

    1. 解析知识库文件（支持 JSON 和 YAML 格式）
    2. 语义路由：提取 2*k 个 ID
    3. 质量精排：基于评分筛选出最终的 k 个

    Args:
        query_text: 查询文本
        top_k: 返回数量
        min_score: 最低评分过滤
        context: 上下文（兼容 experience 接口）
        tags_filter: 标签过滤（如 ["strategy"] 只返回经验）
    """
    knowledge_dir = Path(".cache/knowledge_atoms")

    if not knowledge_dir.exists():
        print(f"[Knowledge System] 警告: 知识库目录不存在 ({knowledge_dir})")
        return []

    # 同时支持 .json 和 .md 文件
    json_files = list(knowledge_dir.glob("*.json"))
    md_files = list(knowledge_dir.glob("*.md"))
    files = json_files + md_files

    if not files:
        print(f"[Knowledge System] 警告: 知识库为空")
        return []

    # --- 阶段 1: 解析所有知识文件 ---
    content_map = {}
    metadata_list = []

    for file_path in files:
        try:
            with open(file_path, "r", encoding="utf-8") as f:
                content = f.read()

            # 根据文件扩展名选择解析方式
            if file_path.suffix == ".json":
                # 解析 JSON 格式
                metadata = json.loads(content)
            else:
                # 解析 YAML frontmatter（兼容旧格式）
                yaml_match = re.search(r'^---\n(.*?)\n---', content, re.DOTALL)
                if not yaml_match:
                    logger.warning(f"跳过无效文件: {file_path}")
                    continue
                metadata = yaml.safe_load(yaml_match.group(1))

            if not isinstance(metadata, dict):
                logger.warning(f"跳过损坏的知识文件: {file_path}")
                continue

            kid = metadata.get("id")
            if not kid:
                logger.warning(f"跳过缺少 id 的知识文件: {file_path}")
                continue

            # 提取 scenario 和 content
            scenario = metadata.get("scenario", "").strip()
            content_text = metadata.get("content", "").strip()

            # 标签过滤
            tags = metadata.get("tags", {})
            if tags_filter:
                # 检查 tags.type 是否包含任何过滤标签
                tag_types = tags.get("type", [])
                if isinstance(tag_types, str):
                    tag_types = [tag_types]
                if not any(tag in tag_types for tag in tags_filter):
                    continue  # 跳过不匹配的标签

            meta_item = {
                "id": kid,
                "tags": tags,
                "scenario": scenario,
                "score": metadata.get("eval", {}).get("score", 3),
                "helpful": metadata.get("metrics", {}).get("helpful", 0),
                "harmful": metadata.get("metrics", {}).get("harmful", 0),
            }
            metadata_list.append(meta_item)
            content_map[kid] = {
                "scenario": scenario,
                "content": content_text,
                "tags": tags,
                "score": meta_item["score"],
                "helpful": meta_item["helpful"],
                "harmful": meta_item["harmful"],
            }
        except Exception as e:
            logger.error(f"解析知识文件失败 {file_path}: {e}")
            continue

    if not metadata_list:
        print(f"[Knowledge System] 警告: 没有有效的知识条目")
        return []

    # --- 阶段 2: 语义路由 (取 2*k) ---
    candidate_ids = await _route_knowledge_by_llm(query_text, metadata_list, k=top_k)

    # --- 阶段 3: 质量精排 (根据评分和反馈选出最终的 k) ---
    print(f"[Step 2: 知识质量精排] 正在根据评分和反馈进行打分...")
    scored_items = []

    for kid in candidate_ids:
        if kid in content_map:
            item = content_map[kid]
            score = item["score"]
            helpful = item["helpful"]
            harmful = item["harmful"]

            # 计算综合分：基础分 + helpful - harmful*2
            quality_score = score + helpful - (harmful * 2.0)

            # 过滤门槛：评分低于 min_score 或质量分过低
            if score < min_score or quality_score < 0:
                print(f"  - 剔除低质量知识: {kid} (Score: {score}, Helpful: {helpful}, Harmful: {harmful})")
                continue

            scored_items.append({
                "id": kid,
                "scenario": item["scenario"],
                "content": item["content"],
                "tags": item["tags"],
                "score": score,
                "quality_score": quality_score,
                "metrics": {
                    "helpful": helpful,
                    "harmful": harmful
                }
            })

    # 按照质量分排序
    final_sorted = sorted(scored_items, key=lambda x: x["quality_score"], reverse=True)

    # 截取最终的 top_k
    result = final_sorted[:top_k]

    print(f"[Step 2: 知识质量精排] 最终选定知识: {[it['id'] for it in result]}")
    print(f"[Knowledge System] 检索结束。\n")
    return result


@tool()
async def search_knowledge(
    query: str,
    top_k: int = 5,
    min_score: int = 3,
    tags_type: Optional[List[str]] = None,
    context: Optional[ToolContext] = None,
) -> ToolResult:
    """
    语义检索原子知识库

    Args:
        query: 搜索查询（任务描述）
        top_k: 返回数量（默认 5）
        min_score: 最低评分过滤（默认 3）
        tags_type: 按类型过滤（tool/usercase/definition/plan）
        context: 工具上下文

    Returns:
        相关知识列表
    """
    try:
        relevant_items = await _get_structured_knowledge(
            query_text=query,
            top_k=top_k,
            min_score=min_score
        )

        if not relevant_items:
            return ToolResult(
                title="🔍 未找到相关知识",
                output=f"查询: {query}\n\n知识库中暂无相关的高质量知识。建议进行调研。",
                long_term_memory=f"知识检索: 未找到相关知识 - {query[:50]}"
            )

        # 格式化输出
        output_lines = [f"查询: {query}\n", f"找到 {len(relevant_items)} 条相关知识:\n"]

        for idx, item in enumerate(relevant_items, 1):
            output_lines.append(f"\n### {idx}. [{item['id']}] (⭐ {item['score']})")
            output_lines.append(f"**场景**: {item['scenario'][:150]}...")
            output_lines.append(f"**内容**: {item['content'][:200]}...")

        return ToolResult(
            title="✅ 知识检索成功",
            output="\n".join(output_lines),
            long_term_memory=f"知识检索: 找到 {len(relevant_items)} 条相关知识 - {query[:50]}",
            metadata={
                "count": len(relevant_items),
                "knowledge_ids": [item["id"] for item in relevant_items],
                "items": relevant_items
            }
        )

    except Exception as e:
        logger.error(f"知识检索失败: {e}")
        return ToolResult(
            title="❌ 检索失败",
            output=f"错误: {str(e)}",
            error=str(e)
        )


@tool(description="通过两阶段检索获取最相关的历史经验（strategy 标签的知识）")
async def get_experience(
    query: str,
    k: int = 3,
    context: Optional[ToolContext] = None,
) -> ToolResult:
    """
    检索历史经验（兼容旧接口，实际调用 search_knowledge 并过滤 strategy 标签）

    Args:
        query: 搜索查询（任务描述）
        k: 返回数量（默认 3）
        context: 工具上下文

    Returns:
        相关经验列表
    """
    try:
        relevant_items = await _get_structured_knowledge(
            query_text=query,
            top_k=k,
            min_score=1,  # 经验的评分门槛较低
            context=context,
            tags_filter=["strategy"]  # 只返回经验
        )

        if not relevant_items:
            return ToolResult(
                title="🔍 未找到相关经验",
                output=f"查询: {query}\n\n经验库中暂无相关的经验。",
                long_term_memory=f"经验检索: 未找到相关经验 - {query[:50]}",
                metadata={"items": [], "count": 0}
            )

        # 格式化输出（兼容旧格式）
        output_lines = [f"查询: {query}\n", f"找到 {len(relevant_items)} 条相关经验:\n"]

        for idx, item in enumerate(relevant_items, 1):
            output_lines.append(f"\n### {idx}. [{item['id']}]")
            output_lines.append(f"{item['content'][:300]}...")

        return ToolResult(
            title="✅ 经验检索成功",
            output="\n".join(output_lines),
            long_term_memory=f"经验检索: 找到 {len(relevant_items)} 条相关经验 - {query[:50]}",
            metadata={
                "items": relevant_items,
                "count": len(relevant_items)
            }
        )

    except Exception as e:
        logger.error(f"经验检索失败: {e}")
        return ToolResult(
            title="❌ 检索失败",
            output=f"错误: {str(e)}",
            error=str(e)
        )


# ===== 批量更新功能（类似经验机制）=====

async def _batch_update_knowledge(
    update_map: Dict[str, Dict[str, Any]],
    context: Optional[Any] = None
) -> int:
    """
    内部函数：批量更新知识（兼容 experience 接口）

    Args:
        update_map: 更新映射 {knowledge_id: {"action": "helpful/harmful/evolve", "feedback": "..."}}
        context: 上下文（兼容 experience 接口）

    Returns:
        成功更新的数量
    """
    if not update_map:
        return 0

    knowledge_dir = Path(".cache/knowledge_atoms")
    if not knowledge_dir.exists():
        return 0

    success_count = 0
    evolution_tasks = []
    evolution_registry = {}  # task_idx -> (file_path, data)

    for knowledge_id, instr in update_map.items():
        try:
            # 查找文件
            json_path = knowledge_dir / f"{knowledge_id}.json"
            md_path = knowledge_dir / f"{knowledge_id}.md"

            file_path = None
            is_json = False
            if json_path.exists():
                file_path = json_path
                is_json = True
            elif md_path.exists():
                file_path = md_path
                is_json = False
            else:
                continue

            # 读取并解析
            with open(file_path, "r", encoding="utf-8") as f:
                content = f.read()

            if is_json:
                data = json.loads(content)
            else:
                yaml_match = re.search(r'^---\n(.*?)\n---', content, re.DOTALL)
                if not yaml_match:
                    continue
                data = yaml.safe_load(yaml_match.group(1))

            # 更新 metrics
            action = instr.get("action")
            feedback = instr.get("feedback", "")

            # 处理 mixed 中间态
            if action == "mixed":
                data["metrics"]["helpful"] = data.get("metrics", {}).get("helpful", 0) + 1
                action = "evolve"

            if action == "helpful":
                data["metrics"]["helpful"] = data.get("metrics", {}).get("helpful", 0) + 1
            elif action == "harmful":
                data["metrics"]["harmful"] = data.get("metrics", {}).get("harmful", 0) + 1
            elif action == "evolve" and feedback:
                # 注册进化任务
                old_content = data.get("content", "")
                task = _evolve_knowledge_with_llm(old_content, feedback)
                evolution_tasks.append(task)
                evolution_registry[len(evolution_tasks) - 1] = (file_path, data, is_json)
                data["metrics"]["helpful"] = data.get("metrics", {}).get("helpful", 0) + 1

            data["updated_at"] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

            # 如果不需要进化，直接保存
            if action != "evolve" or not feedback:
                if is_json:
                    with open(file_path, "w", encoding="utf-8") as f:
                        json.dump(data, f, ensure_ascii=False, indent=2)
                else:
                    meta_str = yaml.dump(data, allow_unicode=True).strip()
                    with open(file_path, "w", encoding="utf-8") as f:
                        f.write(f"---\n{meta_str}\n---\n")
                success_count += 1

        except Exception as e:
            logger.error(f"更新知识失败 {knowledge_id}: {e}")
            continue

    # 并发进化
    if evolution_tasks:
        import asyncio
        print(f"🧬 并发处理 {len(evolution_tasks)} 条知识进化...")
        evolved_results = await asyncio.gather(*evolution_tasks)

        # 回填进化结果
        for task_idx, (file_path, data, is_json) in evolution_registry.items():
            data["content"] = evolved_results[task_idx].strip()

            if is_json:
                with open(file_path, "w", encoding="utf-8") as f:
                    json.dump(data, f, ensure_ascii=False, indent=2)
            else:
                meta_str = yaml.dump(data, allow_unicode=True).strip()
                with open(file_path, "w", encoding="utf-8") as f:
                    f.write(f"---\n{meta_str}\n---\n")
            success_count += 1

    return success_count


@tool()
async def batch_update_knowledge(
    feedback_list: List[Dict[str, Any]],
    context: Optional[ToolContext] = None,
) -> ToolResult:
    """
    批量反馈知识的有效性（类似经验机制）

    Args:
        feedback_list: 评价列表，每个元素包含:
            - knowledge_id: (str) 知识 ID
            - is_effective: (bool) 是否有效
            - feedback: (str, optional) 改进建议，若有效且有建议则触发知识进化

    Returns:
        批量更新结果
    """
    try:
        if not feedback_list:
            return ToolResult(
                title="⚠️ 反馈列表为空",
                output="未提供任何反馈",
                long_term_memory="批量更新知识: 反馈列表为空"
            )

        knowledge_dir = Path(".cache/knowledge_atoms")
        if not knowledge_dir.exists():
            return ToolResult(
                title="❌ 知识库不存在",
                output="知识库目录不存在",
                error="知识库不存在"
            )

        success_count = 0
        failed_items = []

        for item in feedback_list:
            knowledge_id = item.get("knowledge_id")
            is_effective = item.get("is_effective")
            feedback = item.get("feedback", "")

            if not knowledge_id:
                failed_items.append({"id": "unknown", "reason": "缺少 knowledge_id"})
                continue

            try:
                # 查找文件
                json_path = knowledge_dir / f"{knowledge_id}.json"
                md_path = knowledge_dir / f"{knowledge_id}.md"

                file_path = None
                is_json = False
                if json_path.exists():
                    file_path = json_path
                    is_json = True
                elif md_path.exists():
                    file_path = md_path
                    is_json = False
                else:
                    failed_items.append({"id": knowledge_id, "reason": "文件不存在"})
                    continue

                # 读取并解析
                with open(file_path, "r", encoding="utf-8") as f:
                    content = f.read()

                if is_json:
                    data = json.loads(content)
                else:
                    yaml_match = re.search(r'^---\n(.*?)\n---', content, re.DOTALL)
                    if not yaml_match:
                        failed_items.append({"id": knowledge_id, "reason": "格式错误"})
                        continue
                    data = yaml.safe_load(yaml_match.group(1))

                # 更新 metrics
                if is_effective:
                    data["metrics"]["helpful"] = data.get("metrics", {}).get("helpful", 0) + 1
                    # 如果有反馈建议，触发进化
                    if feedback:
                        old_content = data.get("content", "")
                        evolved_content = await _evolve_knowledge_with_llm(old_content, feedback)
                        data["content"] = evolved_content
                else:
                    data["metrics"]["harmful"] = data.get("metrics", {}).get("harmful", 0) + 1

                data["updated_at"] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

                # 保存
                if is_json:
                    with open(file_path, "w", encoding="utf-8") as f:
                        json.dump(data, f, ensure_ascii=False, indent=2)
                else:
                    meta_str = yaml.dump(data, allow_unicode=True).strip()
                    with open(file_path, "w", encoding="utf-8") as f:
                        f.write(f"---\n{meta_str}\n---\n")

                success_count += 1

            except Exception as e:
                failed_items.append({"id": knowledge_id, "reason": str(e)})
                continue

        output_lines = [f"成功更新 {success_count} 条知识"]
        if failed_items:
            output_lines.append(f"\n失败 {len(failed_items)} 条:")
            for item in failed_items:
                output_lines.append(f"  - {item['id']}: {item['reason']}")

        return ToolResult(
            title="✅ 批量更新完成",
            output="\n".join(output_lines),
            long_term_memory=f"批量更新知识: 成功 {success_count} 条，失败 {len(failed_items)} 条"
        )

    except Exception as e:
        logger.error(f"批量更新知识失败: {e}")
        return ToolResult(
            title="❌ 批量更新失败",
            output=f"错误: {str(e)}",
            error=str(e)
        )


# ===== 知识库瘦身功能（类似经验机制）=====

@tool()
async def slim_knowledge(
    model: str = "anthropic/claude-sonnet-4.5",
    context: Optional[ToolContext] = None,
) -> ToolResult:
    """
    知识库瘦身：调用顶级大模型，将知识库中语义相似的知识合并精简

    Args:
        model: 使用的模型（默认 claude-sonnet-4.5）
        context: 工具上下文

    Returns:
        瘦身结果报告
    """
    try:
        knowledge_dir = Path(".cache/knowledge_atoms")

        if not knowledge_dir.exists():
            return ToolResult(
                title="📂 知识库不存在",
                output="知识库目录不存在，无需瘦身",
                long_term_memory="知识库瘦身: 目录不存在"
            )

        # 获取所有文件
        json_files = list(knowledge_dir.glob("*.json"))
        md_files = list(knowledge_dir.glob("*.md"))
        files = json_files + md_files

        if len(files) < 2:
            return ToolResult(
                title="📂 知识库过小",
                output=f"知识库仅有 {len(files)} 条，无需瘦身",
                long_term_memory=f"知识库瘦身: 仅有 {len(files)} 条"
            )

        # 解析所有知识
        parsed = []
        for file_path in files:
            try:
                with open(file_path, "r", encoding="utf-8") as f:
                    content = f.read()

                if file_path.suffix == ".json":
                    data = json.loads(content)
                else:
                    yaml_match = re.search(r'^---\n(.*?)\n---', content, re.DOTALL)
                    if not yaml_match:
                        continue
                    data = yaml.safe_load(yaml_match.group(1))

                parsed.append({
                    "file_path": file_path,
                    "data": data,
                    "is_json": file_path.suffix == ".json"
                })
            except Exception as e:
                logger.error(f"解析文件失败 {file_path}: {e}")
                continue

        if len(parsed) < 2:
            return ToolResult(
                title="📂 有效知识过少",
                output=f"有效知识仅有 {len(parsed)} 条，无需瘦身",
                long_term_memory=f"知识库瘦身: 有效知识 {len(parsed)} 条"
            )

        # 构造发给大模型的内容
        entries_text = ""
        for p in parsed:
            data = p["data"]
            entries_text += f"[ID: {data.get('id')}] [Tags: {data.get('tags', {})}] "
            entries_text += f"[Metrics: {data.get('metrics', {})}] [Score: {data.get('eval', {}).get('score', 3)}]\n"
            entries_text += f"Scenario: {data.get('scenario', 'N/A')}\n"
            entries_text += f"Content: {data.get('content', '')[:200]}...\n\n"

        prompt = f"""你是一个 AI Agent 知识库管理员。以下是当前知识库的全部条目，请执行瘦身操作：

【任务】:
1. 识别语义高度相似或重复的知识，将它们合并为一条更精炼、更通用的知识。
2. 合并时保留 helpful 最高的那条的 ID 和 metrics（metrics 中 helpful/harmful 取各条之和）。
3. 对于独立的、无重复的知识，保持原样不动。
4. 保持原有的知识结构和格式。

【当前知识库】:
{entries_text}

【输出格式要求】:
严格按以下格式输出每条知识，条目之间用 === 分隔：
ID: <保留的id>
TAGS: <yaml格式的tags>
METRICS: <yaml格式的metrics>
SCORE: <评分>
SCENARIO: <场景描述>
CONTENT: <合并后的知识内容>
===

最后一行输出合并报告，格式：
REPORT: 原有 X 条，合并后 Y 条，精简了 Z 条。

禁止输出任何开场白或解释。"""

        print(f"\n[知识瘦身] 正在调用 {model} 分析 {len(parsed)} 条知识...")
        response = await openrouter_llm_call(
            messages=[{"role": "user", "content": prompt}],
            model=model
        )
        content = response.get("content", "").strip()
        if not content:
            return ToolResult(
                title="❌ 大模型返回为空",
                output="大模型返回为空，瘦身失败",
                error="大模型返回为空"
            )

        # 解析大模型输出
        report_line = ""
        new_entries = []
        blocks = [b.strip() for b in content.split("===") if b.strip()]

        for block in blocks:
            if block.startswith("REPORT:"):
                report_line = block
                continue

            lines = block.split("\n")
            kid, tags, metrics, score, scenario, content_lines = None, {}, {}, 3, "", []
            current_field = None

            for line in lines:
                if line.startswith("ID:"):
                    kid = line[3:].strip()
                    current_field = None
                elif line.startswith("TAGS:"):
                    try:
                        tags = yaml.safe_load(line[5:].strip()) or {}
                    except Exception:
                        tags = {}
                    current_field = None
                elif line.startswith("METRICS:"):
                    try:
                        metrics = yaml.safe_load(line[8:].strip()) or {}
                    except Exception:
                        metrics = {"helpful": 0, "harmful": 0}
                    current_field = None
                elif line.startswith("SCORE:"):
                    try:
                        score = int(line[6:].strip())
                    except Exception:
                        score = 3
                    current_field = None
                elif line.startswith("SCENARIO:"):
                    scenario = line[9:].strip()
                    current_field = "scenario"
                elif line.startswith("CONTENT:"):
                    content_lines.append(line[8:].strip())
                    current_field = "content"
                elif current_field == "scenario":
                    scenario += "\n" + line
                elif current_field == "content":
                    content_lines.append(line)

            if kid and content_lines:
                new_data = {
                    "id": kid,
                    "tags": tags,
                    "scenario": scenario,
                    "content": "\n".join(content_lines).strip(),
                    "metrics": metrics,
                    "eval": {
                        "score": score,
                        "helpful": 0,
                        "harmful": 0,
                        "helpful_history": [],
                        "harmful_history": []
                    },
                    "updated_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
                }
                new_entries.append(new_data)

        if not new_entries:
            return ToolResult(
                title="❌ 解析失败",
                output="解析大模型输出失败，知识库未修改",
                error="解析失败"
            )

        # 删除旧文件
        for p in parsed:
            try:
                p["file_path"].unlink()
            except Exception as e:
                logger.error(f"删除旧文件失败 {p['file_path']}: {e}")

        # 写入新文件（统一使用 JSON 格式）
        for data in new_entries:
            file_path = knowledge_dir / f"{data['id']}.json"
            with open(file_path, "w", encoding="utf-8") as f:
                json.dump(data, f, ensure_ascii=False, indent=2)

        result = f"瘦身完成：{len(parsed)} → {len(new_entries)} 条知识"
        if report_line:
            result += f"\n{report_line}"

        print(f"[知识瘦身] {result}")
        return ToolResult(
            title="✅ 知识库瘦身完成",
            output=result,
            long_term_memory=f"知识库瘦身: {len(parsed)} → {len(new_entries)} 条"
        )

    except Exception as e:
        logger.error(f"知识库瘦身失败: {e}")
        return ToolResult(
            title="❌ 瘦身失败",
            output=f"错误: {str(e)}",
            error=str(e)
        )