Openclaw Memory Enhancer

#!/usr/bin/env python3
"""
知识库构建器 - Knowledge Base Builder
功能：
1. 自动整理聊天记录，提取有价值信息
2. 生成 FAQ 问答对
3. 构建个人 wiki/知识库
"""

import os
import json
import re
from datetime import datetime
from pathlib import Path
from typing import List, Dict, Tuple, Optional
import hashlib

class KnowledgeBase:
    """知识库构建器"""
    
    def __init__(self):
        self.home = Path.home()
        self.kb_dir = Path.home() / ".openclaw" / "workspace" / "knowledge-base"
        self.kb_dir.mkdir(parents=True, exist_ok=True)
        
        # 子目录
        self.faqs_dir = self.kb_dir / "faqs"
        self.wiki_dir = self.kb_dir / "wiki"
        self.raw_dir = self.kb_dir / "raw"
        self.index_file = self.kb_dir / "index.json"
        
        for d in [self.faqs_dir, self.wiki_dir, self.raw_dir]:
            d.mkdir(exist_ok=True)
        
        self.load_index()
    
    def load_index(self):
        """加载知识库索引"""
        if self.index_file.exists():
            with open(self.index_file) as f:
                self.index = json.load(f)
        else:
            self.index = {
                "created_at": datetime.now().isoformat(),
                "last_updated": datetime.now().isoformat(),
                "total_entries": 0,
                "categories": {},
                "faqs": [],
                "wiki_pages": []
            }
            self.save_index()
    
    def save_index(self):
        """保存知识库索引"""
        self.index["last_updated"] = datetime.now().isoformat()
        with open(self.index_file, "w") as f:
            json.dump(self.index, f, indent=2, ensure_ascii=False)
    
    def extract_from_chat(self, chat_content: str, source: str = "unknown") -> List[Dict]:
        """
        从聊天记录中提取知识条目
        
        Args:
            chat_content: 聊天内容
            source: 来源标识
            
        Returns:
            提取的知识条目列表
        """
        entries = []
        
        # 提取问答对（Q: ... A: ... 格式）
        qa_pattern = r'(?:Q|问|问题)[：:]\s*(.+?)\n(?:A|答|回答)[：:]\s*(.+?)(?=\n(?:Q|问|问题)[：:]|$)'
        for match in re.finditer(qa_pattern, chat_content, re.DOTALL | re.IGNORECASE):
            entries.append({
                "type": "qa",
                "question": match.group(1).strip(),
                "answer": match.group(2).strip(),
                "source": source,
                "extracted_at": datetime.now().isoformat()
            })
        
        # 提取重要指令（"记住...", "以后..." 等）
        instruction_patterns = [
            r'(?:记住|记住这个|请记住)[：:]\s*(.+?)(?=\n|$)',
            r'(?:以后|下次|以后每次)[：:]\s*(.+?)(?=\n|$)',
            r'(?:重要|注意)[：:]\s*(.+?)(?=\n|$)',
        ]
        for pattern in instruction_patterns:
            for match in re.finditer(pattern, chat_content, re.IGNORECASE):
                entries.append({
                    "type": "instruction",
                    "content": match.group(1).strip(),
                    "source": source,
                    "extracted_at": datetime.now().isoformat()
                })
        
        # 提取技术方案/解决方案
        solution_pattern = r'(?:方案|解决方案|步骤)[：:]\s*\n?(.+?)(?=\n\n|\Z)'
        for match in re.finditer(solution_pattern, chat_content, re.DOTALL | re.IGNORECASE):
            content = match.group(1).strip()
            if len(content) > 50:  # 过滤太短的
                entries.append({
                    "type": "solution",
                    "content": content[:500],  # 限制长度
                    "source": source,
                    "extracted_at": datetime.now().isoformat()
                })
        
        return entries
    
    def add_to_faq(self, question: str, answer: str, category: str = "general") -> str:
        """
        添加 FAQ 条目
        
        Returns:
            entry_id: 条目ID
        """
        entry_id = hashlib.md5(f"{question}{datetime.now()}".encode()).hexdigest()[:12]
        
        faq_entry = {
            "id": entry_id,
            "question": question,
            "answer": answer,
            "category": category,
            "created_at": datetime.now().isoformat(),
            "updated_at": datetime.now().isoformat(),
            "usage_count": 0
        }
        
        # 保存到文件
        faq_file = self.faqs_dir / f"{category}_{entry_id}.json"
        with open(faq_file, "w", encoding="utf-8") as f:
            json.dump(faq_entry, f, indent=2, ensure_ascii=False)
        
        # 更新索引
        self.index["faqs"].append({
            "id": entry_id,
            "question": question,
            "category": category,
            "file": str(faq_file.relative_to(self.kb_dir))
        })
        self.index["total_entries"] += 1
        self.save_index()
        
        return entry_id
    
    def create_wiki_page(self, title: str, content: str, tags: List[str] = None) -> str:
        """
        创建 Wiki 页面
        
        Returns:
            page_id: 页面ID
        """
        page_id = hashlib.md5(f"{title}{datetime.now()}".encode()).hexdigest()[:12]
        
        # 生成 Markdown 文件
        filename = f"{title.replace(' ', '_').replace('/', '_')}.md"
        page_file = self.wiki_dir / filename
        
        md_content = f"""# {title}

> 创建时间: {datetime.now().strftime("%Y-%m-%d %H:%M")}
> 标签: {', '.join(tags or [])}

---

{content}

---

*由 KnowledgeBase Builder 自动生成*
"""
        
        with open(page_file, "w", encoding="utf-8") as f:
            f.write(md_content)
        
        # 更新索引
        self.index["wiki_pages"].append({
            "id": page_id,
            "title": title,
            "file": str(page_file.relative_to(self.kb_dir)),
            "tags": tags or [],
            "created_at": datetime.now().isoformat()
        })
        self.save_index()
        
        return page_id
    
    def search_faqs(self, query: str, limit: int = 5) -> List[Dict]:
        """
        搜索 FAQ
        
        简单实现：基于关键词匹配
        """
        results = []
        query_lower = query.lower()
        
        for faq_info in self.index.get("faqs", []):
            faq_file = self.kb_dir / faq_info["file"]
            if faq_file.exists():
                with open(faq_file) as f:
                    faq = json.load(f)
                
                # 简单匹配
                question_lower = faq["question"].lower()
                answer_lower = faq["answer"].lower()
                
                if any(keyword in question_lower or keyword in answer_lower 
                       for keyword in query_lower.split()):
                    results.append(faq)
                    
                    if len(results) >= limit:
                        break
        
        return results
    
    def get_stats(self) -> Dict:
        """获取知识库统计"""
        return {
            "total_entries": self.index["total_entries"],
            "total_faqs": len(self.index.get("faqs", [])),
            "total_wiki_pages": len(self.index.get("wiki_pages", [])),
            "categories": list(self.index.get("categories", {}).keys()),
            "last_updated": self.index["last_updated"],
            "kb_dir": str(self.kb_dir)
        }
    
    def export_to_markdown(self, output_file: str = None) -> str:
        """
        导出知识库为 Markdown 文档
        """
        if output_file is None:
            output_file = self.kb_dir / f"knowledge_base_{datetime.now().strftime('%Y%m%d')}.md"
        
        lines = [
            "# 📚 知识库",
            "",
            f"> 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}",
            f"> 总条目数: {self.index['total_entries']}",
            "",
            "---",
            "",
            "## 📋 FAQ 问答",
            ""
        ]
        
        # FAQ 部分
        for faq_info in self.index.get("faqs", []):
            faq_file = self.kb_dir / faq_info["file"]
            if faq_file.exists():
                with open(faq_file) as f:
                    faq = json.load(f)
                lines.extend([
                    f"### Q: {faq['question']}",
                    "",
                    f"{faq['answer']}",
                    "",
                    f"*分类: {faq['category']} | 更新时间: {faq['updated_at'][:10]}*",
                    "",
                    "---",
                    ""
                ])
        
        # Wiki 部分
        lines.extend([
            "",
            "## 📝 Wiki 页面",
            ""
        ])
        
        for page_info in self.index.get("wiki_pages", []):
            lines.extend([
                f"- **{page_info['title']}**",
                f"  - 标签: {', '.join(page_info.get('tags', []))}",
                f"  - 文件: `{page_info['file']}`",
                ""
            ])
        
        content = "\n".join(lines)
        
        with open(output_file, "w", encoding="utf-8") as f:
            f.write(content)
        
        return str(output_file)
    
    def auto_build_from_memory(self):
        """
        自动从 memory 目录构建知识库
        """
        memory_dir = Path.home() / ".openclaw" / "workspace" / "memory"
        
        if not memory_dir.exists():
            return {"error": "memory 目录不存在"}
        
        stats = {
            "processed_files": 0,
            "extracted_entries": 0,
            "added_faqs": 0,
            "created_wiki": 0
        }
        
        # 处理每日记忆文件
        for md_file in memory_dir.glob("2026-*.md"):
            stats["processed_files"] += 1
            
            with open(md_file, "r", encoding="utf-8") as f:
                content = f.read()
            
            # 提取知识
            entries = self.extract_from_chat(content, source=md_file.name)
            stats["extracted_entries"] += len(entries)
            
            # 将 QA 添加到 FAQ
            for entry in entries:
                if entry["type"] == "qa":
                    self.add_to_faq(
                        entry["question"],
                        entry["answer"],
                        category="auto_extracted"
                    )
                    stats["added_faqs"] += 1
        
        return stats


