diff --git a/README.md b/README.md index 85a930e..ebffdb3 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,7 @@ Neo Agent 是一个基于 LangChain + LangGraph 的智能对话代理系统， #### 🏗️ 复合框架架构 - **LangChain**: 核心框架，提供LLM抽象和链式调用 - **LangGraph**: 状态图管理，实现复杂对话流程编排和动态多智能体协作 +- **Cognee**: 智能持久化动态AI记忆引擎（[查看详情](docs/COGNEE_INTEGRATION.md)） - **DeepAgents**: 子智能体生成、长期记忆和文件系统（[查看详情](docs/DEEPAGENTS_INTEGRATION.md)） - **多层模型架构**: 根据任务类型智能选择模型 - 主模型 (DeepSeek-V3.2): 处理主要对话、复杂推理和任务编排 @@ -23,7 +24,8 @@ Neo Agent 是一个基于 LangChain + LangGraph 的智能对话代理系统， ### 主要特性 -- 🧠 **分层记忆系统**: 短期记忆、长期记忆、知识库、基础知识 +- 🧠 **Cognee智能记忆**: 基于向量搜索和图数据库的持久动态AI记忆 +- 🌍 **世界观构建**: 基于Markdown的可编辑世界观系统，支持AI辅助生成 - 💭 **智能对话**: 角色扮演、连续对话、记忆检索、情感理解 - 📊 **情感分析**: 印象评估、累计评分、关系可视化 - 🖥️ **现代化GUI**: 基于Tkinter的友好界面 @@ -136,6 +138,36 @@ Neo Agent采用模块化的提示词工程系统，参考了SillyTavern的设计 详见 `prompts/README.md` 获取完整说明。 +### Cognee 智能记忆系统 + +Neo Agent 集成了 [Cognee](https://docs.cognee.ai/) 作为智能记忆引擎： + +- 🧠 **持久动态记忆**: 将原始数据转化为智能体的持久动态AI记忆 +- 🔗 **知识图谱**: 自动构建实体关系图，使知识相互关联 +- 🔍 **语义搜索**: 基于向量的语义检索，按含义搜索记忆 +- 📊 **模块化知识**: 支持自定义知识块和记忆类型 + +### 世界观构建系统 + +基于Markdown的智能体世界观构建系统： + +- 📝 **直接编辑**: 直接编辑Markdown源文件管理世界观 +- 🤖 **AI生成**: 使用自然语言描述，AI辅助生成详细世界观 +- 🔄 **知识同步**: 自动同步到Cognee记忆和知识库 +- 📁 **模块化**: 将世界观解析为模块化知识块 + +世界观文件位于 `prompts/worldview/` 目录。 + +### 鸣谢 + +感谢以下开源项目对 Neo Agent 的支持： + +- **[Cognee](https://github.com/topoteretes/cognee)** - 开源知识引擎，将原始数据转化为持久动态AI记忆 +- **[LangChain](https://github.com/langchain-ai/langchain)** - 构建LLM应用的强大框架 +- **[LangGraph](https://github.com/langchain-ai/langgraph)** - 状态图管理和多智能体协作 +- **[DeepAgents](https://github.com/deepagents/deepagents)** - 子智能体生成和持久化状态管理 +- **[SillyTavern](https://github.com/SillyTavern/SillyTavern)** - 提示词工程系统的设计参考 + ### 许可证 本项目采用 MIT 许可证。详见 [LICENSE](LICENSE) 文件。 @@ -153,6 +185,8 @@ Neo Agent is a LangChain + LangGraph-based intelligent conversation agent system #### 🏗️ Composite Framework Architecture - **LangChain**: Core framework providing LLM abstraction and chain invocation - **LangGraph**: State graph management for complex conversation flow orchestration +- **Cognee**: Intelligent persistent and dynamic AI memory engine ([documentation](docs/COGNEE_INTEGRATION.md)) +- **DeepAgents**: Sub-agent generation, long-term memory, and file system ([documentation](docs/DEEPAGENTS_INTEGRATION.md)) - **Multi-tier Model Architecture**: Intelligent model selection based on task type - Main Model (DeepSeek-V3.2): Handles primary conversations and complex reasoning - Tool Model (GLM-4.6V): Handles tool invocations, intent recognition, and lightweight tasks @@ -160,7 +194,8 @@ Neo Agent is a LangChain + LangGraph-based intelligent conversation agent system ### Key Features -- 🧠 **Hierarchical Memory System**: Short-term memory, long-term memory, knowledge base, base knowledge +- 🧠 **Cognee Smart Memory**: Persistent and dynamic AI memory based on vector search and graph databases +- 🌍 **Worldview Builder**: Editable Markdown-based worldview system with AI-assisted generation - 💭 **Intelligent Conversation**: Role-playing, continuous dialogue, memory retrieval, emotional understanding - 📊 **Emotion Analysis**: Impression assessment, cumulative scoring, relationship visualization - 🖥️ **Modern GUI**: User-friendly Tkinter-based interface @@ -209,6 +244,8 @@ Neo_Agent/ │ ├── gui/ # GUI modules │ ├── tools/ # Utility modules │ └── nps/ # NPS tool system +├── prompts/ # Prompt templates +│ ├── worldview/ # Worldview settings ├── tests/ # Test files ├── examples/ # Example code ├── main.py # Main entry point (recommended) @@ -220,6 +257,8 @@ Neo_Agent/ ### Core Modules +- **cognee_memory**: Cognee smart memory management +- **worldview_builder**: Markdown worldview builder - **chat_agent**: Conversation agent core - **database_manager**: Unified database management - **emotion_analyzer**: Emotional relationship analysis @@ -228,6 +267,16 @@ Neo_Agent/ - **long_term_memory**: Long-term memory system - **schedule_manager**: Schedule management +### Acknowledgments + +Thanks to the following open-source projects for supporting Neo Agent: + +- **[Cognee](https://github.com/topoteretes/cognee)** - Open-source knowledge engine for persistent dynamic AI memory +- **[LangChain](https://github.com/langchain-ai/langchain)** - Powerful framework for building LLM applications +- **[LangGraph](https://github.com/langchain-ai/langgraph)** - State graph management and multi-agent collaboration +- **[DeepAgents](https://github.com/deepagents/deepagents)** - Sub-agent generation and persistent state management +- **[SillyTavern](https://github.com/SillyTavern/SillyTavern)** - Design reference for prompt engineering system + ### License This project is licensed under the MIT License. See [LICENSE](LICENSE) file for details. diff --git a/cognee_manager.py b/cognee_manager.py new file mode 100644 index 0000000..a6083b9 --- /dev/null +++ b/cognee_manager.py @@ -0,0 +1,309 @@ +#!/usr/bin/env python3 +""" +Cognee 智能记忆管理器 - 独立应用 +提供 Cognee 记忆管理、世界观构建和日志查看的可视化界面 + +功能： +1. 🧠 Cognee 记忆管理 - 添加、搜索、查看智能记忆 +2. 🌍 世界观构建 - 创建、编辑 Markdown 世界观文件 +3. 📋 Cognee 日志 - 查看和分析 Cognee 系统日志 + +使用方法: + python cognee_manager.py +""" + +import os +import sys +import tkinter as tk +from tkinter import ttk, messagebox + +# 添加项目根目录到 Python 路径 +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from dotenv import load_dotenv +load_dotenv() + + +def setup_huggingface_tokenizer(): + """ + 配置 HuggingFace 分词器 + 在应用初始化时设置 HuggingFace 环境，替代 tiktoken + """ + # 1. 处理中国大陆 HuggingFace SSL 问题 + if os.getenv('HF_HUB_DISABLE_SSL_VERIFY', '0') == '1': + os.environ['CURL_CA_BUNDLE'] = '' + os.environ['REQUESTS_CA_BUNDLE'] = '' + print("✓ 已禁用 HuggingFace SSL 验证") + + # 2. 设置 HuggingFace 镜像站点（中国大陆加速） + hf_endpoint = os.getenv('HF_ENDPOINT', '') + if hf_endpoint: + os.environ['HF_ENDPOINT'] = hf_endpoint + print(f"✓ 使用 HuggingFace 镜像: {hf_endpoint}") + + # 3. 配置 tokenizer + embedding_model = os.getenv('COGNEE_EMBEDDING_MODEL', 'BAAI/bge-large-zh-v1.5') + huggingface_tokenizer = os.getenv('HUGGINGFACE_TOKENIZER', embedding_model) + os.environ['HUGGINGFACE_TOKENIZER'] = huggingface_tokenizer + + # 4. 尝试预加载 HuggingFace tokenizer + try: + from transformers import AutoTokenizer + print(f"✓ HuggingFace transformers 已加载") + print(f"✓ 使用 tokenizer: {huggingface_tokenizer}") + except ImportError: + print("⚠ transformers 未安装，请运行: pip install transformers") + + +class CogneeManagerApp: + """ + Cognee 智能记忆管理器应用 + 独立的桌面应用，提供完整的 Cognee 管理功能 + """ + + def __init__(self): + """初始化应用""" + # 配置 HuggingFace tokenizer + setup_huggingface_tokenizer() + + # 创建主窗口 + self.root = tk.Tk() + self.root.title("🧠 Cognee 智能记忆管理器") + self.root.geometry("1200x800") + + # 设置图标（如果存在） + try: + icon_path = os.path.join(os.path.dirname(__file__), 'assets', 'icon.ico') + if os.path.exists(icon_path): + self.root.iconbitmap(icon_path) + except: + pass + + # 初始化管理器 + self.cognee_manager = None + self.worldview_builder = None + + self._init_managers() + + # 创建界面 + self._create_ui() + + # 绑定关闭事件 + self.root.protocol("WM_DELETE_WINDOW", self._on_close) + + def _init_managers(self): + """初始化 Cognee 管理器和世界观构建器""" + try: + from src.core.cognee_memory import CogneeMemoryManager, get_cognee_manager + self.cognee_manager = get_cognee_manager() + print("✓ Cognee 记忆管理器已初始化") + except ImportError as e: + print(f"⚠ 无法加载 Cognee 记忆管理器: {e}") + except Exception as e: + print(f"⚠ Cognee 记忆管理器初始化失败: {e}") + + try: + from src.core.worldview_builder import WorldviewBuilder + self.worldview_builder = WorldviewBuilder() + print("✓ 世界观构建器已初始化") + except ImportError as e: + print(f"⚠ 无法加载世界观构建器: {e}") + except Exception as e: + print(f"⚠ 世界观构建器初始化失败: {e}") + + def _create_ui(self): + """创建用户界面""" + # 顶部菜单栏 + self._create_menu() + + # 顶部信息栏 + self._create_header() + + # 主内容区域 + main_frame = ttk.Frame(self.root) + main_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=5) + + # 加载 Cognee GUI + try: + from src.gui.cognee_gui import CogneeWorldviewManagerGUI + self.cognee_gui = CogneeWorldviewManagerGUI( + main_frame, + self.cognee_manager, + self.worldview_builder + ) + except ImportError as e: + self._show_import_error(main_frame, e) + except Exception as e: + self._show_error(main_frame, e) + + # 底部状态栏 + self._create_status_bar() + + def _create_menu(self): + """创建菜单栏""" + menubar = tk.Menu(self.root) + self.root.config(menu=menubar) + + # 文件菜单 + file_menu = tk.Menu(menubar, tearoff=0) + menubar.add_cascade(label="文件", menu=file_menu) + file_menu.add_command(label="刷新", command=self._refresh) + file_menu.add_separator() + file_menu.add_command(label="退出", command=self._on_close) + + # 帮助菜单 + help_menu = tk.Menu(menubar, tearoff=0) + menubar.add_cascade(label="帮助", menu=help_menu) + help_menu.add_command(label="Cognee 文档", command=self._open_cognee_docs) + help_menu.add_command(label="关于", command=self._show_about) + + def _create_header(self): + """创建顶部信息栏""" + header = ttk.Frame(self.root) + header.pack(fill=tk.X, padx=10, pady=5) + + # 标题 + title_label = ttk.Label( + header, + text="🧠 Cognee 智能记忆管理器", + font=("微软雅黑", 14, "bold") + ) + title_label.pack(side=tk.LEFT, padx=5) + + # 状态指示器 + self.status_indicator = ttk.Label( + header, + text="", + font=("微软雅黑", 10) + ) + self.status_indicator.pack(side=tk.RIGHT, padx=5) + self._update_status_indicator() + + def _create_status_bar(self): + """创建底部状态栏""" + status_frame = ttk.Frame(self.root) + status_frame.pack(fill=tk.X, side=tk.BOTTOM, padx=10, pady=5) + + ttk.Separator(self.root, orient=tk.HORIZONTAL).pack(fill=tk.X, side=tk.BOTTOM) + + self.status_label = ttk.Label( + status_frame, + text="就绪", + font=("微软雅黑", 9) + ) + self.status_label.pack(side=tk.LEFT, padx=5) + + # 版本信息 + version_label = ttk.Label( + status_frame, + text="Cognee Manager v1.0", + font=("微软雅黑", 9), + foreground="gray" + ) + version_label.pack(side=tk.RIGHT, padx=5) + + def _update_status_indicator(self): + """更新状态指示器""" + if self.cognee_manager and self.cognee_manager._initialized: + self.status_indicator.config( + text="🟢 Cognee 已连接", + foreground="green" + ) + else: + self.status_indicator.config( + text="🔴 Cognee 未连接", + foreground="red" + ) + + def _show_import_error(self, parent, error): + """显示导入错误""" + frame = ttk.Frame(parent) + frame.pack(fill=tk.BOTH, expand=True) + + ttk.Label( + frame, + text="⚠️ 组件加载失败", + font=("微软雅黑", 14, "bold") + ).pack(pady=20) + + ttk.Label( + frame, + text=f"无法加载 Cognee GUI 组件:\n{str(error)}", + font=("微软雅黑", 10), + wraplength=500 + ).pack(pady=10) + + ttk.Label( + frame, + text="请确保已安装所有依赖:\npip install cognee transformers", + font=("微软雅黑", 10) + ).pack(pady=10) + + def _show_error(self, parent, error): + """显示一般错误""" + frame = ttk.Frame(parent) + frame.pack(fill=tk.BOTH, expand=True) + + ttk.Label( + frame, + text="❌ 初始化失败", + font=("微软雅黑", 14, "bold") + ).pack(pady=20) + + ttk.Label( + frame, + text=str(error), + font=("微软雅黑", 10), + wraplength=500 + ).pack(pady=10) + + def _refresh(self): + """刷新应用""" + self._init_managers() + self._update_status_indicator() + self.status_label.config(text="已刷新") + + def _open_cognee_docs(self): + """打开 Cognee 文档""" + import webbrowser + webbrowser.open("https://docs.cognee.ai/") + + def _show_about(self): + """显示关于对话框""" + messagebox.showinfo( + "关于", + "Cognee 智能记忆管理器\n\n" + "版本: 1.0\n\n" + "基于 Cognee 开源知识引擎\n" + "https://github.com/topoteretes/cognee\n\n" + "功能:\n" + "• 🧠 智能记忆管理\n" + "• 🌍 世界观构建\n" + "• 📋 日志查看\n" + ) + + def _on_close(self): + """关闭应用""" + self.root.destroy() + + def run(self): + """运行应用""" + print("\n" + "=" * 50) + print("🧠 Cognee 智能记忆管理器已启动") + print("=" * 50 + "\n") + self.root.mainloop() + + +def main(): + """主函数""" + print("=" * 50) + print("Cognee 智能记忆管理器") + print("=" * 50) + + # 创建并运行应用 + app = CogneeManagerApp() + app.run() + + +if __name__ == '__main__': + main() diff --git a/docs/COGNEE_INTEGRATION.md b/docs/COGNEE_INTEGRATION.md new file mode 100644 index 0000000..b4b42f7 --- /dev/null +++ b/docs/COGNEE_INTEGRATION.md @@ -0,0 +1,273 @@ +# Cognee 智能记忆系统集成文档 + +## 概述 + +Neo Agent 集成了 [Cognee](https://docs.cognee.ai/) 作为智能记忆引擎，将原始数据（对话、文档、知识）转化为智能体的持久动态 AI 记忆。 + +## 什么是 Cognee？ + +Cognee 是一个开源的知识引擎，它结合了向量搜索和图数据库，使文档既可以按语义搜索，又能通过关系相互关联。 + +### 核心功能 + +- **持久动态记忆**: 将原始数据转化为可持久化的 AI 记忆 +- **知识图谱**: 自动构建实体关系图谱 +- **语义搜索**: 基于向量的含义检索 +- **模块化知识**: 支持自定义知识块 + +## 安装配置 + +### 1. 安装依赖 + +Cognee 已包含在 `requirements.txt` 中： + +```bash +pip install -r requirements.txt +``` + +或单独安装： + +```bash +pip install cognee>=0.5.0 +``` + +### 2. 配置环境变量 + +Neo Agent 会自动使用 SiliconFlow API 作为 Cognee 的 LLM 提供者（OpenAI 兼容端点）。 + +在 `.env` 文件中添加以下配置： + +```env +# 是否启用 Cognee 记忆系统（默认True） +COGNEE_ENABLED=true + +# SiliconFlow API 配置（Cognee 会自动使用这些配置） +SILICONFLOW_API_KEY=your-api-key +SILICONFLOW_API_URL=https://api.siliconflow.cn/v1/chat/completions + +# 可选：指定 Cognee 使用的模型 +# COGNEE_LLM_MODEL=Qwen/Qwen2.5-7B-Instruct +# COGNEE_EMBEDDING_MODEL=BAAI/bge-large-zh-v1.5 +``` + +### 3. 自定义 LLM 提供者 + +Cognee 配置为使用 SiliconFlow 作为自定义 LLM 提供者： +- **LLM_PROVIDER**: `custom`（使用 OpenAI 兼容端点） +- **LLM_ENDPOINT**: SiliconFlow API URL +- **LLM_MODEL**: `openai/{model_name}` 格式 + +这样配置后，Cognee 会使用 SiliconFlow 而不是默认的 OpenAI API。 + +## 使用方式 + +### 1. 自动集成 + +当启用 Cognee 后，Neo Agent 会自动： + +- 将对话消息同步到 Cognee 记忆系统 +- 每 5 轮对话自动构建知识图谱 +- 在知识提取时同步到 Cognee + +### 2. GUI 管理 + +在数据库管理界面中，新增了两个标签页： + +#### 🧠 Cognee 记忆 + +- **添加记忆**: 手动添加对话、知识、世界观记忆 +- **搜索记忆**: 基于语义的智能搜索 +- **统计信息**: 查看 Cognee 系统状态 + +#### 🌍 世界观构建 + +- **文件管理**: 管理 Markdown 世界观文件 +- **编辑器**: 内置 Markdown 编辑器 +- **AI 生成**: 自然语言生成世界观 +- **同步功能**: 同步到 Cognee 和知识库 + +### 3. API 使用 + +```python +from src.core.cognee_memory import CogneeMemoryManager + +# 创建管理器 +manager = CogneeMemoryManager() + +# 添加记忆（异步） +await manager.add_memory("用户: 你好\n助手: 你好！", "conversation") + +# 添加知识 +await manager.add_knowledge("小明", "用户的名字是小明", "fact") + +# 构建知识图谱 +await manager.cognify() + +# 搜索记忆 +results = await manager.search("小明") +``` + +### 4. 同步包装器（同步调用） + +```python +from src.core.cognee_memory import get_cognee_sync_wrapper + +wrapper = get_cognee_sync_wrapper() + +# 同步添加记忆 +wrapper.add_memory("内容", "conversation") + +# 同步搜索 +results = wrapper.search("查询") +``` + +## 记忆类型 + +Cognee 支持三种记忆类型： + +| 类型 | 说明 | 用途 | +|------|------|------| +| `conversation` | 对话记忆 | 存储对话历史 | +| `knowledge` | 知识记忆 | 存储实体知识和事实 | +| `worldview` | 世界观 | 存储虚拟世界设定 | + +## 世界观构建 + +### Markdown 世界观 + +世界观文件存储在 `prompts/worldview/` 目录，使用 Markdown 格式： + +```markdown +# 我的世界观 + +## 世界基本信息 + +**世界名称**：魔法世界 +**时代背景**：中世纪 + +## 世界特征 + +### 1. 魔法系统 + +这个世界存在魔法... +``` + +### AI 生成 + +使用自然语言描述生成世界观： + +```python +from src.core.worldview_builder import WorldviewBuilder + +builder = WorldviewBuilder() + +# 自然语言生成 +content = builder.create_worldview_from_natural_language( + "一个充满魔法的中世纪世界，有精灵、矮人和人类共存", + name="魔法世界", + use_llm=True +) + +# 保存 +builder.save_worldview("magic_world", content) +``` + +### 同步到 Cognee + +```python +# 同步世界观到 Cognee +await builder.sync_to_cognee("magic_world") + +# 同步到传统知识库 +builder.sync_to_knowledge_base("magic_world") +``` + +## 与现有系统集成 + +### LongTermMemoryManager + +`LongTermMemoryManager` 已集成 Cognee： + +```python +from src.core.long_term_memory import LongTermMemoryManager + +# 自动启用 Cognee +manager = LongTermMemoryManager(enable_cognee=True) + +# 添加消息时自动同步到 Cognee +manager.add_message("user", "你好") + +# 搜索 Cognee 记忆 +results = manager.search_cognee_memory("你好") +``` + +### 统计信息 + +```python +stats = manager.get_statistics() +print(stats['cognee']) # Cognee 统计信息 +``` + +## 架构图 + +``` +┌─────────────────────────────────────────────────────┐ +│ Neo Agent │ +├─────────────────────────────────────────────────────┤ +│ │ +│ ┌─────────────┐ ┌─────────────┐ ┌─────────┐ │ +│ │ 对话系统 │───▶│ 长期记忆管理 │───▶│ Cognee │ │ +│ └─────────────┘ └─────────────┘ └─────────┘ │ +│ │ │ │ │ +│ ▼ ▼ ▼ │ +│ ┌─────────────┐ ┌─────────────┐ ┌─────────┐ │ +│ │ 知识提取 │───▶│ 知识库 │◀──▶│ 知识图谱 │ │ +│ └─────────────┘ └─────────────┘ └─────────┘ │ +│ │ +│ ┌─────────────────────────────────────────────────┐ │ +│ │ 世界观构建系统 │ │ +│ │ ┌───────────┐ ┌───────────┐ ┌─────────┐ │ │ +│ │ │ Markdown │───▶│ 模块解析 │───▶│ Cognee │ │ │ +│ │ │ 文件 │ │ │ │ 同步 │ │ │ +│ │ └───────────┘ └───────────┘ └─────────┘ │ │ +│ └─────────────────────────────────────────────────┘ │ +│ │ +└─────────────────────────────────────────────────────┘ +``` + +## 注意事项 + +1. **API 密钥**: Cognee 需要 LLM API 密钥进行知识提取 +2. **异步操作**: Cognee 核心操作是异步的，提供了同步包装器 +3. **性能**: 知识图谱构建可能需要一定时间 +4. **存储**: Cognee 使用本地存储，数据保存在当前目录 + +## 故障排除 + +### Cognee 未初始化 + +``` +⚠ Cognee 初始化失败，将仅使用数据库存储 +``` + +检查： +1. cognee 是否正确安装 +2. LLM API 密钥是否配置 +3. 网络连接是否正常 + +### 导入错误 + +``` +⚠ Cognee 模块导入失败 +``` + +解决方案： +```bash +pip install cognee>=0.5.0 +``` + +## 参考链接 + +- [Cognee 官方文档](https://docs.cognee.ai/) +- [Cognee GitHub](https://github.com/topoteretes/cognee) +- [Neo Agent 主文档](../README.md) diff --git a/example.env b/example.env index a4a0077..96db3e8 100644 --- a/example.env +++ b/example.env @@ -41,6 +41,44 @@ MAX_TOKENS=2000 # 是否启用基于LangGraph的动态协作图（默认True） # USE_DYNAMIC_MULTI_AGENT=True +# Cognee 智能记忆系统配置 +# 是否启用 Cognee 记忆系统（默认True） +# 如果遇到 LLM 连接问题，可以暂时设置为 false +COGNEE_ENABLED=true +# Cognee 使用的 LLM 模型（默认使用 TOOL_MODEL_NAME 或 MODEL_NAME） +# 直接使用服务商平台上的模型名称，如 Qwen/Qwen2.5-7B-Instruct +# COGNEE_LLM_MODEL=Qwen/Qwen2.5-7B-Instruct +# Cognee 使用的 Embedding 模型（SiliconFlow 支持的嵌入模型） +# 直接使用服务商平台上的模型名称，如 BAAI/bge-large-zh-v1.5 +# COGNEE_EMBEDDING_MODEL=BAAI/bge-large-zh-v1.5 +# LLM 超时时间（秒，默认120，建议设置较长以避免超时） +# LLM_TIMEOUT=120 +# HuggingFace Tokenizer（用于 embedding 模型） +# 设置为 embedding 模型名称让 LiteLLM 使用 HuggingFace AutoTokenizer +# HUGGINGFACE_TOKENIZER=BAAI/bge-large-zh-v1.5 +# HuggingFace 镜像站点（中国大陆加速） +# HF_ENDPOINT=https://hf-mirror.com +# 禁用 HuggingFace SSL 验证（解决中国大陆 SSL 问题，设置为 1 启用） +# HF_HUB_DISABLE_SSL_VERIFY=1 +# Instructor 模式（控制 LLM 响应类型，默认 json_mode） +# 可选值: json_mode, function_call, json_schema_mode 等 +# 参考: https://python.useinstructor.com/modes-comparison/ +# LLM_INSTRUCTOR_MODE=json_mode +# +# 技术说明： +# 系统使用 LiteLLM 的 OpenAI 兼容模式配置 SiliconFlow: +# 1. 设置 LLM_PROVIDER="openai" 使用 OpenAI 格式 +# 2. 设置 OPENAI_API_BASE 指向 SiliconFlow API 端点 +# 3. 模型名称直接使用服务商平台格式（如 Qwen/Qwen2.5-7B-Instruct） +# 4. 参考文档: +# - LiteLLM OpenAI Compatible: https://docs.litellm.ai/docs/providers/openai_compatible +# - Cognee LLM Providers: https://docs.cognee.ai/setup-configuration/llm-providers +# 5. 如果遇到连接问题: +# - 检查 API Key 是否正确 +# - 检查网络连接 +# - 尝试增加 LLM_TIMEOUT 值 +# - 或暂时设置 COGNEE_ENABLED=false + # 默认设置 # 角色设定 CHARACTER_NAME=小可 diff --git a/requirements.txt b/requirements.txt index b1169df..8eb0731 100644 --- a/requirements.txt +++ b/requirements.txt @@ -9,6 +9,10 @@ langgraph>=0.2.0 # DeepAgents - 用于子智能体生成、长期记忆和文件系统 deepagents>=0.4.0 +# Cognee - 智能持久化动态AI记忆引擎 +# 将原始数据转化为智能体的持久动态记忆，结合向量搜索和图数据库 +cognee>=0.5.0 + # LangChain集成 langchain-openai>=0.2.0 diff --git a/src/core/cognee_memory.py b/src/core/cognee_memory.py new file mode 100644 index 0000000..171ae88 --- /dev/null +++ b/src/core/cognee_memory.py @@ -0,0 +1,708 @@ +""" +Cognee 智能记忆管理模块 +基于 Cognee 实现持久化动态 AI 记忆，结合向量搜索和图数据库 +将对话、知识转化为可搜索、可关联的智能记忆 + +Cognee: https://docs.cognee.ai/ +""" + +import os +import asyncio +import traceback +from datetime import datetime +from typing import List, Dict, Any, Optional +from dotenv import load_dotenv + +from src.tools.debug_logger import get_debug_logger + +load_dotenv() + +# 获取debug日志记录器 +debug_logger = get_debug_logger() + +# Cognee 配置 +COGNEE_ENABLED = os.getenv('COGNEE_ENABLED', 'true').lower() == 'true' +# 优先使用 SILICONFLOW_API_KEY，然后是 LLM_API_KEY +COGNEE_LLM_API_KEY = os.getenv('SILICONFLOW_API_KEY') or os.getenv('LLM_API_KEY') +# SiliconFlow API URL (OpenAI兼容格式) +SILICONFLOW_API_URL = os.getenv('SILICONFLOW_API_URL', 'https://api.siliconflow.cn/v1/chat/completions') +# 从URL中提取基础端点 +SILICONFLOW_BASE_URL = SILICONFLOW_API_URL.replace('/chat/completions', '') if SILICONFLOW_API_URL else 'https://api.siliconflow.cn/v1' +# Cognee 使用的模型（使用工具模型或主模型） +COGNEE_LLM_MODEL = os.getenv('COGNEE_LLM_MODEL') or os.getenv('TOOL_MODEL_NAME') or os.getenv('MODEL_NAME', 'Qwen/Qwen2.5-7B-Instruct') +# Cognee 使用的 Embedding 模型 +COGNEE_EMBEDDING_MODEL = os.getenv('COGNEE_EMBEDDING_MODEL', 'BAAI/bge-large-zh-v1.5') + + +def _format_exception(e: Exception) -> str: + """格式化异常信息，确保即使 str(e) 为空也能获取有用信息""" + error_type = type(e).__name__ + error_msg = str(e) + if not error_msg: + # 尝试获取更多信息 + if hasattr(e, 'message'): + error_msg = e.message + elif hasattr(e, 'args') and e.args: + error_msg = str(e.args[0]) if e.args[0] else repr(e.args) + else: + error_msg = repr(e) + return f"[{error_type}] {error_msg}" + + +class CogneeMemoryManager: + """ + Cognee 智能记忆管理器 + + 将原始数据（对话、文档、知识）转化为智能体的持久动态记忆： + 1. 使用向量搜索实现语义检索 + 2. 使用图数据库建立知识关联 + 3. 支持动态更新和演化 + + 功能： + - 对话记忆：存储和检索对话历史 + - 知识图谱：自动构建实体关系图 + - 语义搜索：基于含义的智能检索 + - 模块化知识：支持自定义知识块 + """ + + def __init__( + self, + api_key: str = None, + enabled: bool = COGNEE_ENABLED + ): + """ + 初始化 Cognee 记忆管理器 + + Args: + api_key: LLM API 密钥 + enabled: 是否启用 Cognee + """ + self.enabled = enabled + self.api_key = api_key or COGNEE_LLM_API_KEY + self._initialized = False + self._cognee = None + + if self.enabled: + self._setup_cognee() + else: + debug_logger.log_info('CogneeMemoryManager', 'Cognee 已禁用') + + def _setup_cognee(self): + """配置 Cognee 环境，使用 SiliconFlow 作为自定义 LLM 提供者""" + try: + # 验证 API 密钥 + if not self.api_key: + debug_logger.log_warning('CogneeMemoryManager', + 'API 密钥未设置，Cognee 功能可能受限。请设置 SILICONFLOW_API_KEY 或 LLM_API_KEY') + + # 配置 Cognee 使用 SiliconFlow（OpenAI兼容的自定义端点） + # 参考: https://docs.cognee.ai/setup-configuration/llm-providers + # + # SiliconFlow 是 OpenAI 兼容的 API 端点 + # 参考 LiteLLM 文档: https://docs.litellm.ai/docs/providers/openai_compatible + # + # 对于 OpenAI 兼容的端点，需要： + # 1. 使用 openai 作为 provider（因为 SiliconFlow 兼容 OpenAI API 格式） + # 2. 设置 OPENAI_API_BASE 指向 SiliconFlow 端点 + # 3. 设置 OPENAI_API_KEY 为 SiliconFlow API Key + # 4. 模型名称直接使用 SiliconFlow 格式（如 Qwen/Qwen2.5-7B-Instruct） + + # 在导入 cognee 之前设置环境变量（这是 Cognee 推荐的配置方式） + # 使用 OpenAI 兼容模式 + os.environ['LLM_PROVIDER'] = 'openai' + os.environ['LLM_API_KEY'] = self.api_key or '' + os.environ['OPENAI_API_KEY'] = self.api_key or '' # LiteLLM 也会检查这个 + os.environ['OPENAI_API_BASE'] = SILICONFLOW_BASE_URL # 关键：设置基础URL + os.environ['LLM_ENDPOINT'] = SILICONFLOW_BASE_URL + os.environ['LLM_MODEL'] = COGNEE_LLM_MODEL # 直接使用模型名，如 Qwen/Qwen2.5-7B-Instruct + + # Embedding 配置 - 同样使用 OpenAI 兼容模式 + os.environ['EMBEDDING_PROVIDER'] = 'openai' + os.environ['EMBEDDING_API_KEY'] = self.api_key or '' + os.environ['OPENAI_EMBEDDING_API_KEY'] = self.api_key or '' # 备用 + os.environ['EMBEDDING_ENDPOINT'] = SILICONFLOW_BASE_URL + os.environ['EMBEDDING_MODEL'] = COGNEE_EMBEDDING_MODEL # 直接使用模型名 + + # 配置 HuggingFace tokenizer + # 解决 "Could not automatically map xxx to a tokeniser" 错误 + # 使用 HuggingFace AutoTokenizer 替代 tiktoken，更好地支持中文 embedding 模型 + + # 1. 处理中国大陆 HuggingFace SSL 问题 + # 设置环境变量禁用 SSL 验证（可选，解决 SSL 证书问题） + if os.getenv('HF_HUB_DISABLE_SSL_VERIFY', '0') == '1': + os.environ['CURL_CA_BUNDLE'] = '' + os.environ['REQUESTS_CA_BUNDLE'] = '' + debug_logger.log_info('CogneeMemoryManager', '已禁用 HuggingFace SSL 验证') + + # 2. 设置 HuggingFace 镜像站点（中国大陆加速） + hf_endpoint = os.getenv('HF_ENDPOINT', '') + if hf_endpoint: + os.environ['HF_ENDPOINT'] = hf_endpoint + debug_logger.log_info('CogneeMemoryManager', f'使用 HuggingFace 镜像: {hf_endpoint}') + + # 3. 配置 tokenizer + # 优先使用 HuggingFace tokenizer，如果模型不支持则回退到 cl100k_base + embedding_model = COGNEE_EMBEDDING_MODEL + huggingface_tokenizer = os.getenv('HUGGINGFACE_TOKENIZER', embedding_model) + os.environ['HUGGINGFACE_TOKENIZER'] = huggingface_tokenizer + + # 4. 配置 HuggingFace AutoTokenizer（完全替代 tiktoken） + # HuggingFace 分词器更好地支持中文 embedding 模型 + try: + # 尝试预加载 HuggingFace tokenizer + from transformers import AutoTokenizer + + # 为常用中文 embedding 模型配置 tokenizer + tokenizer_model = huggingface_tokenizer + debug_logger.log_info('CogneeMemoryManager', + f'使用 HuggingFace AutoTokenizer: {tokenizer_model}') + + # 尝试加载 tokenizer 以验证配置 + try: + _ = AutoTokenizer.from_pretrained(tokenizer_model, trust_remote_code=True) + debug_logger.log_info('CogneeMemoryManager', + f'HuggingFace tokenizer 加载成功: {tokenizer_model}') + except Exception as load_err: + debug_logger.log_warning('CogneeMemoryManager', + f'预加载 tokenizer 失败（将在使用时重试）: {load_err}') + + except ImportError: + debug_logger.log_warning('CogneeMemoryManager', + 'transformers 未安装，请运行: pip install transformers') + except Exception as e: + debug_logger.log_warning('CogneeMemoryManager', + f'配置 HuggingFace tokenizer 失败: {e}') + + # 设置较长的超时时间以适应网络延迟 + os.environ['LLM_TIMEOUT'] = os.getenv('LLM_TIMEOUT', '120') + + # 设置 Instructor 模式（控制 LLM 的响应类型） + # 参考: https://python.useinstructor.com/modes-comparison/ + os.environ['LLM_INSTRUCTOR_MODE'] = os.getenv('LLM_INSTRUCTOR_MODE', 'json_mode') + + # 禁用多用户访问控制（简化配置） + os.environ['ENABLE_BACKEND_ACCESS_CONTROL'] = 'false' + + debug_logger.log_info('CogneeMemoryManager', 'Cognee 环境变量配置完成', { + 'llm_provider': 'openai', + 'openai_api_base': SILICONFLOW_BASE_URL, + 'llm_model': COGNEE_LLM_MODEL, + 'embedding_provider': 'openai', + 'embedding_model': COGNEE_EMBEDDING_MODEL + }) + + # 导入 cognee（环境变量必须在此之前设置） + import cognee + + self._cognee = cognee + self._initialized = True + + debug_logger.log_info('CogneeMemoryManager', 'Cognee 初始化成功（使用 SiliconFlow）') + print("✓ Cognee 智能记忆系统已初始化（使用 SiliconFlow API）") + + except ImportError as e: + debug_logger.log_error('CogneeMemoryManager', + f'Cognee 未安装，请运行: pip install cognee', e) + self.enabled = False + self._initialized = False + except Exception as e: + debug_logger.log_error('CogneeMemoryManager', + f'Cognee 初始化失败: {_format_exception(e)}', e) + self.enabled = False + self._initialized = False + + async def add_memory( + self, + content: str, + memory_type: str = "conversation", + metadata: Dict[str, Any] = None + ) -> bool: + """ + 添加记忆到 Cognee + + Args: + content: 要存储的内容 + memory_type: 记忆类型（conversation/knowledge/worldview） + metadata: 附加元数据 + + Returns: + 是否成功添加 + """ + if not self._initialized: + debug_logger.log_warning('CogneeMemoryManager', 'Cognee 未初始化，跳过添加记忆') + return False + + try: + # 格式化内容，添加类型标记便于检索 + formatted_content = f"[{memory_type}] {content}" + if metadata: + metadata_str = " | ".join([f"{k}: {v}" for k, v in metadata.items()]) + formatted_content += f" ({metadata_str})" + + # 添加到 Cognee + await self._cognee.add(formatted_content) + + debug_logger.log_info('CogneeMemoryManager', f'记忆已添加: {memory_type}', { + 'content_length': len(content) + }) + + return True + + except Exception as e: + debug_logger.log_error('CogneeMemoryManager', + f'添加记忆失败: {_format_exception(e)}', e) + return False + + async def add_conversation( + self, + messages: List[Dict[str, str]], + session_id: str = None + ) -> bool: + """ + 添加对话记录到记忆系统 + + Args: + messages: 对话消息列表 [{"role": "user/assistant", "content": "..."}] + session_id: 会话ID + + Returns: + 是否成功添加 + """ + if not self._initialized: + return False + + try: + # 格式化对话 + conversation_text = "\n".join([ + f"{'用户' if msg['role'] == 'user' else '助手'}: {msg['content']}" + for msg in messages + ]) + + metadata = { + "session_id": session_id or datetime.now().strftime("%Y%m%d%H%M%S"), + "timestamp": datetime.now().isoformat(), + "message_count": len(messages) + } + + return await self.add_memory(conversation_text, "conversation", metadata) + + except Exception as e: + debug_logger.log_error('CogneeMemoryManager', + f'添加对话失败: {_format_exception(e)}', e) + return False + + async def add_knowledge( + self, + entity_name: str, + knowledge_content: str, + knowledge_type: str = "fact", + source: str = "conversation" + ) -> bool: + """ + 添加知识到记忆系统 + + Args: + entity_name: 实体名称 + knowledge_content: 知识内容 + knowledge_type: 知识类型（fact/definition/relation） + source: 知识来源 + + Returns: + 是否成功添加 + """ + if not self._initialized: + return False + + try: + content = f"关于 {entity_name}: {knowledge_content}" + metadata = { + "entity": entity_name, + "type": knowledge_type, + "source": source, + "timestamp": datetime.now().isoformat() + } + + return await self.add_memory(content, "knowledge", metadata) + + except Exception as e: + debug_logger.log_error('CogneeMemoryManager', f'添加知识失败: {_format_exception(e)}', e) + return False + + async def add_worldview( + self, + worldview_content: str, + category: str = "general" + ) -> bool: + """ + 添加世界观内容到记忆系统 + + Args: + worldview_content: 世界观内容 + category: 分类（general/rules/locations/characters/events） + + Returns: + 是否成功添加 + """ + if not self._initialized: + return False + + try: + metadata = { + "category": category, + "timestamp": datetime.now().isoformat() + } + + return await self.add_memory(worldview_content, "worldview", metadata) + + except Exception as e: + debug_logger.log_error('CogneeMemoryManager', f'添加世界观失败: {_format_exception(e)}', e) + return False + + async def cognify(self) -> bool: + """ + 处理并构建知识图谱 + 将添加的内容转化为可查询的知识图谱 + + Returns: + 是否成功处理 + """ + if not self._initialized: + return False + + try: + await self._cognee.cognify() + + debug_logger.log_info('CogneeMemoryManager', '知识图谱构建完成') + print("✓ Cognee 知识图谱已更新") + + return True + + except Exception as e: + debug_logger.log_error('CogneeMemoryManager', f'构建知识图谱失败: {_format_exception(e)}', e) + return False + + async def memify(self) -> bool: + """ + 应用记忆算法 + 为图谱添加记忆算法，优化检索效果 + + Returns: + 是否成功应用 + """ + if not self._initialized: + return False + + try: + await self._cognee.memify() + + debug_logger.log_info('CogneeMemoryManager', '记忆算法已应用') + + return True + + except Exception as e: + debug_logger.log_error('CogneeMemoryManager', f'应用记忆算法失败: {_format_exception(e)}', e) + return False + + async def search( + self, + query: str, + memory_type: str = None, + max_results: int = 10 + ) -> List[Dict[str, Any]]: + """ + 搜索记忆 + + Args: + query: 搜索查询 + memory_type: 限制搜索的记忆类型（可选） + max_results: 最大结果数量 + + Returns: + 搜索结果列表 + """ + if not self._initialized: + debug_logger.log_warning('CogneeMemoryManager', 'Cognee 未初始化，返回空结果') + return [] + + try: + # 如果指定了记忆类型，添加到查询中 + search_query = query + if memory_type: + search_query = f"[{memory_type}] {query}" + + # 执行搜索 + results = await self._cognee.search(search_query) + + # 格式化结果 + formatted_results = [] + for i, result in enumerate(results[:max_results]): + formatted_results.append({ + "index": i + 1, + "content": str(result), + # 相关度从 1.0 递减，每步 0.1，并在 0.0 处做下限裁剪，避免出现负数 + "relevance": max(0.0, 1.0 - (i * 0.1)) + }) + + debug_logger.log_info('CogneeMemoryManager', f'搜索完成: {query}', { + 'results_count': len(formatted_results) + }) + + return formatted_results + + except Exception as e: + debug_logger.log_error('CogneeMemoryManager', f'搜索失败: {_format_exception(e)}', e) + return [] + + async def search_conversations(self, query: str, max_results: int = 5) -> List[Dict[str, Any]]: + """搜索对话记忆""" + return await self.search(query, memory_type="conversation", max_results=max_results) + + async def search_knowledge(self, query: str, max_results: int = 10) -> List[Dict[str, Any]]: + """搜索知识记忆""" + return await self.search(query, memory_type="knowledge", max_results=max_results) + + async def search_worldview(self, query: str, max_results: int = 5) -> List[Dict[str, Any]]: + """搜索世界观记忆""" + return await self.search(query, memory_type="worldview", max_results=max_results) + + async def get_related_knowledge( + self, + entity_name: str, + max_results: int = 10 + ) -> List[Dict[str, Any]]: + """ + 获取与实体相关的知识 + + Args: + entity_name: 实体名称 + max_results: 最大结果数量 + + Returns: + 相关知识列表 + """ + if not self._initialized: + return [] + + try: + query = f"关于 {entity_name} 的所有信息和关联" + return await self.search(query, max_results=max_results) + + except Exception as e: + debug_logger.log_error('CogneeMemoryManager', + f'获取相关知识失败: {_format_exception(e)}', e) + return [] + + async def process_and_store( + self, + content: str, + memory_type: str = "conversation", + metadata: Dict[str, Any] = None + ) -> bool: + """ + 一站式处理：添加内容并更新知识图谱 + + Args: + content: 要处理的内容 + memory_type: 记忆类型 + metadata: 附加元数据 + + Returns: + 是否成功处理 + """ + try: + # 添加内容 + if not await self.add_memory(content, memory_type, metadata): + return False + + # 构建知识图谱 + if not await self.cognify(): + return False + + # 应用记忆算法 + await self.memify() + + return True + + except Exception as e: + debug_logger.log_error('CogneeMemoryManager', + f'一站式处理失败: {_format_exception(e)}', e) + return False + + async def clear_all_memory(self) -> bool: + """ + 清空所有记忆 + + Returns: + 是否成功清空 + """ + if not self._initialized: + return False + + try: + # Cognee 删除所有数据 + await self._cognee.prune.prune_data() + await self._cognee.prune.prune_system(metadata=True) + + debug_logger.log_info('CogneeMemoryManager', '所有记忆已清空') + print("✓ Cognee 记忆已清空") + + return True + + except Exception as e: + debug_logger.log_error('CogneeMemoryManager', f'清空记忆失败: {_format_exception(e)}', e) + return False + + def get_statistics(self) -> Dict[str, Any]: + """ + 获取记忆系统统计信息 + + Returns: + 统计信息字典 + """ + return { + "enabled": self.enabled, + "initialized": self._initialized, + "api_key_configured": bool(self.api_key), + "backend": "cognee" + } + + +# 同步包装器，方便在非异步环境中使用 +class CogneeMemorySyncWrapper: + """ + Cognee 记忆管理器的同步包装器 + 在非异步环境中提供便捷的同步接口 + """ + + def __init__(self, manager: CogneeMemoryManager = None): + """ + 初始化同步包装器 + + Args: + manager: CogneeMemoryManager 实例 + """ + self.manager = manager or CogneeMemoryManager() + + def _run_async(self, coro): + """运行异步协程""" + try: + loop = asyncio.get_event_loop() + except RuntimeError: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + return loop.run_until_complete(coro) + + def add_memory(self, content: str, memory_type: str = "conversation", + metadata: Dict[str, Any] = None) -> bool: + """同步添加记忆""" + return self._run_async(self.manager.add_memory(content, memory_type, metadata)) + + def add_conversation(self, messages: List[Dict[str, str]], + session_id: str = None) -> bool: + """同步添加对话""" + return self._run_async(self.manager.add_conversation(messages, session_id)) + + def add_knowledge(self, entity_name: str, knowledge_content: str, + knowledge_type: str = "fact", source: str = "conversation") -> bool: + """同步添加知识""" + return self._run_async( + self.manager.add_knowledge(entity_name, knowledge_content, knowledge_type, source) + ) + + def add_worldview(self, worldview_content: str, category: str = "general") -> bool: + """同步添加世界观""" + return self._run_async(self.manager.add_worldview(worldview_content, category)) + + def cognify(self) -> bool: + """同步构建知识图谱""" + return self._run_async(self.manager.cognify()) + + def memify(self) -> bool: + """同步应用记忆算法""" + return self._run_async(self.manager.memify()) + + def search(self, query: str, memory_type: str = None, + max_results: int = 10) -> List[Dict[str, Any]]: + """同步搜索""" + return self._run_async(self.manager.search(query, memory_type, max_results)) + + def process_and_store(self, content: str, memory_type: str = "conversation", + metadata: Dict[str, Any] = None) -> bool: + """同步一站式处理""" + return self._run_async(self.manager.process_and_store(content, memory_type, metadata)) + + def clear_all_memory(self) -> bool: + """同步清空记忆""" + return self._run_async(self.manager.clear_all_memory()) + + def get_statistics(self) -> Dict[str, Any]: + """获取统计信息""" + return self.manager.get_statistics() + + +# 全局实例 +_global_cognee_manager: Optional[CogneeMemoryManager] = None + + +def get_cognee_manager() -> CogneeMemoryManager: + """ + 获取全局 Cognee 记忆管理器实例（单例模式） + + Returns: + CogneeMemoryManager 实例 + """ + global _global_cognee_manager + if _global_cognee_manager is None: + _global_cognee_manager = CogneeMemoryManager() + return _global_cognee_manager + + +def get_cognee_sync_wrapper() -> CogneeMemorySyncWrapper: + """ + 获取 Cognee 同步包装器 + + Returns: + CogneeMemorySyncWrapper 实例 + """ + return CogneeMemorySyncWrapper(get_cognee_manager()) + + +if __name__ == '__main__': + print("=" * 60) + print("Cognee 智能记忆管理器测试") + print("=" * 60) + + async def test(): + manager = CogneeMemoryManager() + + print("\n统计信息:") + stats = manager.get_statistics() + print(f" 启用: {stats['enabled']}") + print(f" 已初始化: {stats['initialized']}") + print(f" API密钥配置: {stats['api_key_configured']}") + + if manager._initialized: + # 测试添加记忆 + print("\n测试添加对话...") + await manager.add_conversation([ + {"role": "user", "content": "你好，我叫小明"}, + {"role": "assistant", "content": "你好小明！很高兴认识你"} + ]) + + # 测试添加知识 + print("测试添加知识...") + await manager.add_knowledge("小明", "用户的名字是小明", "fact", "conversation") + + # 构建知识图谱 + print("构建知识图谱...") + await manager.cognify() + + # 测试搜索 + print("测试搜索...") + results = await manager.search("小明") + print(f"找到 {len(results)} 条结果") + for r in results: + print(f" - {r['content'][:50]}...") + + asyncio.run(test()) + print("\n✓ 测试完成") diff --git a/src/core/knowledge_base.py b/src/core/knowledge_base.py index a4b9a5f..063f58f 100644 --- a/src/core/knowledge_base.py +++ b/src/core/knowledge_base.py @@ -6,6 +6,7 @@ import os import json +import re import uuid from datetime import datetime from typing import List, Dict, Any, Optional @@ -21,6 +22,99 @@ debug_logger = get_debug_logger() +def robust_json_parse(content: str, context: str = "JSON解析") -> Optional[Any]: + """ + 鲁棒性JSON解析，尝试多种方式解析LLM返回的JSON内容 + + Args: + content: 要解析的内容 + context: 上下文描述（用于日志） + + Returns: + 解析后的JSON对象，失败返回None + """ + if not content or not content.strip(): + debug_logger.log_info('JSONParser', f'{context}: 内容为空') + return None + + original_content = content + content = content.strip() + + # 尝试1：直接解析 + try: + return json.loads(content) + except json.JSONDecodeError: + pass + + # 尝试2：清理markdown代码块 + if '```' in content: + try: + # 提取代码块内容 + code_blocks = re.findall(r'```(?:json)?\s*([\s\S]*?)\s*```', content) + if code_blocks: + for block in code_blocks: + try: + return json.loads(block.strip()) + except json.JSONDecodeError: + continue + except Exception: + pass + + # 尝试3：提取JSON数组或对象 + try: + # 寻找JSON数组 [...] + array_match = re.search(r'\[[\s\S]*\]', content) + if array_match: + try: + return json.loads(array_match.group()) + except json.JSONDecodeError: + pass + + # 寻找JSON对象 {...} + object_match = re.search(r'\{[\s\S]*\}', content) + if object_match: + try: + return json.loads(object_match.group()) + except json.JSONDecodeError: + pass + except Exception: + pass + + # 尝试4：修复常见的JSON格式问题 + try: + # 替换单引号为双引号 + fixed_content = content.replace("'", '"') + return json.loads(fixed_content) + except json.JSONDecodeError: + pass + + # 尝试5：处理可能的尾随逗号 + try: + # 移除数组或对象末尾的逗号 + fixed_content = re.sub(r',\s*([}\]])', r'\1', content) + return json.loads(fixed_content) + except json.JSONDecodeError: + pass + + # 尝试6：处理无引号的键名 + try: + # 为无引号的键名添加引号 + fixed_content = re.sub(r'(\{|,)\s*([a-zA-Z_][a-zA-Z0-9_]*)\s*:', r'\1"\2":', content) + return json.loads(fixed_content) + except json.JSONDecodeError: + pass + + # 所有尝试都失败，记录原始输出到debug日志 + debug_logger.log_raw_output( + 'JSONParser', + context, + original_content, + f'{context}失败：无法解析为有效的JSON' + ) + + return None + + class KnowledgeBase: """ 知识库管理器 @@ -482,33 +576,26 @@ def extract_knowledge(self, messages: List[Dict[str, Any]]) -> Optional[List[Dic max_tokens=1500 ) - content = content.strip() - - # 尝试解析JSON - # 清理可能的markdown代码块标记 - if content.startswith('```'): - content = content.split('```')[1] - if content.startswith('json'): - content = content[4:] - content = content.strip() - - try: - knowledge_list = json.loads(content) - if isinstance(knowledge_list, list): - return knowledge_list - else: - print(f"✗ 返回的不是列表格式: {type(knowledge_list)}") - debug_logger.log_error('KnowledgeBase', f'返回的不是列表格式: {type(knowledge_list)}') - return None - except json.JSONDecodeError as e: - print(f"✗ JSON解析失败: {e}") - print(f"原始内容: {content[:200]}...") - debug_logger.log_error('KnowledgeBase', 'JSON解析失败', e) - debug_logger.log_info('KnowledgeBase', '原始内容', {'content': content[:500]}) + # 使用鲁棒性JSON解析 + knowledge_list = robust_json_parse(content, '知识提取') + + if knowledge_list is None: + return None + + if isinstance(knowledge_list, list): + return knowledge_list + else: + debug_logger.log_raw_output( + 'KnowledgeBase', + '知识提取', + content, + f'返回的不是列表格式: {type(knowledge_list).__name__}' + ) return None except Exception as e: print(f"✗ 提取知识时出错: {e}") + debug_logger.log_error('KnowledgeBase', '提取知识时出错', e) return None def extract_entities_from_query(self, query: str) -> List[str]: @@ -547,32 +634,27 @@ def extract_entities_from_query(self, query: str) -> List[str]: max_tokens=500 ) - content = content.strip() - - # 清理markdown代码块 - if content.startswith('```'): - content = content.split('```')[1] - if content.startswith('json'): - content = content[4:] - content = content.strip() - - try: - entities = json.loads(content) - if isinstance(entities, list): - result = [e for e in entities if isinstance(e, str)] - debug_logger.log_info('KnowledgeBase', '实体提取成功', { - 'query': query, - 'entities': result - }) - return result - else: - debug_logger.log_info('KnowledgeBase', '实体提取结果不是列表', {'content': content}) - return [] - except json.JSONDecodeError as e: - print(f"✗ 实体提取JSON解析失败") - debug_logger.log_error('KnowledgeBase', 'JSON解析失败', e) - debug_logger.log_info('KnowledgeBase', '无法解析的内容', {'content': content[:500]}) - print(f"调试信息 - 原始内容: {content[:200]}...") + # 使用鲁棒性JSON解析 + entities = robust_json_parse(content, '实体提取') + + if entities is None: + debug_logger.log_info('KnowledgeBase', '未识别到实体', {'query': query}) + return [] + + if isinstance(entities, list): + result = [e for e in entities if isinstance(e, str)] + debug_logger.log_info('KnowledgeBase', '实体提取成功', { + 'query': query, + 'entities': result + }) + return result + else: + debug_logger.log_raw_output( + 'KnowledgeBase', + '实体提取', + content, + f'实体提取结果不是列表，而是 {type(entities).__name__}' + ) return [] except Exception as e: diff --git a/src/core/long_term_memory.py b/src/core/long_term_memory.py index 44c0448..f83d7ac 100644 --- a/src/core/long_term_memory.py +++ b/src/core/long_term_memory.py @@ -2,11 +2,13 @@ 长效记忆管理模块 实现分层记忆系统：短期记忆（最近20轮）+ 长期概括记忆 + 知识库 使用数据库替代JSON文件存储 +集成 Cognee 智能记忆系统实现持久动态 AI 记忆 """ import os import json import uuid +import asyncio from datetime import datetime from typing import List, Dict, Any, Optional from dotenv import load_dotenv @@ -16,19 +18,24 @@ load_dotenv() +# Cognee 集成配置 +COGNEE_ENABLED = os.getenv('COGNEE_ENABLED', 'true').lower() == 'true' + class LongTermMemoryManager: """ 长效记忆管理器 负责管理短期详细记忆和长期概括记忆的分层存储 使用数据库替代JSON文件 + 集成 Cognee 实现持久动态 AI 记忆 """ def __init__(self, db_manager: DatabaseManager = None, api_key: str = None, api_url: str = None, - model_name: str = None): + model_name: str = None, + enable_cognee: bool = COGNEE_ENABLED): """ 初始化长效记忆管理器 @@ -37,6 +44,7 @@ def __init__(self, api_key: API密钥 api_url: API地址 model_name: 模型名称 + enable_cognee: 是否启用 Cognee 智能记忆 """ # 使用共享的数据库管理器 self.db = db_manager or DatabaseManager() @@ -61,10 +69,32 @@ def __init__(self, model_name=self.model_name ) + # 初始化 Cognee 智能记忆（如果启用） + self.cognee_manager = None + self.enable_cognee = enable_cognee + if enable_cognee: + self._init_cognee() + # 检查是否需要从JSON迁移数据 self._check_and_migrate_json() - print(f"✓ 长效记忆管理器已初始化（使用数据库存储）") + cognee_status = "Cognee已启用" if self.cognee_manager else "仅数据库存储" + print(f"✓ 长效记忆管理器已初始化（{cognee_status}）") + + def _init_cognee(self): + """初始化 Cognee 智能记忆系统""" + try: + from src.core.cognee_memory import CogneeMemoryManager + self.cognee_manager = CogneeMemoryManager(api_key=self.api_key) + if not self.cognee_manager._initialized: + print("⚠ Cognee 初始化失败，将仅使用数据库存储") + self.cognee_manager = None + except ImportError as e: + print(f"⚠ Cognee 模块导入失败: {e}") + self.cognee_manager = None + except Exception as e: + print(f"⚠ Cognee 初始化异常: {e}") + self.cognee_manager = None def _check_and_migrate_json(self): """检查并迁移旧的JSON文件""" @@ -87,6 +117,7 @@ def _check_and_migrate_json(self): def add_message(self, role: str, content: str): """ 添加新消息到短期记忆（使用数据库） + 同时同步到 Cognee 智能记忆系统（如果启用） Args: role: 角色类型 ('user' 或 'assistant') @@ -95,6 +126,10 @@ def add_message(self, role: str, content: str): # 添加到数据库 self.db.add_short_term_message(role, content) + # 同步到 Cognee（如果启用） + if self.cognee_manager: + self._sync_message_to_cognee(role, content) + # 更新元数据 if role == 'user': total_conversations = self.db.get_metadata('total_conversations', 0) @@ -105,10 +140,54 @@ def add_message(self, role: str, content: str): if total_conversations % self.knowledge_extraction_interval == 0: print(f"\n📚 已达到 {total_conversations} 轮对话，开始提取知识...") self._extract_and_save_knowledge() + + # 每5轮构建一次 Cognee 知识图谱 + if self.cognee_manager: + self._run_cognee_cognify() # 检查是否需要归档 self._check_and_archive() + def _sync_message_to_cognee(self, role: str, content: str): + """ + 同步消息到 Cognee + + Args: + role: 角色类型 + content: 消息内容 + """ + loop: Optional[asyncio.AbstractEventLoop] = None + try: + loop = asyncio.new_event_loop() + + # 格式化为对话记忆 + role_name = "用户" if role == "user" else "助手" + formatted_content = f"{role_name}: {content}" + + loop.run_until_complete( + self.cognee_manager.add_memory(formatted_content, "conversation") + ) + except Exception as e: + print(f"⚠ 同步到 Cognee 失败: {e}") + finally: + # 确保事件循环被正确关闭，避免资源泄漏 + if loop is not None and not loop.is_closed(): + loop.close() + + def _run_cognee_cognify(self): + """运行 Cognee 知识图谱构建""" + loop: Optional[asyncio.AbstractEventLoop] = None + try: + loop = asyncio.new_event_loop() + loop.run_until_complete(self.cognee_manager.cognify()) + print("✓ Cognee 知识图谱已更新") + except Exception as e: + print(f"⚠ Cognee 知识图谱构建失败: {e}") + finally: + # 确保事件循环被正确关闭，避免资源泄漏 + if loop is not None and not loop.is_closed(): + loop.close() + def _check_and_archive(self): """ 检查短期记忆是否超过限制，如果超过则归档旧记忆 @@ -371,6 +450,11 @@ def get_statistics(self) -> Dict[str, Any]: # 获取知识库详细统计 kb_stats = self.knowledge_base.get_statistics() + # 获取 Cognee 统计信息 + cognee_stats = {} + if self.cognee_manager: + cognee_stats = self.cognee_manager.get_statistics() + return { 'short_term': { 'total_messages': len(short_term_messages), @@ -390,19 +474,62 @@ def get_statistics(self) -> Dict[str, Any]: 'total_definitions': kb_stats['total_definitions'], 'total_related_info': kb_stats['total_related_info'] }, + 'cognee': cognee_stats, # Cognee 智能记忆统计 'total_conversations': self.db.get_metadata('total_conversations', 0), 'database_size_kb': db_stats.get('db_size_kb', 0) } def clear_all_memory(self): """ - 清空所有记忆（短期、长期） + 清空所有记忆（短期、长期、Cognee） """ self.db.clear_short_term_memory() self.db.clear_long_term_memory() self.db.set_metadata('total_conversations', 0) + + # 清空 Cognee 记忆 + if self.cognee_manager: + loop: Optional[asyncio.AbstractEventLoop] = None + try: + loop = asyncio.new_event_loop() + loop.run_until_complete(self.cognee_manager.clear_all_memory()) + except Exception as e: + print(f"⚠ 清空 Cognee 记忆失败: {e}") + finally: + # 确保事件循环被正确关闭，避免资源泄漏 + if loop is not None and not loop.is_closed(): + loop.close() + print("✓ 所有记忆已清空") + def search_cognee_memory(self, query: str, memory_type: str = None) -> List[Dict[str, Any]]: + """ + 搜索 Cognee 智能记忆 + + Args: + query: 搜索查询 + memory_type: 记忆类型过滤 + + Returns: + 搜索结果列表 + """ + if not self.cognee_manager: + return [] + + loop: Optional[asyncio.AbstractEventLoop] = None + try: + loop = asyncio.new_event_loop() + return loop.run_until_complete( + self.cognee_manager.search(query, memory_type=memory_type) + ) + except Exception as e: + print(f"⚠ 搜索 Cognee 记忆失败: {e}") + return [] + finally: + # 确保事件循环被正确关闭，避免资源泄漏 + if loop is not None and not loop.is_closed(): + loop.close() + def get_context_for_chat(self, recent_count: int = 10) -> str: """ 获取用于聊天的上下文（包含长期记忆概括和短期记忆） diff --git a/src/core/worldview_builder.py b/src/core/worldview_builder.py new file mode 100644 index 0000000..f05b2e8 --- /dev/null +++ b/src/core/worldview_builder.py @@ -0,0 +1,672 @@ +""" +世界观构建系统 +基于 Markdown 的智能体世界观构建模块，支持： +1. 直接编辑源文本的世界观管理 +2. 自然语言世界观构建 +3. 通过 Cognee 转化为模块化知识块 +4. 与现有框架的知识库集成 + +用户可以通过自然语言描述来构建自己的虚拟世界观 +""" + +import os +import re +from datetime import datetime +from pathlib import Path +from typing import List, Dict, Any, Optional +from dotenv import load_dotenv + +from src.tools.debug_logger import get_debug_logger +from src.core.prompt_manager import get_prompt_manager +from src.core.database_manager import DatabaseManager + +load_dotenv() + +# 获取debug日志记录器 +debug_logger = get_debug_logger() + + +class WorldviewModule: + """ + 世界观模块数据类 + 表示一个世界观的独立模块/知识块 + """ + + def __init__( + self, + name: str, + category: str, + content: str, + source_file: str = None, + metadata: Dict[str, Any] = None + ): + """ + 初始化世界观模块 + + Args: + name: 模块名称 + category: 分类（general/rules/locations/characters/events/items） + content: 模块内容 + source_file: 来源文件路径 + metadata: 附加元数据 + """ + self.name = name + self.category = category + self.content = content + self.source_file = source_file + self.metadata = metadata or {} + self.created_at = datetime.now().isoformat() + self.updated_at = self.created_at + + def to_dict(self) -> Dict[str, Any]: + """转换为字典""" + return { + "name": self.name, + "category": self.category, + "content": self.content, + "source_file": self.source_file, + "metadata": self.metadata, + "created_at": self.created_at, + "updated_at": self.updated_at + } + + @staticmethod + def from_dict(data: Dict[str, Any]) -> 'WorldviewModule': + """从字典创建""" + module = WorldviewModule( + name=data["name"], + category=data["category"], + content=data["content"], + source_file=data.get("source_file"), + metadata=data.get("metadata", {}) + ) + module.created_at = data.get("created_at", module.created_at) + module.updated_at = data.get("updated_at", module.updated_at) + return module + + +class WorldviewBuilder: + """ + 世界观构建器 + + 提供基于 Markdown 的世界观管理和构建功能： + 1. 加载和解析 Markdown 世界观文件 + 2. 自然语言转化为结构化世界观 + 3. 模块化知识块管理 + 4. 与 Cognee 和现有知识库集成 + """ + + # 世界观分类 + CATEGORIES = { + "general": "基本信息", + "rules": "规则设定", + "locations": "地点场所", + "characters": "角色人物", + "events": "事件历史", + "items": "物品道具", + "culture": "文化习俗", + "technology": "科技水平" + } + + def __init__( + self, + worldview_dir: str = None, + db_manager: DatabaseManager = None, + cognee_manager = None + ): + """ + 初始化世界观构建器 + + Args: + worldview_dir: 世界观文件目录 + db_manager: 数据库管理器 + cognee_manager: Cognee 记忆管理器 + """ + # 设置世界观目录 + if worldview_dir is None: + current_file = Path(__file__).resolve() + project_root = current_file.parent.parent.parent + worldview_dir = project_root / "prompts" / "worldview" + + self.worldview_dir = Path(worldview_dir) + self.db = db_manager or DatabaseManager() + self.cognee_manager = cognee_manager + self.prompt_manager = get_prompt_manager() + + # 模块缓存 + self._modules_cache: Dict[str, WorldviewModule] = {} + + # 确保目录存在 + self.worldview_dir.mkdir(parents=True, exist_ok=True) + + debug_logger.log_module('WorldviewBuilder', '世界观构建器初始化', { + 'worldview_dir': str(self.worldview_dir) + }) + + print(f"✓ 世界观构建系统已初始化 (目录: {self.worldview_dir})") + + def list_worldview_files(self) -> List[Dict[str, Any]]: + """ + 列出所有世界观文件 + + Returns: + 世界观文件信息列表 + """ + files = [] + for md_file in self.worldview_dir.glob("*.md"): + stat = md_file.stat() + files.append({ + "name": md_file.stem, + "path": str(md_file), + "size": stat.st_size, + "modified": datetime.fromtimestamp(stat.st_mtime).isoformat() + }) + return sorted(files, key=lambda x: x["name"]) + + def load_worldview(self, name: str) -> str: + """ + 加载世界观文件内容 + + Args: + name: 世界观名称（不含 .md 后缀） + + Returns: + 世界观内容 + """ + file_path = self.worldview_dir / f"{name}.md" + + if not file_path.exists(): + raise FileNotFoundError(f"世界观文件不存在: {file_path}") + + with open(file_path, 'r', encoding='utf-8') as f: + return f.read() + + def save_worldview(self, name: str, content: str) -> bool: + """ + 保存世界观到文件 + + Args: + name: 世界观名称 + content: 世界观内容 + + Returns: + 是否成功保存 + """ + try: + file_path = self.worldview_dir / f"{name}.md" + + with open(file_path, 'w', encoding='utf-8') as f: + f.write(content) + + debug_logger.log_info('WorldviewBuilder', f'世界观已保存: {name}') + print(f"✓ 世界观已保存: {file_path}") + + return True + + except Exception as e: + debug_logger.log_error('WorldviewBuilder', f'保存世界观失败: {str(e)}', e) + return False + + def delete_worldview(self, name: str) -> bool: + """ + 删除世界观文件 + + Args: + name: 世界观名称 + + Returns: + 是否成功删除 + """ + try: + file_path = self.worldview_dir / f"{name}.md" + + if file_path.exists(): + file_path.unlink() + debug_logger.log_info('WorldviewBuilder', f'世界观已删除: {name}') + return True + else: + debug_logger.log_warning('WorldviewBuilder', f'世界观文件不存在: {name}') + return False + + except Exception as e: + debug_logger.log_error('WorldviewBuilder', f'删除世界观失败: {str(e)}', e) + return False + + def parse_worldview_to_modules(self, content: str, source_file: str = None) -> List[WorldviewModule]: + """ + 解析世界观内容为模块化知识块 + + Args: + content: 世界观 Markdown 内容 + source_file: 来源文件名 + + Returns: + 世界观模块列表 + """ + modules = [] + + # 解析二级和三级标题作为模块分隔 + # 匹配 ## 标题 或 ### 标题 + sections = re.split(r'\n(?=#{2,3}\s+)', content) + + current_category = "general" + + for section in sections: + if not section.strip(): + continue + + # 提取标题（支持 ## 和 ###） + title_match = re.match(r'^#{2,3}\s+(.+?)(?:\n|$)', section) + if title_match: + title = title_match.group(1).strip() + section_content = section[title_match.end():].strip() + else: + title = "基本信息" + section_content = section.strip() + + # 根据标题推断分类 + category = self._infer_category(title) + + # 创建模块 + if section_content: + module = WorldviewModule( + name=title, + category=category, + content=section_content, + source_file=source_file, + metadata={ + "parsed_at": datetime.now().isoformat() + } + ) + modules.append(module) + + debug_logger.log_info('WorldviewBuilder', f'解析完成', { + 'source': source_file, + 'modules_count': len(modules) + }) + + return modules + + def _infer_category(self, title: str) -> str: + """ + 根据标题推断分类 + + Args: + title: 标题文本 + + Returns: + 分类名称 + """ + title_lower = title.lower() + + # 关键字全部小写，确保匹配一致性 + category_keywords = { + "general": ["基本", "简介", "概述", "世界", "背景"], + "rules": ["规则", "法则", "限制", "物理", "魔法"], + "locations": ["地点", "场所", "位置", "环境", "地理"], + "characters": ["角色", "人物", "npc", "种族"], + "events": ["事件", "历史", "时间", "故事"], + "items": ["物品", "道具", "装备", "物体"], + "culture": ["文化", "习俗", "社会", "传统"], + "technology": ["科技", "技术", "工具", "发明"] + } + + for category, keywords in category_keywords.items(): + for keyword in keywords: + if keyword in title_lower: + return category + + return "general" + + def create_worldview_from_natural_language( + self, + description: str, + name: str = None, + use_llm: bool = True + ) -> str: + """ + 从自然语言描述创建世界观 + + Args: + description: 用户的自然语言描述 + name: 世界观名称（可选） + use_llm: 是否使用 LLM 增强生成（默认 True） + + Returns: + 生成的 Markdown 世界观内容 + """ + if use_llm: + return self._generate_worldview_with_llm(description, name) + else: + return self._generate_worldview_template(description, name) + + def _generate_worldview_template(self, description: str, name: str = None) -> str: + """ + 生成基础世界观模板 + + Args: + description: 描述 + name: 名称 + + Returns: + Markdown 内容 + """ + worldview_name = name or "自定义世界观" + + template = f"""# {worldview_name} + +## 世界基本信息 + +**世界名称**：{worldview_name} + +**描述**：{description} + +**创建时间**：{datetime.now().strftime("%Y年%m月%d日")} + +## 世界特征 + +### 1. 基本设定 + +{description} + +### 2. 规则与限制 + + + +### 3. 重要地点 + + + +### 4. 重要人物 + + + +### 5. 历史事件 + + + +## 世界观融入 + +### 对话中的体现 + +- 自然地融入对话中 +- 保持一致性 +- 根据上下文适当展示 + +--- + +*这个世界观由用户自然语言描述生成，可以直接编辑此文件进行修改。* +""" + return template + + def _generate_worldview_with_llm(self, description: str, name: str = None) -> str: + """ + 使用 LLM 增强生成世界观 + + Args: + description: 用户描述 + name: 世界观名称 + + Returns: + 生成的 Markdown 内容 + """ + import requests + + api_key = os.getenv('SILICONFLOW_API_KEY') + api_url = os.getenv('SILICONFLOW_API_URL', 'https://api.siliconflow.cn/v1/chat/completions') + model_name = os.getenv('TOOL_MODEL_NAME', os.getenv('MODEL_NAME', 'Qwen/Qwen2.5-7B-Instruct')) + + if not api_key: + debug_logger.log_warning('WorldviewBuilder', 'API密钥未配置，使用模板生成') + return self._generate_worldview_template(description, name) + + prompt = f"""你是一个专业的世界观设计师。请根据以下用户描述，生成一个详细的 Markdown 格式世界观设定文档。 + +用户描述： +{description} + +请生成一个完整的世界观文档，包含以下部分： +1. 世界基本信息（名称、时代背景、地理位置） +2. 世界特征（社会环境、文化氛围、日常生活） +3. 物理规则（基本物理、时间流逝） +4. 地点设定（常见场所、虚拟环境） +5. 社会关系（人际关系、交流方式） +6. 科技水平（可用科技、科技限制） +7. 特殊设定（独特元素、魔法系统等，如有） +8. 注意事项（保持真实感、一致性等） + +请使用中文输出，格式为 Markdown。保持专业且有创意。""" + + try: + headers = { + 'Authorization': f'Bearer {api_key}', + 'Content-Type': 'application/json' + } + + payload = { + 'model': model_name, + 'messages': [ + {'role': 'system', 'content': '你是一个专业的世界观设计师，擅长创建丰富、一致的虚拟世界设定。'}, + {'role': 'user', 'content': prompt} + ], + 'temperature': 0.7, + 'max_tokens': 3000, + 'stream': False + } + + response = requests.post( + api_url, + headers=headers, + json=payload, + timeout=60 + ) + + response.raise_for_status() + result = response.json() + + if 'choices' in result and len(result['choices']) > 0: + content = result['choices'][0]['message']['content'].strip() + + # 确保内容以标题开头 + if not content.startswith('#'): + worldview_name = name or "自定义世界观" + content = f"# {worldview_name}\n\n{content}" + + debug_logger.log_info('WorldviewBuilder', 'LLM 世界观生成成功') + return content + + except Exception as e: + debug_logger.log_error('WorldviewBuilder', f'LLM 生成失败: {str(e)}', e) + + # 失败时降级到模板 + return self._generate_worldview_template(description, name) + + async def sync_to_cognee( + self, + worldview_name: str = None, + modules: List[WorldviewModule] = None + ) -> bool: + """ + 将世界观同步到 Cognee 记忆系统 + + Args: + worldview_name: 世界观名称（加载并解析） + modules: 或直接提供模块列表 + + Returns: + 是否成功同步 + """ + if self.cognee_manager is None: + debug_logger.log_warning('WorldviewBuilder', 'Cognee 管理器未配置，跳过同步') + return False + + try: + # 如果提供了世界观名称，加载并解析 + if worldview_name: + content = self.load_worldview(worldview_name) + modules = self.parse_worldview_to_modules(content, worldview_name) + + if not modules: + debug_logger.log_warning('WorldviewBuilder', '没有模块需要同步') + return False + + # 同步每个模块到 Cognee + for module in modules: + await self.cognee_manager.add_worldview( + worldview_content=f"[{module.category}] {module.name}: {module.content}", + category=module.category + ) + + # 构建知识图谱 + await self.cognee_manager.cognify() + + debug_logger.log_info('WorldviewBuilder', f'世界观已同步到 Cognee', { + 'modules_count': len(modules) + }) + + print(f"✓ {len(modules)} 个世界观模块已同步到 Cognee") + return True + + except Exception as e: + debug_logger.log_error('WorldviewBuilder', f'同步到 Cognee 失败: {str(e)}', e) + return False + + def sync_to_knowledge_base( + self, + worldview_name: str = None, + modules: List[WorldviewModule] = None + ) -> int: + """ + 将世界观同步到现有知识库 + + Args: + worldview_name: 世界观名称 + modules: 或直接提供模块列表 + + Returns: + 同步的模块数量 + """ + try: + # 如果提供了世界观名称，加载并解析 + if worldview_name: + content = self.load_worldview(worldview_name) + modules = self.parse_worldview_to_modules(content, worldview_name) + + if not modules: + return 0 + + synced_count = 0 + + for module in modules: + # 查找或创建实体 + entity_uuid = self.db.find_or_create_entity(f"世界观_{module.name}") + + # 设置定义 + self.db.set_entity_definition( + entity_uuid=entity_uuid, + content=module.content, + type_=f"世界观-{self.CATEGORIES.get(module.category, module.category)}", + source=f"worldview:{module.source_file or 'manual'}", + confidence=1.0, + priority=80, + is_base_knowledge=False + ) + + synced_count += 1 + + debug_logger.log_info('WorldviewBuilder', f'世界观已同步到知识库', { + 'synced_count': synced_count + }) + + print(f"✓ {synced_count} 个世界观模块已同步到知识库") + return synced_count + + except Exception as e: + debug_logger.log_error('WorldviewBuilder', f'同步到知识库失败: {str(e)}', e) + return 0 + + def get_worldview_for_prompt(self, name: str = None) -> str: + """ + 获取用于提示词的世界观内容 + + Args: + name: 世界观名称（如果为 None，使用默认世界观） + + Returns: + 格式化的世界观提示词 + """ + try: + return self.prompt_manager.get_worldview_prompt(worldview_name=name) + except FileNotFoundError: + debug_logger.log_warning('WorldviewBuilder', f'世界观文件未找到: {name}') + return "" + + def get_statistics(self) -> Dict[str, Any]: + """ + 获取世界观统计信息 + + Returns: + 统计信息字典 + """ + files = self.list_worldview_files() + + total_size = sum(f["size"] for f in files) + + return { + "worldview_count": len(files), + "total_size_kb": total_size / 1024, + "worldview_dir": str(self.worldview_dir), + "categories": list(self.CATEGORIES.keys()) + } + + +# 全局实例 +_global_worldview_builder: Optional[WorldviewBuilder] = None + + +def get_worldview_builder(cognee_manager=None) -> WorldviewBuilder: + """ + 获取全局世界观构建器实例（单例模式） + + Args: + cognee_manager: Cognee 记忆管理器（可选） + + Returns: + WorldviewBuilder 实例 + """ + global _global_worldview_builder + if _global_worldview_builder is None: + _global_worldview_builder = WorldviewBuilder(cognee_manager=cognee_manager) + return _global_worldview_builder + + +if __name__ == '__main__': + print("=" * 60) + print("世界观构建系统测试") + print("=" * 60) + + builder = WorldviewBuilder() + + # 测试列出世界观 + print("\n可用的世界观文件:") + files = builder.list_worldview_files() + for f in files: + print(f" - {f['name']} ({f['size']} bytes)") + + # 测试统计信息 + print("\n统计信息:") + stats = builder.get_statistics() + print(f" 世界观数量: {stats['worldview_count']}") + print(f" 总大小: {stats['total_size_kb']:.2f} KB") + + # 测试自然语言生成（模板） + print("\n测试自然语言生成:") + content = builder.create_worldview_from_natural_language( + "一个充满魔法的中世纪世界，有精灵、矮人和人类共存", + name="魔法世界", + use_llm=False + ) + print(f"生成内容长度: {len(content)} 字符") + print(content[:500] + "...") + + print("\n✓ 测试完成") diff --git a/src/gui/cognee_gui.py b/src/gui/cognee_gui.py new file mode 100644 index 0000000..7e876a5 --- /dev/null +++ b/src/gui/cognee_gui.py @@ -0,0 +1,1463 @@ +""" +Cognee 智能记忆系统 GUI +提供 Cognee 记忆管理和世界观构建的可视化界面 + +功能： +1. Cognee 记忆管理 - 添加、搜索、查看记忆 +2. 世界观构建 - 创建、编辑、管理 Markdown 世界观文件 +3. 知识图谱可视化 - 展示记忆关联 +""" + +import asyncio +import tkinter as tk +from tkinter import ttk, scrolledtext, messagebox, simpledialog +from datetime import datetime +from typing import Dict, Any, List + +from src.tools.debug_logger import get_debug_logger + +# 获取debug日志记录器 +debug_logger = get_debug_logger() + + +class CogneeMemoryGUI: + """ + Cognee 智能记忆管理 GUI + 提供记忆的添加、搜索、查看功能 + """ + + def __init__(self, parent_frame, cognee_manager=None): + """ + 初始化 Cognee 记忆管理 GUI + + Args: + parent_frame: 父容器 + cognee_manager: Cognee 记忆管理器实例 + """ + self.parent = parent_frame + self.cognee_manager = cognee_manager + + # 搜索结果缓存 + self._search_results_cache = [] + + # 创建界面 + self.create_widgets() + + debug_logger.log_info('CogneeMemoryGUI', 'Cognee 记忆管理 GUI 已初始化') + + def create_widgets(self): + """创建所有 GUI 组件""" + # 顶部标题栏 + header = ttk.Frame(self.parent) + header.pack(fill=tk.X, padx=5, pady=5) + + ttk.Label( + header, + text="🧠 Cognee 智能记忆系统", + font=("微软雅黑", 12, "bold") + ).pack(side=tk.LEFT, padx=5) + + # 状态指示器 + self.status_label = ttk.Label( + header, + text="", + font=("微软雅黑", 9) + ) + self.status_label.pack(side=tk.RIGHT, padx=5) + self._update_status() + + # 分割线 + ttk.Separator(self.parent, orient=tk.HORIZONTAL).pack(fill=tk.X, padx=5, pady=2) + + # 创建标签页 + self.notebook = ttk.Notebook(self.parent) + self.notebook.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) + + # 标签页1: 添加记忆 + self._create_add_memory_tab() + + # 标签页2: 搜索记忆 + self._create_search_tab() + + # 标签页3: 记忆统计 + self._create_stats_tab() + + def _update_status(self): + """更新状态指示器""" + if self.cognee_manager and self.cognee_manager._initialized: + self.status_label.config(text="🟢 Cognee 已连接", foreground="green") + else: + self.status_label.config(text="🔴 Cognee 未连接", foreground="red") + + def _create_add_memory_tab(self): + """创建添加记忆标签页""" + tab = ttk.Frame(self.notebook) + self.notebook.add(tab, text="➕ 添加记忆") + + # 记忆类型选择 + type_frame = ttk.Frame(tab) + type_frame.pack(fill=tk.X, padx=10, pady=10) + + ttk.Label(type_frame, text="记忆类型:", font=("微软雅黑", 10)).pack(side=tk.LEFT, padx=5) + + self.memory_type_var = tk.StringVar(value="conversation") + memory_types = [ + ("对话记忆", "conversation"), + ("知识记忆", "knowledge"), + ("世界观", "worldview") + ] + + for text, value in memory_types: + ttk.Radiobutton( + type_frame, + text=text, + value=value, + variable=self.memory_type_var + ).pack(side=tk.LEFT, padx=10) + + # 内容输入 + content_frame = ttk.LabelFrame(tab, text="记忆内容", padding=10) + content_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=5) + + self.memory_content_text = scrolledtext.ScrolledText( + content_frame, + height=15, + font=("微软雅黑", 10), + wrap=tk.WORD + ) + self.memory_content_text.pack(fill=tk.BOTH, expand=True) + + # 元数据输入 + meta_frame = ttk.LabelFrame(tab, text="元数据（可选）", padding=10) + meta_frame.pack(fill=tk.X, padx=10, pady=5) + + ttk.Label(meta_frame, text="实体/主题:").grid(row=0, column=0, padx=5, pady=5, sticky="w") + self.entity_entry = ttk.Entry(meta_frame, width=30) + self.entity_entry.grid(row=0, column=1, padx=5, pady=5, sticky="w") + + ttk.Label(meta_frame, text="来源:").grid(row=0, column=2, padx=5, pady=5, sticky="w") + self.source_entry = ttk.Entry(meta_frame, width=30) + self.source_entry.grid(row=0, column=3, padx=5, pady=5, sticky="w") + + # 按钮 + btn_frame = ttk.Frame(tab) + btn_frame.pack(fill=tk.X, padx=10, pady=10) + + ttk.Button( + btn_frame, + text="📝 添加记忆", + command=self._add_memory, + width=15 + ).pack(side=tk.LEFT, padx=5) + + ttk.Button( + btn_frame, + text="📝 添加并构建图谱", + command=self._add_and_cognify, + width=18 + ).pack(side=tk.LEFT, padx=5) + + ttk.Button( + btn_frame, + text="🔄 构建知识图谱", + command=self._cognify, + width=15 + ).pack(side=tk.LEFT, padx=5) + + ttk.Button( + btn_frame, + text="🧹 清空", + command=self._clear_input, + width=10 + ).pack(side=tk.RIGHT, padx=5) + + def _create_search_tab(self): + """创建搜索记忆标签页""" + tab = ttk.Frame(self.notebook) + self.notebook.add(tab, text="🔍 搜索记忆") + + # 搜索框 + search_frame = ttk.Frame(tab) + search_frame.pack(fill=tk.X, padx=10, pady=10) + + ttk.Label(search_frame, text="搜索:", font=("微软雅黑", 10)).pack(side=tk.LEFT, padx=5) + + self.search_entry = ttk.Entry(search_frame, width=50, font=("微软雅黑", 10)) + self.search_entry.pack(side=tk.LEFT, padx=5, fill=tk.X, expand=True) + self.search_entry.bind('', lambda e: self._search_memory()) + + ttk.Button( + search_frame, + text="🔍 搜索", + command=self._search_memory, + width=10 + ).pack(side=tk.LEFT, padx=5) + + # 搜索类型过滤 + filter_frame = ttk.Frame(tab) + filter_frame.pack(fill=tk.X, padx=10, pady=5) + + ttk.Label(filter_frame, text="过滤类型:").pack(side=tk.LEFT, padx=5) + + self.search_type_var = tk.StringVar(value="all") + search_types = [ + ("全部", "all"), + ("对话", "conversation"), + ("知识", "knowledge"), + ("世界观", "worldview") + ] + + for text, value in search_types: + ttk.Radiobutton( + filter_frame, + text=text, + value=value, + variable=self.search_type_var + ).pack(side=tk.LEFT, padx=10) + + # 搜索结果 + result_frame = ttk.LabelFrame(tab, text="搜索结果", padding=10) + result_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=5) + + # 创建 Treeview + columns = ("index", "content", "relevance") + self.search_tree = ttk.Treeview( + result_frame, + columns=columns, + show="headings", + height=15 + ) + + self.search_tree.heading("index", text="#") + self.search_tree.heading("content", text="内容") + self.search_tree.heading("relevance", text="相关度") + + self.search_tree.column("index", width=50, minwidth=40, stretch=False) + self.search_tree.column("content", width=500, minwidth=300, stretch=True) + self.search_tree.column("relevance", width=80, minwidth=60, stretch=False) + + # 滚动条 + scrollbar = ttk.Scrollbar(result_frame, orient=tk.VERTICAL, command=self.search_tree.yview) + self.search_tree.configure(yscrollcommand=scrollbar.set) + + self.search_tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) + scrollbar.pack(side=tk.RIGHT, fill=tk.Y) + + # 双击查看详情 + self.search_tree.bind("", self._show_result_detail) + + def _create_stats_tab(self): + """创建记忆统计标签页""" + tab = ttk.Frame(self.notebook) + self.notebook.add(tab, text="📊 统计信息") + + # 统计信息显示 + stats_frame = ttk.LabelFrame(tab, text="Cognee 记忆系统状态", padding=20) + stats_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) + + self.stats_text = scrolledtext.ScrolledText( + stats_frame, + height=20, + font=("Consolas", 10), + wrap=tk.WORD + ) + self.stats_text.pack(fill=tk.BOTH, expand=True) + + # 按钮 + btn_frame = ttk.Frame(tab) + btn_frame.pack(fill=tk.X, padx=10, pady=10) + + ttk.Button( + btn_frame, + text="🔄 刷新统计", + command=self._refresh_stats, + width=15 + ).pack(side=tk.LEFT, padx=5) + + ttk.Button( + btn_frame, + text="🗑️ 清空所有记忆", + command=self._clear_all_memory, + width=15 + ).pack(side=tk.RIGHT, padx=5) + + # 初始加载统计 + self._refresh_stats() + + def _run_async(self, coro): + """运行异步协程""" + try: + loop = asyncio.get_event_loop() + except RuntimeError: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + return loop.run_until_complete(coro) + + def _add_memory(self): + """添加记忆""" + if not self.cognee_manager or not self.cognee_manager._initialized: + messagebox.showwarning("警告", "Cognee 未初始化，无法添加记忆") + return + + content = self.memory_content_text.get("1.0", tk.END).strip() + if not content: + messagebox.showwarning("警告", "请输入记忆内容") + return + + memory_type = self.memory_type_var.get() + entity = self.entity_entry.get().strip() + source = self.source_entry.get().strip() + + metadata = {} + if entity: + metadata["entity"] = entity + if source: + metadata["source"] = source + + try: + success = self._run_async( + self.cognee_manager.add_memory(content, memory_type, metadata) + ) + + if success: + messagebox.showinfo("成功", "记忆已添加到 Cognee") + self._clear_input() + else: + messagebox.showerror("错误", "添加记忆失败") + + except Exception as e: + messagebox.showerror("错误", f"添加记忆时出错: {str(e)}") + + def _add_and_cognify(self): + """添加记忆并构建知识图谱""" + if not self.cognee_manager or not self.cognee_manager._initialized: + messagebox.showwarning("警告", "Cognee 未初始化") + return + + content = self.memory_content_text.get("1.0", tk.END).strip() + if not content: + messagebox.showwarning("警告", "请输入记忆内容") + return + + memory_type = self.memory_type_var.get() + + try: + success = self._run_async( + self.cognee_manager.process_and_store(content, memory_type) + ) + + if success: + messagebox.showinfo("成功", "记忆已添加并构建知识图谱") + self._clear_input() + else: + messagebox.showerror("错误", "处理失败") + + except Exception as e: + messagebox.showerror("错误", f"处理时出错: {str(e)}") + + def _cognify(self): + """构建知识图谱""" + if not self.cognee_manager or not self.cognee_manager._initialized: + messagebox.showwarning("警告", "Cognee 未初始化") + return + + try: + success = self._run_async(self.cognee_manager.cognify()) + + if success: + messagebox.showinfo("成功", "知识图谱已构建") + else: + messagebox.showerror("错误", "构建知识图谱失败") + + except Exception as e: + messagebox.showerror("错误", f"构建知识图谱时出错: {str(e)}") + + def _search_memory(self): + """搜索记忆""" + if not self.cognee_manager or not self.cognee_manager._initialized: + messagebox.showwarning("警告", "Cognee 未初始化") + return + + query = self.search_entry.get().strip() + if not query: + messagebox.showwarning("警告", "请输入搜索内容") + return + + search_type = self.search_type_var.get() + memory_type = None if search_type == "all" else search_type + + try: + results = self._run_async( + self.cognee_manager.search(query, memory_type=memory_type) + ) + + # 清空现有结果 + for item in self.search_tree.get_children(): + self.search_tree.delete(item) + + # 缓存结果 + self._search_results_cache = results + + # 添加结果 + for result in results: + content = result.get("content", "") + # 截断显示 + display_content = content[:100] + "..." if len(content) > 100 else content + + self.search_tree.insert("", "end", values=( + result.get("index", ""), + display_content, + f"{result.get('relevance', 0):.2f}" + )) + + if not results: + messagebox.showinfo("提示", "未找到相关记忆") + + except Exception as e: + messagebox.showerror("错误", f"搜索时出错: {str(e)}") + + def _show_result_detail(self, event): + """显示搜索结果详情""" + selection = self.search_tree.selection() + if not selection: + return + + item = self.search_tree.item(selection[0]) + index = int(item["values"][0]) - 1 + + if 0 <= index < len(self._search_results_cache): + result = self._search_results_cache[index] + content = result.get("content", "") + + # 显示详情对话框 + detail_window = tk.Toplevel(self.parent) + detail_window.title("记忆详情") + detail_window.geometry("600x400") + + text = scrolledtext.ScrolledText( + detail_window, + font=("微软雅黑", 10), + wrap=tk.WORD + ) + text.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) + text.insert("1.0", content) + text.config(state=tk.DISABLED) + + def _refresh_stats(self): + """刷新统计信息""" + self.stats_text.config(state=tk.NORMAL) + self.stats_text.delete("1.0", tk.END) + + if self.cognee_manager: + stats = self.cognee_manager.get_statistics() + + stats_text = """╔══════════════════════════════════════════════════════╗ +║ Cognee 智能记忆系统状态 ║ +╠══════════════════════════════════════════════════════╣ +║ ║ +║ 系统状态 ║ +║ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ ║ +""" + stats_text += f"║ • 启用状态: {'✓ 已启用' if stats.get('enabled') else '✗ 未启用':40}║\n" + stats_text += f"║ • 初始化: {'✓ 已初始化' if stats.get('initialized') else '✗ 未初始化':42}║\n" + stats_text += f"║ • API配置: {'✓ 已配置' if stats.get('api_key_configured') else '✗ 未配置':42}║\n" + stats_text += f"║ • 后端: {stats.get('backend', 'N/A'):45}║\n" + stats_text += """║ ║ +║ 功能说明 ║ +║ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ ║ +║ • 对话记忆: 存储和检索对话历史 ║ +║ • 知识记忆: 自动构建实体关系图谱 ║ +║ • 世界观: 虚拟世界设定和规则 ║ +║ • 语义搜索: 基于含义的智能检索 ║ +║ ║ +╚══════════════════════════════════════════════════════╝ + +Cognee 是一个开源的知识引擎，将原始数据转化为 +智能体的持久动态记忆。它结合向量搜索和图数据库， +使文档既可以按语义搜索，又能通过关系相互关联。 + +官方文档: https://docs.cognee.ai/ +""" + self.stats_text.insert("1.0", stats_text) + else: + self.stats_text.insert("1.0", "Cognee 管理器未初始化") + + self.stats_text.config(state=tk.DISABLED) + + def _clear_input(self): + """清空输入""" + self.memory_content_text.delete("1.0", tk.END) + self.entity_entry.delete(0, tk.END) + self.source_entry.delete(0, tk.END) + + def _clear_all_memory(self): + """清空所有记忆""" + if not self.cognee_manager or not self.cognee_manager._initialized: + messagebox.showwarning("警告", "Cognee 未初始化") + return + + if messagebox.askyesno("确认", "确定要清空所有 Cognee 记忆吗？\n此操作不可恢复！"): + try: + success = self._run_async(self.cognee_manager.clear_all_memory()) + + if success: + messagebox.showinfo("成功", "所有记忆已清空") + self._refresh_stats() + else: + messagebox.showerror("错误", "清空记忆失败") + + except Exception as e: + messagebox.showerror("错误", f"清空记忆时出错: {str(e)}") + + +class WorldviewBuilderGUI: + """ + 世界观构建系统 GUI + 提供 Markdown 世界观的创建、编辑、管理功能 + """ + + def __init__(self, parent_frame, worldview_builder=None): + """ + 初始化世界观构建 GUI + + Args: + parent_frame: 父容器 + worldview_builder: 世界观构建器实例 + """ + self.parent = parent_frame + self.worldview_builder = worldview_builder + + # 当前编辑的文件 + self.current_file = None + self.is_modified = False + + # 创建界面 + self.create_widgets() + + # 加载世界观列表 + self._refresh_worldview_list() + + debug_logger.log_info('WorldviewBuilderGUI', '世界观构建 GUI 已初始化') + + def create_widgets(self): + """创建所有 GUI 组件""" + # 主分割面板 + self.paned = ttk.PanedWindow(self.parent, orient=tk.HORIZONTAL) + self.paned.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) + + # 左侧：文件列表 + self._create_file_list_panel() + + # 右侧：编辑器 + self._create_editor_panel() + + def _create_file_list_panel(self): + """创建文件列表面板""" + left_frame = ttk.Frame(self.paned) + self.paned.add(left_frame, weight=1) + + # 标题 + header = ttk.Frame(left_frame) + header.pack(fill=tk.X, padx=5, pady=5) + + ttk.Label( + header, + text="🌍 世界观文件", + font=("微软雅黑", 10, "bold") + ).pack(side=tk.LEFT) + + # 工具栏 + toolbar = ttk.Frame(left_frame) + toolbar.pack(fill=tk.X, padx=5, pady=5) + + ttk.Button( + toolbar, + text="➕ 新建", + command=self._new_worldview, + width=8 + ).pack(side=tk.LEFT, padx=2) + + ttk.Button( + toolbar, + text="🗑️ 删除", + command=self._delete_worldview, + width=8 + ).pack(side=tk.LEFT, padx=2) + + ttk.Button( + toolbar, + text="🔄", + command=self._refresh_worldview_list, + width=3 + ).pack(side=tk.LEFT, padx=2) + + # 文件列表 + list_frame = ttk.Frame(left_frame) + list_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) + + self.file_listbox = tk.Listbox( + list_frame, + font=("微软雅黑", 10), + selectmode=tk.SINGLE + ) + + scrollbar = ttk.Scrollbar(list_frame, orient=tk.VERTICAL, command=self.file_listbox.yview) + self.file_listbox.configure(yscrollcommand=scrollbar.set) + + self.file_listbox.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) + scrollbar.pack(side=tk.RIGHT, fill=tk.Y) + + # 绑定选择事件 + self.file_listbox.bind('<>', self._on_file_select) + self.file_listbox.bind('', self._on_file_double_click) + + # 文件统计 + self.file_stats_label = ttk.Label(left_frame, text="", font=("微软雅黑", 9)) + self.file_stats_label.pack(fill=tk.X, padx=5, pady=5) + + def _create_editor_panel(self): + """创建编辑器面板""" + right_frame = ttk.Frame(self.paned) + self.paned.add(right_frame, weight=3) + + # 标题栏 + header = ttk.Frame(right_frame) + header.pack(fill=tk.X, padx=5, pady=5) + + self.file_title_label = ttk.Label( + header, + text="未选择文件", + font=("微软雅黑", 10, "bold") + ) + self.file_title_label.pack(side=tk.LEFT) + + self.modified_label = ttk.Label( + header, + text="", + font=("微软雅黑", 9), + foreground="orange" + ) + self.modified_label.pack(side=tk.LEFT, padx=10) + + # 工具栏 + toolbar = ttk.Frame(right_frame) + toolbar.pack(fill=tk.X, padx=5, pady=5) + + ttk.Button( + toolbar, + text="💾 保存", + command=self._save_worldview, + width=8 + ).pack(side=tk.LEFT, padx=2) + + ttk.Button( + toolbar, + text="🔄 重新加载", + command=self._reload_worldview, + width=10 + ).pack(side=tk.LEFT, padx=2) + + ttk.Separator(toolbar, orient=tk.VERTICAL).pack(side=tk.LEFT, padx=10, fill=tk.Y) + + ttk.Button( + toolbar, + text="🤖 AI生成", + command=self._generate_with_ai, + width=10 + ).pack(side=tk.LEFT, padx=2) + + ttk.Button( + toolbar, + text="☁️ 同步到Cognee", + command=self._sync_to_cognee, + width=12 + ).pack(side=tk.LEFT, padx=2) + + ttk.Button( + toolbar, + text="📚 同步到知识库", + command=self._sync_to_knowledge_base, + width=12 + ).pack(side=tk.LEFT, padx=2) + + # 编辑器 + editor_frame = ttk.LabelFrame(right_frame, text="Markdown 编辑器", padding=5) + editor_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) + + self.editor_text = scrolledtext.ScrolledText( + editor_frame, + font=("Consolas", 11), + wrap=tk.WORD, + undo=True + ) + self.editor_text.pack(fill=tk.BOTH, expand=True) + + # 绑定修改事件 + self.editor_text.bind('<>', self._on_text_modified) + self.editor_text.bind('', self._on_key_release) + + def _refresh_worldview_list(self): + """刷新世界观文件列表""" + self.file_listbox.delete(0, tk.END) + + if self.worldview_builder: + files = self.worldview_builder.list_worldview_files() + + for f in files: + self.file_listbox.insert(tk.END, f["name"]) + + self.file_stats_label.config( + text=f"共 {len(files)} 个世界观文件" + ) + else: + self.file_stats_label.config(text="世界观构建器未初始化") + + def _on_file_select(self, event): + """文件选择事件""" + pass # 单击只选中，不加载 + + def _on_file_double_click(self, event): + """文件双击事件""" + selection = self.file_listbox.curselection() + if not selection: + return + + # 检查是否有未保存的修改 + if self.is_modified: + if messagebox.askyesno("确认", "当前文件有未保存的修改，是否保存？"): + self._save_worldview() + + filename = self.file_listbox.get(selection[0]) + self._load_worldview(filename) + + def _load_worldview(self, name: str): + """加载世界观文件""" + if not self.worldview_builder: + messagebox.showwarning("警告", "世界观构建器未初始化") + return + + try: + content = self.worldview_builder.load_worldview(name) + + self.current_file = name + self.is_modified = False + + self.editor_text.delete("1.0", tk.END) + self.editor_text.insert("1.0", content) + self.editor_text.edit_modified(False) + + self.file_title_label.config(text=f"📄 {name}.md") + self.modified_label.config(text="") + + except FileNotFoundError: + messagebox.showerror("错误", f"文件不存在: {name}.md") + except Exception as e: + messagebox.showerror("错误", f"加载文件时出错: {str(e)}") + + def _save_worldview(self): + """保存世界观文件""" + if not self.current_file: + messagebox.showwarning("警告", "请先选择或创建一个文件") + return + + if not self.worldview_builder: + messagebox.showwarning("警告", "世界观构建器未初始化") + return + + content = self.editor_text.get("1.0", tk.END) + + try: + success = self.worldview_builder.save_worldview(self.current_file, content) + + if success: + self.is_modified = False + self.editor_text.edit_modified(False) + self.modified_label.config(text="") + messagebox.showinfo("成功", f"世界观已保存: {self.current_file}.md") + else: + messagebox.showerror("错误", "保存失败") + + except Exception as e: + messagebox.showerror("错误", f"保存时出错: {str(e)}") + + def _reload_worldview(self): + """重新加载当前文件""" + if not self.current_file: + return + + if self.is_modified: + if not messagebox.askyesno("确认", "放弃未保存的修改？"): + return + + self._load_worldview(self.current_file) + + def _new_worldview(self): + """新建世界观""" + name = simpledialog.askstring("新建世界观", "请输入世界观名称:") + + if not name: + return + + # 清理文件名 + name = name.strip().replace(" ", "_") + + if not self.worldview_builder: + messagebox.showwarning("警告", "世界观构建器未初始化") + return + + # 检查是否已存在 + files = self.worldview_builder.list_worldview_files() + if any(f["name"] == name for f in files): + messagebox.showwarning("警告", f"世界观 '{name}' 已存在") + return + + # 创建空模板 + template = f"""# {name} + +## 世界基本信息 + +**世界名称**：{name} + +**时代背景**： + +**地理位置**： + +## 世界特征 + +### 1. 基本设定 + + + +### 2. 规则与限制 + + + +### 3. 重要地点 + + + +### 4. 重要人物 + + + +## 注意事项 + +- 保持世界观的一致性 +- 自然地融入对话中 + +--- + +*创建时间: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}* +""" + + try: + self.worldview_builder.save_worldview(name, template) + self._refresh_worldview_list() + self._load_worldview(name) + messagebox.showinfo("成功", f"世界观 '{name}' 已创建") + + except Exception as e: + messagebox.showerror("错误", f"创建失败: {str(e)}") + + def _delete_worldview(self): + """删除世界观""" + selection = self.file_listbox.curselection() + if not selection: + messagebox.showwarning("警告", "请先选择要删除的世界观") + return + + filename = self.file_listbox.get(selection[0]) + + # 不允许删除默认世界观 + if filename == "default_world": + messagebox.showwarning("警告", "不能删除默认世界观") + return + + if not messagebox.askyesno("确认", f"确定要删除世界观 '{filename}' 吗？\n此操作不可恢复！"): + return + + if not self.worldview_builder: + return + + try: + success = self.worldview_builder.delete_worldview(filename) + + if success: + # 如果删除的是当前编辑的文件，清空编辑器 + if self.current_file == filename: + self.current_file = None + self.editor_text.delete("1.0", tk.END) + self.file_title_label.config(text="未选择文件") + + self._refresh_worldview_list() + messagebox.showinfo("成功", f"世界观 '{filename}' 已删除") + else: + messagebox.showerror("错误", "删除失败") + + except Exception as e: + messagebox.showerror("错误", f"删除时出错: {str(e)}") + + def _generate_with_ai(self): + """使用 AI 生成世界观""" + description = simpledialog.askstring( + "AI 生成世界观", + "请描述你想要的世界观:\n(例如: 一个充满魔法的中世纪世界)", + parent=self.parent + ) + + if not description: + return + + if not self.worldview_builder: + messagebox.showwarning("警告", "世界观构建器未初始化") + return + + try: + # 显示进度 + self.file_title_label.config(text="⏳ 正在生成...") + self.parent.update() + + content = self.worldview_builder.create_worldview_from_natural_language( + description, + use_llm=True + ) + + # 插入到编辑器 + self.editor_text.delete("1.0", tk.END) + self.editor_text.insert("1.0", content) + + self.is_modified = True + self.file_title_label.config(text="🆕 新生成的世界观（未保存）") + self.modified_label.config(text="● 未保存") + + messagebox.showinfo("成功", "世界观已生成，请编辑后保存") + + except Exception as e: + messagebox.showerror("错误", f"生成失败: {str(e)}") + self.file_title_label.config(text="生成失败") + + def _sync_to_cognee(self): + """同步到 Cognee""" + if not self.current_file: + messagebox.showwarning("警告", "请先选择一个世界观文件") + return + + if not self.worldview_builder: + messagebox.showwarning("警告", "世界观构建器未初始化") + return + + if not self.worldview_builder.cognee_manager: + messagebox.showwarning("警告", "Cognee 管理器未配置") + return + + loop = None + try: + # 运行异步方法 + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + success = loop.run_until_complete( + self.worldview_builder.sync_to_cognee(self.current_file) + ) + + if success: + messagebox.showinfo("成功", f"世界观已同步到 Cognee") + else: + messagebox.showerror("错误", "同步失败") + + except Exception as e: + messagebox.showerror("错误", f"同步时出错: {str(e)}") + finally: + # 确保事件循环在使用后被关闭，避免资源泄漏 + if loop is not None and not loop.is_closed(): + loop.close() + + def _sync_to_knowledge_base(self): + """同步到知识库""" + if not self.current_file: + messagebox.showwarning("警告", "请先选择一个世界观文件") + return + + if not self.worldview_builder: + messagebox.showwarning("警告", "世界观构建器未初始化") + return + + try: + count = self.worldview_builder.sync_to_knowledge_base(self.current_file) + + if count > 0: + messagebox.showinfo("成功", f"已同步 {count} 个世界观模块到知识库") + else: + messagebox.showwarning("警告", "没有模块被同步") + + except Exception as e: + messagebox.showerror("错误", f"同步时出错: {str(e)}") + + def _on_text_modified(self, event): + """文本修改事件""" + if self.editor_text.edit_modified(): + self.is_modified = True + self.modified_label.config(text="● 未保存") + + def _on_key_release(self, event): + """按键释放事件""" + if self.editor_text.edit_modified() and not self.is_modified: + self.is_modified = True + self.modified_label.config(text="● 未保存") + + +class CogneeLogViewerGUI: + """ + Cognee 日志查看器 GUI + 读取并展示 Cognee 系统日志 + """ + + def __init__(self, parent_frame): + """ + 初始化日志查看器 GUI + + Args: + parent_frame: 父容器 + """ + self.parent = parent_frame + self.log_files = [] + self.current_log = None + + # 自动刷新 + self.auto_refresh = False + self.refresh_job = None + + # 创建界面 + self.create_widgets() + + # 加载日志列表 + self._refresh_log_list() + + debug_logger.log_info('CogneeLogViewerGUI', 'Cognee 日志查看器 GUI 已初始化') + + def create_widgets(self): + """创建所有 GUI 组件""" + # 主分割面板 + self.paned = ttk.PanedWindow(self.parent, orient=tk.HORIZONTAL) + self.paned.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) + + # 左侧：日志文件列表 + self._create_log_list_panel() + + # 右侧：日志内容 + self._create_log_content_panel() + + def _create_log_list_panel(self): + """创建日志文件列表面板""" + left_frame = ttk.Frame(self.paned) + self.paned.add(left_frame, weight=1) + + # 标题 + header = ttk.Frame(left_frame) + header.pack(fill=tk.X, padx=5, pady=5) + + ttk.Label( + header, + text="📋 Cognee 日志文件", + font=("微软雅黑", 10, "bold") + ).pack(side=tk.LEFT) + + # 工具栏 + toolbar = ttk.Frame(left_frame) + toolbar.pack(fill=tk.X, padx=5, pady=5) + + ttk.Button( + toolbar, + text="🔄 刷新", + command=self._refresh_log_list, + width=8 + ).pack(side=tk.LEFT, padx=2) + + ttk.Button( + toolbar, + text="📂 打开目录", + command=self._open_log_directory, + width=10 + ).pack(side=tk.LEFT, padx=2) + + # 日志文件列表 + list_frame = ttk.Frame(left_frame) + list_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) + + self.log_listbox = tk.Listbox( + list_frame, + font=("微软雅黑", 9), + selectmode=tk.SINGLE + ) + + scrollbar = ttk.Scrollbar(list_frame, orient=tk.VERTICAL, command=self.log_listbox.yview) + self.log_listbox.configure(yscrollcommand=scrollbar.set) + + self.log_listbox.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) + scrollbar.pack(side=tk.RIGHT, fill=tk.Y) + + # 绑定选择事件 + self.log_listbox.bind('<>', self._on_log_select) + + # 文件统计 + self.log_stats_label = ttk.Label(left_frame, text="共 0 个日志文件", font=("微软雅黑", 9)) + self.log_stats_label.pack(fill=tk.X, padx=5, pady=5) + + def _create_log_content_panel(self): + """创建日志内容面板""" + right_frame = ttk.Frame(self.paned) + self.paned.add(right_frame, weight=3) + + # 标题栏 + header = ttk.Frame(right_frame) + header.pack(fill=tk.X, padx=5, pady=5) + + self.log_title_label = ttk.Label( + header, + text="选择一个日志文件查看", + font=("微软雅黑", 10, "bold") + ) + self.log_title_label.pack(side=tk.LEFT) + + # 工具栏 + toolbar = ttk.Frame(right_frame) + toolbar.pack(fill=tk.X, padx=5, pady=5) + + ttk.Button( + toolbar, + text="🔄 重新加载", + command=self._reload_log, + width=10 + ).pack(side=tk.LEFT, padx=2) + + # 自动刷新开关 + self.auto_refresh_var = tk.BooleanVar(value=False) + ttk.Checkbutton( + toolbar, + text="自动刷新", + variable=self.auto_refresh_var, + command=self._toggle_auto_refresh + ).pack(side=tk.LEFT, padx=10) + + # 过滤级别 + ttk.Label(toolbar, text="过滤:").pack(side=tk.LEFT, padx=(20, 5)) + self.filter_var = tk.StringVar(value="all") + filter_combo = ttk.Combobox( + toolbar, + textvariable=self.filter_var, + values=["all", "error", "warning", "info", "debug"], + width=10, + state="readonly" + ) + filter_combo.pack(side=tk.LEFT, padx=2) + filter_combo.bind('<>', lambda e: self._apply_filter()) + + # 搜索框 + ttk.Label(toolbar, text="搜索:").pack(side=tk.LEFT, padx=(20, 5)) + self.search_entry = ttk.Entry(toolbar, width=20) + self.search_entry.pack(side=tk.LEFT, padx=2) + self.search_entry.bind('', lambda e: self._search_log()) + + ttk.Button( + toolbar, + text="🔍", + command=self._search_log, + width=3 + ).pack(side=tk.LEFT, padx=2) + + # 日志内容 + content_frame = ttk.LabelFrame(right_frame, text="日志内容", padding=5) + content_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) + + self.log_text = scrolledtext.ScrolledText( + content_frame, + font=("Consolas", 9), + wrap=tk.NONE, + state=tk.DISABLED + ) + self.log_text.pack(fill=tk.BOTH, expand=True) + + # 水平滚动条 + h_scrollbar = ttk.Scrollbar(content_frame, orient=tk.HORIZONTAL, command=self.log_text.xview) + self.log_text.configure(xscrollcommand=h_scrollbar.set) + h_scrollbar.pack(fill=tk.X) + + # 配置文本标签（颜色高亮） + self.log_text.tag_configure("error", foreground="red") + self.log_text.tag_configure("warning", foreground="orange") + self.log_text.tag_configure("info", foreground="blue") + self.log_text.tag_configure("debug", foreground="gray") + self.log_text.tag_configure("highlight", background="yellow") + + def _get_log_directory(self): + """获取 Cognee 日志目录""" + import sys + import os + + # 尝试多个可能的日志目录 + possible_dirs = [] + + # 1. site-packages/logs + for path in sys.path: + if 'site-packages' in path: + logs_dir = os.path.join(os.path.dirname(path), 'site-packages', 'logs') + possible_dirs.append(logs_dir) + # 也检查直接在 site-packages 下的 logs + logs_dir2 = os.path.join(path, 'logs') + possible_dirs.append(logs_dir2) + + # 2. .venv/Lib/site-packages/logs (Windows) + venv_logs = os.path.join(os.getcwd(), '.venv', 'Lib', 'site-packages', 'logs') + possible_dirs.append(venv_logs) + + # 3. .venv/lib/python*/site-packages/logs (Linux/Mac) + import glob + linux_pattern = os.path.join(os.getcwd(), '.venv', 'lib', 'python*', 'site-packages', 'logs') + possible_dirs.extend(glob.glob(linux_pattern)) + + # 4. cognee 包内部的 logs 目录 + try: + import cognee + cognee_path = os.path.dirname(cognee.__file__) + possible_dirs.append(os.path.join(os.path.dirname(cognee_path), 'logs')) + possible_dirs.append(os.path.join(cognee_path, '..', 'logs')) + except ImportError: + pass + + # 查找存在的目录 + for logs_dir in possible_dirs: + if os.path.isdir(logs_dir): + return logs_dir + + return None + + def _refresh_log_list(self): + """刷新日志文件列表""" + import os + + self.log_listbox.delete(0, tk.END) + self.log_files = [] + + log_dir = self._get_log_directory() + + if log_dir and os.path.isdir(log_dir): + # 获取所有 .log 文件 + log_files = [f for f in os.listdir(log_dir) if f.endswith('.log')] + # 按时间倒序排列（最新的在前） + log_files.sort(reverse=True) + + for log_file in log_files: + self.log_listbox.insert(tk.END, log_file) + self.log_files.append(os.path.join(log_dir, log_file)) + + self.log_stats_label.config(text=f"共 {len(log_files)} 个日志文件 ({log_dir})") + else: + self.log_stats_label.config(text="未找到 Cognee 日志目录") + + def _on_log_select(self, event): + """日志文件选择事件""" + selection = self.log_listbox.curselection() + if selection: + index = selection[0] + if index < len(self.log_files): + self.current_log = self.log_files[index] + self._load_log_content() + + def _load_log_content(self): + """加载日志内容""" + import os + + if not self.current_log or not os.path.isfile(self.current_log): + return + + try: + with open(self.current_log, 'r', encoding='utf-8', errors='ignore') as f: + content = f.read() + + # 更新标题 + self.log_title_label.config(text=os.path.basename(self.current_log)) + + # 显示内容 + self.log_text.config(state=tk.NORMAL) + self.log_text.delete(1.0, tk.END) + + # 按行添加并高亮 + for line in content.split('\n'): + self._insert_log_line(line) + + self.log_text.config(state=tk.DISABLED) + + # 滚动到末尾 + self.log_text.see(tk.END) + + except Exception as e: + messagebox.showerror("错误", f"读取日志失败: {str(e)}") + + def _insert_log_line(self, line): + """插入日志行并应用颜色""" + line_lower = line.lower() + + if '[error' in line_lower or 'error]' in line_lower or 'exception' in line_lower: + tag = "error" + elif '[warning' in line_lower or 'warning]' in line_lower: + tag = "warning" + elif '[info' in line_lower or 'info]' in line_lower: + tag = "info" + elif '[debug' in line_lower or 'debug]' in line_lower: + tag = "debug" + else: + tag = None + + if tag: + self.log_text.insert(tk.END, line + '\n', tag) + else: + self.log_text.insert(tk.END, line + '\n') + + def _reload_log(self): + """重新加载当前日志""" + if self.current_log: + self._load_log_content() + + def _toggle_auto_refresh(self): + """切换自动刷新""" + if self.auto_refresh_var.get(): + self._start_auto_refresh() + else: + self._stop_auto_refresh() + + def _start_auto_refresh(self): + """启动自动刷新""" + self.auto_refresh = True + self._do_auto_refresh() + + def _stop_auto_refresh(self): + """停止自动刷新""" + self.auto_refresh = False + if self.refresh_job: + self.parent.after_cancel(self.refresh_job) + self.refresh_job = None + + def _do_auto_refresh(self): + """执行自动刷新""" + if self.auto_refresh and self.current_log: + self._load_log_content() + self.refresh_job = self.parent.after(2000, self._do_auto_refresh) + + def _apply_filter(self): + """应用过滤""" + if not self.current_log: + return + + import os + + filter_level = self.filter_var.get() + + try: + with open(self.current_log, 'r', encoding='utf-8', errors='ignore') as f: + content = f.read() + + self.log_text.config(state=tk.NORMAL) + self.log_text.delete(1.0, tk.END) + + for line in content.split('\n'): + line_lower = line.lower() + + # 根据过滤级别筛选 + if filter_level == "all": + self._insert_log_line(line) + elif filter_level == "error" and ('[error' in line_lower or 'error]' in line_lower or 'exception' in line_lower): + self._insert_log_line(line) + elif filter_level == "warning" and ('[warning' in line_lower or 'warning]' in line_lower): + self._insert_log_line(line) + elif filter_level == "info" and ('[info' in line_lower or 'info]' in line_lower): + self._insert_log_line(line) + elif filter_level == "debug" and ('[debug' in line_lower or 'debug]' in line_lower): + self._insert_log_line(line) + + self.log_text.config(state=tk.DISABLED) + + except Exception as e: + messagebox.showerror("错误", f"过滤日志失败: {str(e)}") + + def _search_log(self): + """搜索日志""" + search_text = self.search_entry.get().strip() + if not search_text or not self.current_log: + return + + # 移除之前的高亮 + self.log_text.tag_remove("highlight", "1.0", tk.END) + + # 搜索并高亮 + start_pos = "1.0" + while True: + pos = self.log_text.search(search_text, start_pos, stopindex=tk.END, nocase=True) + if not pos: + break + + end_pos = f"{pos}+{len(search_text)}c" + self.log_text.tag_add("highlight", pos, end_pos) + start_pos = end_pos + + # 滚动到第一个匹配 + first_match = self.log_text.tag_nextrange("highlight", "1.0") + if first_match: + self.log_text.see(first_match[0]) + + def _open_log_directory(self): + """打开日志目录""" + import os + import subprocess + import platform + + log_dir = self._get_log_directory() + + if log_dir and os.path.isdir(log_dir): + if platform.system() == 'Windows': + os.startfile(log_dir) + elif platform.system() == 'Darwin': # macOS + subprocess.run(['open', log_dir]) + else: # Linux + subprocess.run(['xdg-open', log_dir]) + else: + messagebox.showwarning("警告", "未找到 Cognee 日志目录") + + +class CogneeWorldviewManagerGUI: + """ + Cognee 记忆与世界观管理综合 GUI + 整合 Cognee 记忆管理、世界观构建和日志查看功能 + """ + + def __init__(self, parent_frame, cognee_manager=None, worldview_builder=None): + """ + 初始化综合管理 GUI + + Args: + parent_frame: 父容器 + cognee_manager: Cognee 记忆管理器 + worldview_builder: 世界观构建器 + """ + self.parent = parent_frame + self.cognee_manager = cognee_manager + self.worldview_builder = worldview_builder + + # 创建主标签页 + self.notebook = ttk.Notebook(self.parent) + self.notebook.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) + + # Cognee 记忆管理标签页 + cognee_frame = ttk.Frame(self.notebook) + self.notebook.add(cognee_frame, text="🧠 Cognee 记忆") + self.cognee_gui = CogneeMemoryGUI(cognee_frame, cognee_manager) + + # 世界观构建标签页 + worldview_frame = ttk.Frame(self.notebook) + self.notebook.add(worldview_frame, text="🌍 世界观构建") + self.worldview_gui = WorldviewBuilderGUI(worldview_frame, worldview_builder) + + # 日志查看标签页（新增） + log_frame = ttk.Frame(self.notebook) + self.notebook.add(log_frame, text="📋 Cognee 日志") + self.log_gui = CogneeLogViewerGUI(log_frame) + + debug_logger.log_info('CogneeWorldviewManagerGUI', 'Cognee 与世界观管理 GUI 已初始化') diff --git a/src/gui/database_gui.py b/src/gui/database_gui.py index 47b2eaa..199b2a1 100644 --- a/src/gui/database_gui.py +++ b/src/gui/database_gui.py @@ -1,6 +1,7 @@ """ 数据库管理GUI组件 提供可视化界面管理数据库内容 +集成 Cognee 智能记忆和世界观构建功能 """ import tkinter as tk @@ -16,16 +17,21 @@ class DatabaseManagerGUI: 提供数据库的可视化管理、编辑功能 """ - def __init__(self, parent_frame, db_manager: DatabaseManager): + def __init__(self, parent_frame, db_manager: DatabaseManager, + cognee_manager=None, worldview_builder=None): """ 初始化数据库管理GUI Args: parent_frame: 父容器 db_manager: 数据库管理器实例 + cognee_manager: Cognee 记忆管理器实例（可选） + worldview_builder: 世界观构建器实例（可选） """ self.parent = parent_frame self.db = db_manager + self.cognee_manager = cognee_manager + self.worldview_builder = worldview_builder # 自动刷新相关 self.auto_refresh_enabled = True @@ -124,8 +130,11 @@ def create_widgets(self): # 标签页8：日程管理 self.create_schedules_tab() - # 标签页8：日程管理 - self.create_schedules_tab() + # 标签页9：Cognee 智能记忆（新增） + self.create_cognee_memory_tab() + + # 标签页10：世界观构建（新增） + self.create_worldview_tab() def create_base_knowledge_tab(self): """ @@ -1756,3 +1765,159 @@ def delete_schedule_data(self): except Exception as e: messagebox.showerror("错误", f"删除失败:\n{str(e)}") + + # ==================== Cognee 智能记忆标签页 ==================== + + def create_cognee_memory_tab(self): + """ + 创建 Cognee 智能记忆管理标签页 + """ + tab = ttk.Frame(self.notebook) + self.notebook.add(tab, text="🧠 Cognee记忆") + + # 检查 Cognee 是否可用 + if self.cognee_manager is None: + # 显示未配置提示 + info_frame = ttk.Frame(tab) + info_frame.pack(fill=tk.BOTH, expand=True, padx=20, pady=20) + + ttk.Label( + info_frame, + text="🧠 Cognee 智能记忆系统", + font=("微软雅黑", 14, "bold") + ).pack(pady=20) + + info_text = """ +Cognee 是一个开源的知识引擎，将原始数据转化为 +智能体的持久动态记忆。它结合向量搜索和图数据库， +使文档既可以按语义搜索，又能通过关系相互关联。 + +功能特性： +• 对话记忆：存储和检索对话历史 +• 知识图谱：自动构建实体关系图谱 +• 语义搜索：基于含义的智能检索 +• 模块化知识：支持自定义知识块 + +要启用 Cognee 记忆系统，请： +1. 安装 cognee: pip install cognee +2. 在 .env 中配置 COGNEE_ENABLED=true +3. 配置 LLM_API_KEY 或 SILICONFLOW_API_KEY + +官方文档: https://docs.cognee.ai/ + """ + + ttk.Label( + info_frame, + text=info_text, + font=("微软雅黑", 10), + justify=tk.LEFT + ).pack(pady=10) + + return + + # 如果 Cognee 可用，加载完整的 GUI + try: + from src.gui.cognee_gui import CogneeMemoryGUI + self.cognee_gui = CogneeMemoryGUI(tab, self.cognee_manager) + except ImportError as e: + ttk.Label( + tab, + text=f"加载 Cognee GUI 失败: {str(e)}", + font=("微软雅黑", 10), + foreground="red" + ).pack(pady=20) + + # ==================== 世界观构建标签页 ==================== + + def create_worldview_tab(self): + """ + 创建世界观构建标签页 + """ + tab = ttk.Frame(self.notebook) + self.notebook.add(tab, text="🌍 世界观") + + # 检查世界观构建器是否可用 + if self.worldview_builder is None: + # 显示未配置提示 + info_frame = ttk.Frame(tab) + info_frame.pack(fill=tk.BOTH, expand=True, padx=20, pady=20) + + ttk.Label( + info_frame, + text="🌍 世界观构建系统", + font=("微软雅黑", 14, "bold") + ).pack(pady=20) + + info_text = """ +世界观构建系统允许您创建和管理智能体的虚拟世界设定。 + +功能特性： +• 基于 Markdown 的世界观文件管理 +• 直接编辑源文本 +• 自然语言创建世界观 +• AI 辅助生成详细设定 +• 与 Cognee 记忆系统集成 +• 同步到知识库 + +世界观文件位于: prompts/worldview/ + +您可以直接编辑 Markdown 文件，或使用此界面管理。 + """ + + ttk.Label( + info_frame, + text=info_text, + font=("微软雅黑", 10), + justify=tk.LEFT + ).pack(pady=10) + + # 提供基础功能按钮 + btn_frame = ttk.Frame(info_frame) + btn_frame.pack(pady=20) + + ttk.Button( + btn_frame, + text="📁 打开世界观目录", + command=self._open_worldview_directory, + width=20 + ).pack(pady=5) + + return + + # 如果世界观构建器可用，加载完整的 GUI + try: + from src.gui.cognee_gui import WorldviewBuilderGUI + self.worldview_gui = WorldviewBuilderGUI(tab, self.worldview_builder) + except ImportError as e: + ttk.Label( + tab, + text=f"加载世界观 GUI 失败: {str(e)}", + font=("微软雅黑", 10), + foreground="red" + ).pack(pady=20) + + def _open_worldview_directory(self): + """打开世界观文件目录""" + import os + import subprocess + import platform + from pathlib import Path + + # 获取世界观目录路径 + current_file = Path(__file__).resolve() + project_root = current_file.parent.parent.parent + worldview_dir = project_root / "prompts" / "worldview" + + if not worldview_dir.exists(): + worldview_dir.mkdir(parents=True, exist_ok=True) + + # 根据操作系统打开目录 + try: + if platform.system() == "Windows": + os.startfile(str(worldview_dir)) + elif platform.system() == "Darwin": # macOS + subprocess.run(["open", str(worldview_dir)]) + else: # Linux + subprocess.run(["xdg-open", str(worldview_dir)]) + except Exception as e: + messagebox.showerror("错误", f"无法打开目录: {str(e)}") diff --git a/src/tools/debug_logger.py b/src/tools/debug_logger.py index c614cc8..2e8ab5f 100644 --- a/src/tools/debug_logger.py +++ b/src/tools/debug_logger.py @@ -326,6 +326,81 @@ def log_info(self, module_name: str, message: str, data: Any = None): self._write_to_file(log_message) self._notify_listeners(log_entry) + def log_warning(self, module_name: str, message: str, data: Any = None): + """ + 记录警告信息 + + Args: + module_name: 模块名称 + message: 警告内容 + data: 附加数据 + """ + if not self.debug_mode: + return + + log_entry = { + 'timestamp': datetime.now().isoformat(), + 'type': 'warning', + 'module': module_name, + 'message': message, + 'data': data + } + + self.logs.append(log_entry) + # 统计到info类型（或可以新增warning统计） + self.log_stats['info'] += 1 + + log_message = f"[{log_entry['timestamp']}] [WARNING] {module_name} | {message}" + if data: + log_message += f"\n 数据: {json.dumps(data, ensure_ascii=False) if isinstance(data, (dict, list)) else str(data)}" + + print(log_message) + self._write_to_file(log_message) + self._notify_listeners(log_entry) + + def log_raw_output(self, module_name: str, context: str, raw_content: str, error_message: str = None): + """ + 记录原始LLM输出（用于调试JSON解析失败等问题） + + Args: + module_name: 模块名称 + context: 上下文描述（如"实体提取"、"知识提取"等） + raw_content: 完整的原始输出内容 + error_message: 可选的错误消息 + """ + if not self.debug_mode: + return + + log_entry = { + 'timestamp': datetime.now().isoformat(), + 'type': 'raw_output', + 'module': module_name, + 'context': context, + 'raw_content': raw_content, + 'error_message': error_message + } + + self.logs.append(log_entry) + # 统计到info类型 + self.log_stats['info'] += 1 + + # 控制台输出摘要 + display_content = raw_content[:200] + "..." if len(raw_content) > 200 else raw_content + log_message = f"[{log_entry['timestamp']}] [RAW_OUTPUT] {module_name} | {context}" + if error_message: + log_message += f"\n 错误: {error_message}" + log_message += f"\n 内容预览: {display_content}" + + print(log_message) + + # 写入完整内容到文件（便于排查问题） + self._write_to_file(log_message) + self._write_to_file(f" ===== 完整原始输出开始 =====") + self._write_to_file(raw_content) + self._write_to_file(f" ===== 完整原始输出结束 =====\n") + + self._notify_listeners(log_entry) + def get_logs(self, log_type: str = None, module_name: str = None, limit: int = None) -> List[Dict[str, Any]]: """ 获取日志列表 diff --git a/tests/test_cognee_integration.py b/tests/test_cognee_integration.py new file mode 100644 index 0000000..d842fa3 --- /dev/null +++ b/tests/test_cognee_integration.py @@ -0,0 +1,340 @@ +""" +Cognee 智能记忆系统和世界观构建系统测试 +""" + +import os +import sys +import unittest +from pathlib import Path + +# 添加项目根目录到路径 +sys.path.insert(0, str(Path(__file__).parent.parent)) + + +class TestCogneeMemoryManager(unittest.TestCase): + """测试 Cognee 记忆管理器""" + + def setUp(self): + """测试前设置""" + # 记录原始环境变量值，避免污染其他测试 + self._old_cognee_enabled = os.environ.get('COGNEE_ENABLED') + # 禁用 Cognee 以避免依赖问题 + os.environ['COGNEE_ENABLED'] = 'false' + + def tearDown(self): + """测试后恢复环境变量""" + if self._old_cognee_enabled is None: + # 原来未设置，则删除 + os.environ.pop('COGNEE_ENABLED', None) + else: + # 恢复原始值 + os.environ['COGNEE_ENABLED'] = self._old_cognee_enabled + + def test_import_cognee_memory(self): + """测试导入 Cognee 记忆模块""" + from src.core.cognee_memory import CogneeMemoryManager, CogneeMemorySyncWrapper + self.assertTrue(callable(CogneeMemoryManager)) + self.assertTrue(callable(CogneeMemorySyncWrapper)) + + def test_cognee_manager_disabled(self): + """测试禁用状态下的 Cognee 管理器""" + from src.core.cognee_memory import CogneeMemoryManager + + manager = CogneeMemoryManager(enabled=False) + + self.assertFalse(manager.enabled) + self.assertFalse(manager._initialized) + + def test_cognee_manager_statistics(self): + """测试获取统计信息""" + from src.core.cognee_memory import CogneeMemoryManager + + manager = CogneeMemoryManager(enabled=False) + stats = manager.get_statistics() + + self.assertIn('enabled', stats) + self.assertIn('initialized', stats) + self.assertIn('backend', stats) + self.assertEqual(stats['backend'], 'cognee') + + def test_sync_wrapper_creation(self): + """测试同步包装器创建""" + from src.core.cognee_memory import CogneeMemorySyncWrapper, CogneeMemoryManager + + manager = CogneeMemoryManager(enabled=False) + wrapper = CogneeMemorySyncWrapper(manager) + + self.assertIsNotNone(wrapper.manager) + + +class TestWorldviewBuilder(unittest.TestCase): + """测试世界观构建器""" + + def setUp(self): + """测试前设置""" + self.test_dir = Path(__file__).parent / "test_worldview_temp" + self.test_dir.mkdir(exist_ok=True) + + def tearDown(self): + """测试后清理""" + import shutil + if self.test_dir.exists(): + shutil.rmtree(self.test_dir) + + def test_import_worldview_builder(self): + """测试导入世界观构建模块""" + from src.core.worldview_builder import WorldviewBuilder, WorldviewModule + self.assertTrue(callable(WorldviewBuilder)) + self.assertTrue(callable(WorldviewModule)) + + def test_worldview_module_creation(self): + """测试世界观模块创建""" + from src.core.worldview_builder import WorldviewModule + + module = WorldviewModule( + name="测试模块", + category="general", + content="这是测试内容" + ) + + self.assertEqual(module.name, "测试模块") + self.assertEqual(module.category, "general") + self.assertEqual(module.content, "这是测试内容") + + def test_worldview_module_to_dict(self): + """测试世界观模块转字典""" + from src.core.worldview_builder import WorldviewModule + + module = WorldviewModule( + name="测试", + category="rules", + content="规则内容" + ) + + data = module.to_dict() + + self.assertIn('name', data) + self.assertIn('category', data) + self.assertIn('content', data) + self.assertIn('created_at', data) + + def test_worldview_module_from_dict(self): + """测试从字典创建世界观模块""" + from src.core.worldview_builder import WorldviewModule + + data = { + 'name': '测试', + 'category': 'locations', + 'content': '地点内容' + } + + module = WorldviewModule.from_dict(data) + + self.assertEqual(module.name, '测试') + self.assertEqual(module.category, 'locations') + + def test_worldview_builder_creation(self): + """测试世界观构建器创建""" + from src.core.worldview_builder import WorldviewBuilder + + builder = WorldviewBuilder(worldview_dir=str(self.test_dir)) + + self.assertTrue(self.test_dir.exists()) + + def test_worldview_builder_categories(self): + """测试世界观分类""" + from src.core.worldview_builder import WorldviewBuilder + + self.assertIn('general', WorldviewBuilder.CATEGORIES) + self.assertIn('rules', WorldviewBuilder.CATEGORIES) + self.assertIn('locations', WorldviewBuilder.CATEGORIES) + self.assertIn('characters', WorldviewBuilder.CATEGORIES) + + def test_create_worldview_template(self): + """测试创建世界观模板""" + from src.core.worldview_builder import WorldviewBuilder + + builder = WorldviewBuilder(worldview_dir=str(self.test_dir)) + + content = builder.create_worldview_from_natural_language( + "一个测试世界", + name="测试世界", + use_llm=False + ) + + self.assertIn("测试世界", content) + self.assertIn("#", content) # Markdown 标题 + + def test_save_and_load_worldview(self): + """测试保存和加载世界观""" + from src.core.worldview_builder import WorldviewBuilder + + builder = WorldviewBuilder(worldview_dir=str(self.test_dir)) + + # 保存 + test_content = "# 测试世界观\n\n这是测试内容" + success = builder.save_worldview("test_world", test_content) + self.assertTrue(success) + + # 加载 + loaded_content = builder.load_worldview("test_world") + self.assertEqual(loaded_content, test_content) + + def test_list_worldview_files(self): + """测试列出世界观文件""" + from src.core.worldview_builder import WorldviewBuilder + + builder = WorldviewBuilder(worldview_dir=str(self.test_dir)) + + # 创建测试文件 + builder.save_worldview("world1", "# World 1") + builder.save_worldview("world2", "# World 2") + + files = builder.list_worldview_files() + + self.assertEqual(len(files), 2) + names = [f['name'] for f in files] + self.assertIn('world1', names) + self.assertIn('world2', names) + + def test_delete_worldview(self): + """测试删除世界观""" + from src.core.worldview_builder import WorldviewBuilder + + builder = WorldviewBuilder(worldview_dir=str(self.test_dir)) + + # 创建并删除 + builder.save_worldview("to_delete", "# Delete me") + self.assertTrue((self.test_dir / "to_delete.md").exists()) + + success = builder.delete_worldview("to_delete") + self.assertTrue(success) + self.assertFalse((self.test_dir / "to_delete.md").exists()) + + def test_parse_worldview_to_modules(self): + """测试解析世界观为模块""" + from src.core.worldview_builder import WorldviewBuilder + + builder = WorldviewBuilder(worldview_dir=str(self.test_dir)) + + content = """# 测试世界 + +## 基本信息 + +这是基本信息内容。 + +## 规则设定 + +这是规则内容。 + +## 重要地点 + +这是地点内容。 +""" + + modules = builder.parse_worldview_to_modules(content) + + self.assertGreater(len(modules), 0) + # 检查模块是否被正确解析 + module_names = [m.name for m in modules] + self.assertIn("基本信息", module_names) + + def test_infer_category(self): + """测试分类推断""" + from src.core.worldview_builder import WorldviewBuilder + + builder = WorldviewBuilder(worldview_dir=str(self.test_dir)) + + self.assertEqual(builder._infer_category("基本信息"), "general") + self.assertEqual(builder._infer_category("规则设定"), "rules") + self.assertEqual(builder._infer_category("重要地点"), "locations") + self.assertEqual(builder._infer_category("角色人物"), "characters") + + def test_get_statistics(self): + """测试获取统计信息""" + from src.core.worldview_builder import WorldviewBuilder + + builder = WorldviewBuilder(worldview_dir=str(self.test_dir)) + + # 创建一些测试文件 + builder.save_worldview("stat_test1", "# Test 1") + builder.save_worldview("stat_test2", "# Test 2") + + stats = builder.get_statistics() + + self.assertEqual(stats['worldview_count'], 2) + self.assertIn('total_size_kb', stats) + self.assertIn('categories', stats) + + +class TestLongTermMemoryIntegration(unittest.TestCase): + """测试长期记忆与 Cognee 集成""" + + def setUp(self): + """测试前设置""" + os.environ['COGNEE_ENABLED'] = 'false' + self.test_db = Path(__file__).parent / "test_ltm_temp.db" + + def tearDown(self): + """测试后清理""" + if self.test_db.exists(): + self.test_db.unlink() + + def test_long_term_memory_import(self): + """测试导入长期记忆模块""" + from src.core.long_term_memory import LongTermMemoryManager + self.assertTrue(callable(LongTermMemoryManager)) + + def test_long_term_memory_cognee_disabled(self): + """测试禁用 Cognee 的长期记忆""" + from src.core.long_term_memory import LongTermMemoryManager + from src.core.database_manager import DatabaseManager + + # 使用临时文件数据库 + db = DatabaseManager(db_path=str(self.test_db)) + manager = LongTermMemoryManager(db_manager=db, enable_cognee=False) + + self.assertIsNone(manager.cognee_manager) + + def test_statistics_includes_cognee(self): + """测试统计信息包含 Cognee""" + from src.core.long_term_memory import LongTermMemoryManager + from src.core.database_manager import DatabaseManager + + db = DatabaseManager(db_path=str(self.test_db)) + manager = LongTermMemoryManager(db_manager=db, enable_cognee=False) + + stats = manager.get_statistics() + + self.assertIn('cognee', stats) + + +class TestCogneeGUI(unittest.TestCase): + """测试 Cognee GUI 模块""" + + def test_import_cognee_gui(self): + """测试导入 Cognee GUI""" + from src.gui.cognee_gui import CogneeMemoryGUI, WorldviewBuilderGUI, CogneeWorldviewManagerGUI + self.assertTrue(callable(CogneeMemoryGUI)) + self.assertTrue(callable(WorldviewBuilderGUI)) + self.assertTrue(callable(CogneeWorldviewManagerGUI)) + + +class TestDatabaseGUIIntegration(unittest.TestCase): + """测试数据库 GUI 集成""" + + def test_database_gui_accepts_cognee_params(self): + """测试数据库 GUI 接受 Cognee 参数""" + from src.gui.database_gui import DatabaseManagerGUI + import inspect + + # 检查初始化方法是否接受 cognee_manager 和 worldview_builder 参数 + sig = inspect.signature(DatabaseManagerGUI.__init__) + params = list(sig.parameters.keys()) + + self.assertIn('cognee_manager', params) + self.assertIn('worldview_builder', params) + + +if __name__ == '__main__': + unittest.main()