def main():
    """命令行接口"""
    import argparse
    
    parser = argparse.ArgumentParser(description="知识库构建器")
    parser.add_argument("action", choices=[
        "extract", "add-faq", "create-wiki", "search", "stats", "export", "auto-build"
    ], help="操作类型")
    parser.add_argument("--file", "-f", help="聊天文件路径")
    parser.add_argument("--question", "-q", help="FAQ 问题")
    parser.add_argument("--answer", "-a", help="FAQ 答案")
    parser.add_argument("--title", "-t", help="Wiki 标题")
    parser.add_argument("--content", "-c", help="Wiki 内容或聊天内容")
    parser.add_argument("--category", default="general", help="分类")
    parser.add_argument("--tags", help="标签（逗号分隔）")
    
    args = parser.parse_args()
    
    kb = KnowledgeBase()
    
    if args.action == "extract":
        if not args.content and not args.file:
            print("❌ 请提供 --content 或 --file")
            return
        
        content = args.content
        if args.file:
            with open(args.file) as f:
                content = f.read()
        
        entries = kb.extract_from_chat(content, args.file or "cli")
        print(f"✅ 提取了 {len(entries)} 条知识:")
        for i, e in enumerate(entries[:5], 1):
            print(f"\n  {i}. [{e['type']}] {str(e)[:100]}...")
        if len(entries) > 5:
            print(f"\n  ... 还有 {len(entries) - 5} 条")
    
    elif args.action == "add-faq":
        if not args.question or not args.answer:
            print("❌ 请提供 --question 和 --answer")
            return
        entry_id = kb.add_to_faq(args.question, args.answer, args.category)
        print(f"✅ FAQ 已添加 (ID: {entry_id})")
    
    elif args.action == "create-wiki":
        if not args.title or not args.content:
            print("❌ 请提供 --title 和 --content")
            return
        tags = args.tags.split(",") if args.tags else []
        page_id = kb.create_wiki_page(args.title, args.content, tags)
        print(f"✅ Wiki 页面已创建 (ID: {page_id})")
    
    elif args.action == "search":
        if not args.question:
            print("❌ 请提供 --question 作为搜索关键词")
            return
        results = kb.search_faqs(args.question)
        print(f"🔍 找到 {len(results)} 个相关 FAQ:")
        for r in results:
            print(f"\n  Q: {r['question']}")
            print(f"  A: {r['answer'][:100]}...")
    
    elif args.action == "stats":
        stats = kb.get_stats()
        print("📊 知识库统计:")
        for k, v in stats.items():
            print(f"   {k}: {v}")
    
    elif args.action == "export":
        output = kb.export_to_markdown()
        print(f"✅ 知识库已导出: {output}")
    
    elif args.action == "auto-build":
        print("🔄 自动构建知识库...")
        result = kb.auto_build_from_memory()
        print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()

MIT License

Copyright (c) 2025 Henry

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

#!/usr/bin/env python3
"""
OpenClaw Memory Enhancer - Edge Optimized Version
边缘计算优化版本：内存占用 < 10MB，无需外部模型依赖
"""

import os
import json
import re
import hashlib
from datetime import datetime
from pathlib import Path
from typing import List, Dict, Optional, Tuple

class MemoryEnhancerEdge:
    """
    OpenClaw 记忆增强器 - 边缘计算优化版
    
    特点：
    - 零外部依赖（无需 sentence-transformers）
    - 内存占用 < 10MB
    - 纯本地计算，无网络请求
    - 适合 Jetson / Raspberry Pi / 嵌入式设备
    """
    
    # 版本信息
    VERSION = "2.0-edge"
    
    def __init__(self, workspace_path: str = None):
        """
        初始化
        
        Args:
            workspace_path: OpenClaw workspace 路径，默认 ~/.openclaw/workspace
        """
        if workspace_path:
            self.workspace = Path(workspace_path)
        else:
            self.workspace = Path.home() / ".openclaw" / "workspace"
        
        self.memory_dir = self.workspace / "memory"
        self.kb_dir = self.workspace / "knowledge-base"
        self.vector_dir = self.kb_dir / "vectors"
        
        # 确保目录存在
        self.vector_dir.mkdir(parents=True, exist_ok=True)
        
        # 文件路径
        self.index_file = self.kb_dir / "memory_index.json"
        self.embeddings_file = self.vector_dir / "embeddings.json"  # 使用JSON而非npy，更轻量
        self.metadata_file = self.vector_dir / "metadata.json"
        
        # 配置：边缘优化参数
        self.config = {
            "vector_dim": 128,  # 降低维度，节省内存（标准版384）
            "max_memory_size": 1000,  # 最大记忆条数限制
            "chunk_size": 500,  # 内容分块大小
            "min_keyword_len": 2,  # 最小关键词长度
        }
        
        # 加载数据
        self.embeddings = []  # 使用list而非numpy数组，节省内存
        self.metadata = []
        self.index = {}
        
        self._load_all()
        
        print(f"✅ Memory Enhancer Edge v{self.VERSION} 已初始化")
        print(f"   工作目录: {self.workspace}")
        print(f"   当前记忆: {len(self.metadata)} 条")
    
    def _load_all(self):
        """加载所有数据"""
        # 加载索引
        if self.index_file.exists():
            try:
                with open(self.index_file, 'r', encoding='utf-8') as f:
                    self.index = json.load(f)
            except:
                self.index = self._create_default_index()
        else:
            self.index = self._create_default_index()
        
        # 加载向量（使用JSON格式，更轻量）
        if self.embeddings_file.exists():
            try:
                with open(self.embeddings_file, 'r', encoding='utf-8') as f:
                    self.embeddings = json.load(f)
            except:
                self.embeddings = []
        
        # 加载元数据
        if self.metadata_file.exists():
            try:
                with open(self.metadata_file, 'r', encoding='utf-8') as f:
                    self.metadata = json.load(f)
            except:
                self.metadata = []
    
    def _create_default_index(self) -> Dict:
        """创建默认索引"""
        return {
            "version": self.VERSION,
            "created_at": datetime.now().isoformat(),
            "last_updated": datetime.now().isoformat(),
            "total_memories": 0,
            "sources": [],
            "edge_optimized": True
        }
    
    def _save_all(self):
        """保存所有数据（原子操作）"""
        # 更新索引
        self.index["last_updated"] = datetime.now().isoformat()
        self.index["total_memories"] = len(self.metadata)
        
        # 保存为JSON（边缘设备友好）
        with open(self.index_file, 'w', encoding='utf-8') as f:
            json.dump(self.index, f, indent=2, ensure_ascii=False)
        
        with open(self.embeddings_file, 'w', encoding='utf-8') as f:
            json.dump(self.embeddings, f)
        
        with open(self.metadata_file, 'w', encoding='utf-8') as f:
            json.dump(self.metadata, f, indent=2, ensure_ascii=False)
    
    def _extract_keywords(self, text: str) -> List[str]:
        """
        提取关键词（轻量级，无需NLP库）
        
        使用简单但有效的规则：
        1. 转换为小写
        2. 分词（按非字母数字字符分割）
        3. 过滤短词和常见停用词
        4. 去重
        """
        # 简单的停用词列表（中文+英文）
        stopwords = {'the', 'a', 'an', 'is', 'are', 'was', 'were', 
                     '的', '了', '和', '是', '在', '有', '我', '你',
                     '它', '我们', '你们', '这个', '那个', '可以',
                     '使用', '进行', '通过', '需要', '进行'}
        
        # 分词
        words = re.findall(r'[\u4e00-\u9fa5]+|[a-zA-Z]+', text.lower())
        
        # 过滤
        keywords = []
        for word in words:
            if len(word) >= self.config["min_keyword_len"] and word not in stopwords:
                keywords.append(word)
        
        return list(set(keywords))  # 去重
    
    def _encode_text(self, text: str) -> List[float]:
        """
        将文本编码为向量（边缘优化版）
        
        使用关键词哈希 + TF-IDF 思想的简化版
        无需外部模型，纯本地计算
        """
        keywords = self._extract_keywords(text)
        
        # 创建稀疏向量（使用哈希）
        vector = [0.0] * self.config["vector_dim"]
        
        for keyword in keywords:
            # 使用哈希将关键词映射到向量位置
            hash_val = hash(keyword) % self.config["vector_dim"]
            vector[hash_val] += 1.0  # 简单的词频统计
        
        # 归一化（L2范数）
        magnitude = sum(x**2 for x in vector) ** 0.5
        if magnitude > 0:
            vector = [x / magnitude for x in vector]
        
        return vector
    
    def _cosine_similarity(self, vec1: List[float], vec2: List[float]) -> float:
        """计算余弦相似度"""
        dot_product = sum(a * b for a, b in zip(vec1, vec2))
        magnitude1 = sum(x**2 for x in vec1) ** 0.5
        magnitude2 = sum(x**2 for x in vec2) ** 0.5
        
        if magnitude1 == 0 or magnitude2 == 0:
            return 0.0
        
        return dot_product / (magnitude1 * magnitude2)
    
    def add_memory(self, content: str, source: str = "manual", 
                   memory_type: str = "general", metadata: Dict = None) -> str:
        """
        添加记忆
        
        Args:
            content: 记忆内容
            source: 来源标识
            memory_type: 记忆类型
            metadata: 额外元数据
        
        Returns:
            记忆ID
        """
        # 限制最大记忆数（边缘设备保护）
        if len(self.metadata) >= self.config["max_memory_size"]:
            # 移除最旧的记忆
            self.metadata.pop(0)
            self.embeddings.pop(0)
            print(f"⚠️  记忆数量达到上限，移除最旧的记忆")
        
        # 生成ID
        memory_id = hashlib.md5(f"{content}{datetime.now()}".encode()).hexdigest()[:12]
        
        # 截断过长内容（边缘设备保护）
        if len(content) > self.config["chunk_size"]:
            content = content[:self.config["chunk_size"]] + "..."
        
        # 创建记忆条目
        memory_entry = {
            "id": memory_id,
            "content": content,
            "source": source,
            "type": memory_type,
            "created_at": datetime.now().isoformat(),
            "keywords": self._extract_keywords(content),  # 保存关键词，加速检索
            "metadata": metadata or {}
        }
        
        # 编码
        embedding = self._encode_text(content)
        
        # 添加
        self.metadata.append(memory_entry)
        self.embeddings.append(embedding)
        
        # 更新来源记录
        if source not in self.index["sources"]:
            self.index["sources"].append(source)
        
        # 保存
        self._save_all()
        
        return memory_id
    
    def search_memory(self, query: str, top_k: int = 5, threshold: float = 0.3) -> List[Dict]:
        """
        搜索记忆
        
        Args:
            query: 查询内容
            top_k: 返回数量
            threshold: 相似度阈值
        
        Returns:
            相关记忆列表
        """
        if not self.metadata:
            return []
        
        # 编码查询
        query_vec = self._encode_text(query)
        query_keywords = set(self._extract_keywords(query))
        
        # 计算相似度（优化：先用关键词过滤）
        scored_memories = []
        
        for idx, (mem, emb) in enumerate(zip(self.metadata, self.embeddings)):
            # 快速预过滤：检查是否有共同关键词
            mem_keywords = set(mem.get("keywords", []))
            if not query_keywords.intersection(mem_keywords):
                # 没有共同关键词，跳过详细计算
                continue
            
            # 计算向量相似度
            similarity = self._cosine_similarity(query_vec, emb)
            
            if similarity >= threshold:
                scored_memories.append((idx, similarity))
        
        # 排序取Top-K
        scored_memories.sort(key=lambda x: x[1], reverse=True)
        top_results = scored_memories[:top_k]
        
        # 构建结果
        results = []
        for idx, score in top_results:
            mem = self.metadata[idx].copy()
            mem["similarity"] = round(score, 3)
            results.append(mem)
        
        return results
    
    def recall_for_prompt(self, user_input: str, max_memories: int = 3) -> str:
        """
        为对话召回相关记忆
        
        Args:
            user_input: 用户输入
            max_memories: 最多召回几条
        
        Returns:
            格式化的记忆上下文
        """
        memories = self.search_memory(user_input, top_k=max_memories, threshold=0.25)
        
        if not memories:
            return ""
        
        context = "\n[相关记忆]\n"
        for i, mem in enumerate(memories, 1):
            # 截断内容，保持简洁
            content = mem['content'][:150] + "..." if len(mem['content']) > 150 else mem['content']
            context += f"{i}. {content}\n"
        
        return context
    
    def load_openclaw_memory(self) -> int:
        """
        加载 OpenClaw 记忆文件
        
        Returns:
            加载的记忆数量
        """
        loaded_count = 0
        
        if not self.memory_dir.exists():
            print(f"⚠️  记忆目录不存在: {self.memory_dir}")
            return 0
        
        # 加载 .md 文件
        for md_file in sorted(self.memory_dir.glob("*.md")):
            try:
                with open(md_file, 'r', encoding='utf-8') as f:
                    content = f.read()
                
                # 分割段落
                paragraphs = content.split('\n\n')
                for para in paragraphs:
                    para = para.strip()
                    # 过滤：有意义且不太长的段落
                    if 50 < len(para) < self.config["chunk_size"]:
                        self.add_memory(
                            content=para,
                            source=f"memory/{md_file.name}",
                            memory_type="daily_log"
                        )
                        loaded_count += 1
                
                print(f"✅ 已加载 {md_file.name}")
                
            except Exception as e:
                print(f"⚠️  加载 {md_file.name} 失败: {e}")
        
        # 保存
        self._save_all()
        print(f"\n📊 共加载 {loaded_count} 条记忆")
        return loaded_count
    
    def extract_from_chat(self, chat_content: str, source: str = "chat") -> List[str]:
        """从聊天记录中提取记忆"""
        memory_ids = []
        
        # 提取问答对（简化版）
        qa_pattern = r'(?:Q|问)[：:]\s*(.+?)\n+(?:A|答)[：:]\s*(.+?)(?=\n(?:Q|问)[：:]|\Z)'
        for match in re.finditer(qa_pattern, chat_content, re.DOTALL | re.IGNORECASE):
            question = match.group(1).strip()
            answer = match.group(2).strip()
            
            if len(question) > 10 and len(answer) > 20:
                content = f"Q: {question[:100]}\nA: {answer[:200]}"
                mid = self.add_memory(content=content, source=source, memory_type="qa")
                memory_ids.append(mid)
        
        return memory_ids
    
    def get_stats(self) -> Dict:
        """获取统计信息"""
        return {
            "version": self.VERSION,
            "total_memories": len(self.metadata),
            "vector_dim": self.config["vector_dim"],
            "sources": list(set(m["source"] for m in self.metadata)),
            "edge_optimized": True
        }
    
    def export_memories(self, output_file: str = None) -> str:
        """导出记忆"""
        if output_file is None:
            output_file = self.kb_dir / f"memories_export_{datetime.now().strftime('%Y%m%d')}.md"
        
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write(f"# OpenClaw 记忆导出 (Edge版)\n\n")
            f.write(f"导出时间: {datetime.now().isoformat()}\n")
            f.write(f"记忆总数: {len(self.metadata)}\n")
            f.write(f"向量维度: {self.config['vector_dim']}\n")
            f.write(f"边缘优化: 是\n\n")
            f.write("---\n\n")
            
            for mem in sorted(self.metadata, key=lambda x: x["created_at"], reverse=True):
                f.write(f"## {mem['id']}\n\n")
                f.write(f"**类型**: {mem['type']} | **来源**: {mem['source']}\n\n")
                f.write(f"**关键词**: {', '.join(mem.get('keywords', []))}\n\n")
                f.write(f"**内容**:\n\n{mem['content']}\n\n")
                f.write("---\n\n")
        
        return str(output_file)


# CLI 接口
if __name__ == "__main__":
    import argparse
    
    parser = argparse.ArgumentParser(description="OpenClaw Memory Enhancer (Edge)")
    parser.add_argument("--load", action="store_true", help="加载 OpenClaw 记忆")
    parser.add_argument("--search", type=str, help="搜索记忆")
    parser.add_argument("--add", type=str, help="添加记忆")
    parser.add_argument("--export", action="store_true", help="导出记忆")
    parser.add_argument("--stats", action="store_true", help="显示统计")
    parser.add_argument("--workspace", type=str, help="指定 workspace 路径")
    
    args = parser.parse_args()
    
    enhancer = MemoryEnhancerEdge(workspace_path=args.workspace)
    
    if args.load:
        print("🔄 加载 OpenClaw 记忆文件...")
        count = enhancer.load_openclaw_memory()
        print(f"✅ 完成！共加载 {count} 条记忆")
    
    elif args.search:
        print(f"🔍 搜索: {args.search}")
        results = enhancer.search_memory(args.search)
        for i, r in enumerate(results, 1):
            print(f"\n{i}. [{r['similarity']:.3f}] {r['content'][:80]}...")
            print(f"   关键词: {', '.join(r.get('keywords', [])[:5])}")
    
    elif args.add:
        mid = enhancer.add_memory(args.add)
        print(f"✅ 记忆已添加，ID: {mid}")
    
    elif args.export:
        file = enhancer.export_memories()
        print(f"✅ 已导出到: {file}")
    
    elif args.stats:
        stats = enhancer.get_stats()
        print(f"📊 记忆统计 (Edge版):")
        for k, v in stats.items():
            print(f"  {k}: {v}")
    
    else:
        parser.print_help()

#!/usr/bin/env python3
"""
OpenClaw 记忆增强器 - Memory Enhancer with RAG
功能：
1. 向量语义检索（Sentence Transformers）
2. 自动加载 OpenClaw 记忆文件
3. 对话中智能召回相关记忆
4. 记忆关联图谱构建
"""

import os
import json
import re
import numpy as np
from datetime import datetime
from pathlib import Path
from typing import List, Dict, Tuple, Optional
import hashlib

class MemoryEnhancer:
    """OpenClaw 记忆增强器 - 让 OpenClaw 拥有长期记忆"""
    
    def __init__(self):
        self.home = Path.home()
        self.workspace = self.home / ".openclaw" / "workspace"
        self.memory_dir = self.workspace / "memory"
        self.kb_dir = self.workspace / "knowledge-base"
        self.vector_dir = self.kb_dir / "vectors"
        
        # 确保目录存在
        for d in [self.kb_dir, self.vector_dir, self.kb_dir / "faqs", self.kb_dir / "wiki"]:
            d.mkdir(parents=True, exist_ok=True)
        
        self.index_file = self.kb_dir / "memory_index.json"
        self.embeddings_file = self.vector_dir / "embeddings.npy"
        self.metadata_file = self.vector_dir / "metadata.json"
        
        # 加载或初始化
        self.load_index()
        self.load_embeddings()
        
        # 尝试加载 sentence-transformers
        self.encoder = None
        self._init_encoder()
    
    def _init_encoder(self):
        """初始化向量编码器"""
        try:
            from sentence_transformers import SentenceTransformer
            # 使用轻量级模型，适合本地运行
            self.encoder = SentenceTransformer('paraphrase-MiniLM-L3-v2')
            print("✅ 向量编码器已加载")
        except ImportError:
            print("⚠️  sentence-transformers 未安装，使用简单词频匹配")
            print("     安装: pip install sentence-transformers")
        except Exception as e:
            print(f"⚠️  编码器加载失败: {e}")
    
    def load_index(self):
        """加载记忆索引"""
        if self.index_file.exists():
            with open(self.index_file) as f:
                self.index = json.load(f)
        else:
            self.index = {
                "version": "2.0",
                "created_at": datetime.now().isoformat(),
                "last_updated": datetime.now().isoformat(),
                "total_memories": 0,
                "categories": {},
                "sources": []
            }
            self.save_index()
    
    def save_index(self):
        """保存记忆索引"""
        self.index["last_updated"] = datetime.now().isoformat()
        with open(self.index_file, "w") as f:
            json.dump(self.index, f, indent=2, ensure_ascii=False)
    
    def load_embeddings(self):
        """加载向量嵌入"""
        if self.embeddings_file.exists() and self.metadata_file.exists():
            try:
                self.embeddings = np.load(self.embeddings_file)
                with open(self.metadata_file) as f:
                    self.metadata = json.load(f)
                print(f"✅ 已加载 {len(self.metadata)} 条记忆向量")
            except Exception as e:
                print(f"⚠️  加载向量失败: {e}")
                self.embeddings = np.array([])
                self.metadata = []
        else:
            self.embeddings = np.array([])
            self.metadata = []
    
    def save_embeddings(self):
        """保存向量嵌入"""
        if len(self.embeddings) > 0:
            np.save(self.embeddings_file, self.embeddings)
            with open(self.metadata_file, "w") as f:
                json.dump(self.metadata, f, indent=2, ensure_ascii=False)
    
    def encode_text(self, text: str) -> np.ndarray:
        """将文本编码为向量"""
        if self.encoder:
            return self.encoder.encode(text)
        else:
            # fallback: 简单的词频向量
            words = set(text.lower().split())
            # 创建一个简单的哈希向量
            vec = np.zeros(384)
            for word in words:
                hash_val = hash(word) % 384
                vec[hash_val] = 1
            return vec
    
    def add_memory(self, content: str, source: str = "manual", 
                   memory_type: str = "general", metadata: Dict = None) -> str:
        """
        添加记忆
        
        Args:
            content: 记忆内容
            source: 来源（如 memory/2026-02-22.md, chat, faq）
            memory_type: 类型（general, qa, instruction, solution）
            metadata: 额外元数据
        """
        memory_id = hashlib.md5(f"{content}{datetime.now()}".encode()).hexdigest()[:12]
        
        memory_entry = {
            "id": memory_id,
            "content": content,
            "source": source,
            "type": memory_type,
            "created_at": datetime.now().isoformat(),
            "metadata": metadata or {}
        }
        
        # 编码为向量
        embedding = self.encode_text(content)
        
        # 添加到向量库
        if len(self.embeddings) == 0:
            self.embeddings = np.array([embedding])
        else:
            self.embeddings = np.vstack([self.embeddings, embedding])
        
        self.metadata.append(memory_entry)
        
        # 更新索引
        self.index["total_memories"] = len(self.metadata)
        if source not in self.index["sources"]:
            self.index["sources"].append(source)
        
        # 保存
        self.save_embeddings()
        self.save_index()
        
        return memory_id
    
    def search_memory(self, query: str, top_k: int = 5, threshold: float = 0.5) -> List[Dict]:
        """
        语义搜索记忆
        
        Args:
            query: 查询内容
            top_k: 返回最相关的k条
            threshold: 相似度阈值（0-1）
            
        Returns:
            相关记忆列表
        """
        if len(self.metadata) == 0:
            return []
        
        # 编码查询
        query_vec = self.encode_text(query)
        
        # 计算相似度
        if len(self.embeddings.shape) == 1:
            similarities = np.dot(self.embeddings, query_vec)
        else:
            similarities = np.dot(self.embeddings, query_vec) / (
                np.linalg.norm(self.embeddings, axis=1) * np.linalg.norm(query_vec) + 1e-8
            )
        
        # 获取 top-k
        top_indices = np.argsort(similarities)[::-1][:top_k]
        
        results = []
        for idx in top_indices:
            if similarities[idx] >= threshold:
                memory = self.metadata[idx].copy()
                memory["similarity"] = float(similarities[idx])
                results.append(memory)
        
        return results
    
    def load_openclaw_memory(self):
        """自动加载 OpenClaw 的记忆文件"""
        loaded_count = 0
        
        # 1. 加载 memory/ 目录下的所有 .md 文件
        if self.memory_dir.exists():
            for md_file in self.memory_dir.glob("*.md"):
                try:
                    with open(md_file, 'r', encoding='utf-8') as f:
                        content = f.read()
                    
                    # 按段落分割
                    paragraphs = content.split('\n\n')
                    for para in paragraphs:
                        para = para.strip()
                        if len(para) > 50:  # 只保存有意义的段落
                            self.add_memory(
                                content=para,
                                source=f"memory/{md_file.name}",
                                memory_type="daily_log"
                            )
                            loaded_count += 1
                    
                    print(f"✅ 已加载 {md_file.name}")
                except Exception as e:
                    print(f"⚠️  加载 {md_file.name} 失败: {e}")
        
        # 2. 加载 CAPABILITIES.md
        capabilities_file = self.memory_dir / "CAPABILITIES.md"
        if capabilities_file.exists():
            try:
                with open(capabilities_file, 'r', encoding='utf-8') as f:
                    content = f.read()
                
                # 提取能力条目
                sections = re.split(r'###\s+', content)
                for section in sections[1:]:  # 跳过标题
                    if len(section) > 100:
                        self.add_memory(
                            content=section[:500],  # 限制长度
                            source="memory/CAPABILITIES.md",
                            memory_type="capability"
                        )
                        loaded_count += 1
                
                print(f"✅ 已加载 CAPABILITIES.md")
            except Exception as e:
                print(f"⚠️  加载 CAPABILITIES.md 失败: {e}")
        
        # 3. 加载 MEMORY.md
        memory_md = self.memory_dir / "MEMORY.md"
        if memory_md.exists():
            try:
                with open(memory_md, 'r', encoding='utf-8') as f:
                    content = f.read()
                
                # 提取重要约定
                sections = content.split('\n## ')
                for section in sections[1:]:
                    if len(section) > 50:
                        self.add_memory(
                            content=section[:500],
                            source="memory/MEMORY.md",
                            memory_type="core_memory"
                        )
                        loaded_count += 1
                
                print(f"✅ 已加载 MEMORY.md")
            except Exception as e:
                print(f"⚠️  加载 MEMORY.md 失败: {e}")
        
        print(f"\n📊 共加载 {loaded_count} 条记忆")
        return loaded_count
    
    def recall_for_prompt(self, user_input: str, max_memories: int = 3) -> str:
        """
        为用户输入召回相关记忆，用于增强 prompt
        
        Args:
            user_input: 用户输入内容
            max_memories: 最多召回几条记忆
            
        Returns:
            格式化的记忆上下文
        """
        memories = self.search_memory(user_input, top_k=max_memories, threshold=0.3)
        
        if not memories:
            return ""
        
        context = "\n[相关记忆]\n"
        for i, mem in enumerate(memories, 1):
            context += f"{i}. {mem['content'][:200]}...\n"
            context += f"   来源: {mem['source']}\n"
        
        return context
    
    def extract_from_chat(self, chat_content: str, source: str = "chat") -> List[str]:
        """
        从聊天记录中提取有价值的记忆
        
        Returns:
            提取的记忆ID列表
        """
        memory_ids = []
        
        # 提取问答对
        qa_patterns = [
            r'(?:Q|问|问题)[：:]\s*(.+?)\n+(?:A|答|回答)[：:]\s*(.+?)(?=\n(?:Q|问|问题)[：:]|\Z)',
            r'(?:用户|User)[：:]\s*(.+?)\n+(?:助手|Assistant|AI)[：:]\s*(.+?)(?=\n(?:用户|User)[：:]|\Z)',
        ]
        
        for pattern in qa_patterns:
            for match in re.finditer(pattern, chat_content, re.DOTALL | re.IGNORECASE):
                question = match.group(1).strip()
                answer = match.group(2).strip()
                
                if len(question) > 10 and len(answer) > 20:
                    content = f"Q: {question}\nA: {answer}"
                    mid = self.add_memory(
                        content=content,
                        source=source,
                        memory_type="qa"
                    )
                    memory_ids.append(mid)
        
        # 提取重要约定
        instruction_patterns = [
            r'(?:记住|请记住|记住这个)[：:]\s*(.+?)(?=\n|$)',
            r'(?:以后|下次|以后每次)[：:]\s*(.+?)(?=\n|$)',
            r'(?:重要|注意|提醒)[：:]\s*(.+?)(?=\n|$)',
        ]
        
        for pattern in instruction_patterns:
            for match in re.finditer(pattern, chat_content, re.IGNORECASE):
                content = match.group(1).strip()
                if len(content) > 10:
                    mid = self.add_memory(
                        content=content,
                        source=source,
                        memory_type="instruction"
                    )
                    memory_ids.append(mid)
        
        # 提取技术方案
        solution_pattern = r'(?:方案|解决方案|解决步骤|操作步骤)[：:]\s*\n?(.+?)(?=\n\n|\Z)'
        for match in re.finditer(solution_pattern, chat_content, re.DOTALL | re.IGNORECASE):
            content = match.group(1).strip()
            if len(content) > 50:
                mid = self.add_memory(
                    content=content,
                    source=source,
                    memory_type="solution"
                )
                memory_ids.append(mid)
        
        return memory_ids
    
    def get_stats(self) -> Dict:
        """获取记忆统计信息"""
        return {
            "total_memories": len(self.metadata),
            "sources": list(set(m["source"] for m in self.metadata)),
            "types": {}
        }
    
    def export_memories(self, output_file: str = None) -> str:
        """导出所有记忆为 Markdown"""
        if output_file is None:
            output_file = self.kb_dir / f"memories_export_{datetime.now().strftime('%Y%m%d')}.md"
        
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write("# OpenClaw 记忆导出\n\n")
            f.write(f"导出时间: {datetime.now().isoformat()}\n")
            f.write(f"记忆总数: {len(self.metadata)}\n\n")
            f.write("---\n\n")
            
            for mem in sorted(self.metadata, key=lambda x: x["created_at"], reverse=True):
                f.write(f"## {mem['id']}\n\n")
                f.write(f"**类型**: {mem['type']}\n\n")
                f.write(f"**来源**: {mem['source']}\n\n")
                f.write(f"**时间**: {mem['created_at']}\n\n")
                f.write(f"**内容**:\n\n{mem['content']}\n\n")
                f.write("---\n\n")
        
        return str(output_file)


# CLI 接口
if __name__ == "__main__":
    import argparse
    
    parser = argparse.ArgumentParser(description="OpenClaw Memory Enhancer")
    parser.add_argument("--load", action="store_true", help="加载 OpenClaw 记忆文件")
    parser.add_argument("--search", type=str, help="搜索记忆")
    parser.add_argument("--add", type=str, help="添加记忆")
    parser.add_argument("--extract", type=str, help="从文件提取聊天记录")
    parser.add_argument("--export", action="store_true", help="导出所有记忆")
    parser.add_argument("--stats", action="store_true", help="显示统计")
    
    args = parser.parse_args()
    
    enhancer = MemoryEnhancer()
    
    if args.load:
        print("🔄 加载 OpenClaw 记忆文件...")
        count = enhancer.load_openclaw_memory()
        print(f"✅ 完成！共加载 {count} 条记忆")
    
    elif args.search:
        print(f"🔍 搜索: {args.search}")
        results = enhancer.search_memory(args.search)
        for i, r in enumerate(results, 1):
            print(f"\n{i}. [{r['similarity']:.2f}] {r['content'][:100]}...")
            print(f"   来源: {r['source']}")
    
    elif args.add:
        mid = enhancer.add_memory(args.add)
        print(f"✅ 记忆已添加，ID: {mid}")
    
    elif args.extract:
        with open(args.extract, 'r') as f:
            content = f.read()
        ids = enhancer.extract_from_chat(content, source=args.extract)
        print(f"✅ 提取了 {len(ids)} 条记忆")
    
    elif args.export:
        file = enhancer.export_memories()
        print(f"✅ 已导出到: {file}")
    
    elif args.stats:
        stats = enhancer.get_stats()
        print(f"📊 记忆统计:")
        print(f"  总数: {stats['total_memories']}")
        print(f"  来源: {', '.join(stats['sources'])}")
    
    else:
        parser.print_help()