gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(server): 将记忆可视化工具和统计功能整合进主服务

Browse Source 
将原先独立的记忆可视化工具（Memory Visualizer）和LLM使用统计逻辑深度整合到项目主服务中。

主要变更包括：
- **移除独立的可视化工具**: 删除了 `tools/memory_visualizer` 目录下的所有独立服务器、脚本和文档，清理了项目结构。
- **API路由整合**: 在主 FastAPI 应用中注册了记忆可视化工具的路由，使其成为核心功能的一部分，可通过 `/visualizer` 访问。
- **统计逻辑重构**: 将LLM使用统计的计算逻辑从API路由层 `statistic_router.py` 中剥离，迁移到 `src/chat/utils/statistic.py` 中，实现了逻辑的解耦和复用。API路由现在直接调用重构后的统计任务。
- **依赖清理与添加**: 添加了 `jinja2` 作为模板渲染的依赖，并清除了与独立可视化工具相关的旧依赖。

此次重构简化了项目的维护和部署，将原本分散的功能统一管理，提升了代码的内聚性和可维护性。
This commit is contained in:
minecraft1024a
2025-11-07 21:12:11 +08:00
parent 33897bec53
commit 077628930b
23 changed files with 376 additions and 2543 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
requirements.txt
View File

@@ -71,9 +71,10 @@ lunar_python
fuzzywuzzy
python-multipart
aiofiles
jinja2
inkfox
soundfile
pedalboard
# For local speech-to-text functionality (stt_whisper_plugin)
openai-whisper
openai-whisper

361
src/api/memory_visualizer_router.py Normal file
View File

@@ -0,0 +1,361 @@
"""
记忆图可视化 - API 路由模块
提供 Web API 用于可视化记忆图数据
"""
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional
import orjson
from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.templating import Jinja2Templates
# 调整项目根目录的计算方式
project_root = Path(__file__).parent.parent.parent
data_dir = project_root / "data" / "memory_graph"
# 缓存
graph_data_cache = None
current_data_file = None
# FastAPI 路由
router = APIRouter()
# Jinja2 模板引擎
templates = Jinja2Templates(directory=str(Path(__file__).parent / "templates"))
def find_available_data_files() -> List[Path]:
"""查找所有可用的记忆图数据文件"""
files = []
if not data_dir.exists():
return files
possible_files = ["graph_store.json", "memory_graph.json", "graph_data.json"]
for filename in possible_files:
file_path = data_dir / filename
if file_path.exists():
files.append(file_path)
for pattern in ["graph_store_*.json", "memory_graph_*.json", "graph_data_*.json"]:
for backup_file in data_dir.glob(pattern):
if backup_file not in files:
files.append(backup_file)
backups_dir = data_dir / "backups"
if backups_dir.exists():
for backup_file in backups_dir.glob("**/*.json"):
if backup_file not in files:
files.append(backup_file)
backup_dir = data_dir.parent / "backup"
if backup_dir.exists():
for pattern in ["**/graph_*.json", "**/memory_*.json"]:
for backup_file in backup_dir.glob(pattern):
if backup_file not in files:
files.append(backup_file)
return sorted(files, key=lambda f: f.stat().st_mtime, reverse=True)
def load_graph_data_from_file(file_path: Optional[Path] = None) -> Dict[str, Any]:
"""从磁盘加载图数据"""
global graph_data_cache, current_data_file
if file_path and file_path != current_data_file:
graph_data_cache = None
current_data_file = file_path
if graph_data_cache:
return graph_data_cache
try:
graph_file = current_data_file
if not graph_file:
available_files = find_available_data_files()
if not available_files:
return {"error": "未找到数据文件", "nodes": [], "edges": [], "stats": {}}
graph_file = available_files[0]
current_data_file = graph_file
if not graph_file.exists():
return {"error": f"文件不存在: {graph_file}", "nodes": [], "edges": [], "stats": {}}
with open(graph_file, "r", encoding="utf-8") as f:
data = orjson.loads(f.read())
nodes = data.get("nodes", [])
edges = data.get("edges", [])
metadata = data.get("metadata", {})
nodes_dict = {
node["id"]: {
**node,
"label": node.get("content", ""),
"group": node.get("node_type", ""),
"title": f"{node.get('node_type', '')}: {node.get('content', '')}",
}
for node in nodes
if node.get("id")
}
edges_list = [
{
**edge,
"from": edge.get("source", edge.get("source_id")),
"to": edge.get("target", edge.get("target_id")),
"label": edge.get("relation", ""),
"arrows": "to",
}
for edge in edges
]
stats = metadata.get("statistics", {})
total_memories = stats.get("total_memories", 0)
graph_data_cache = {
"nodes": list(nodes_dict.values()),
"edges": edges_list,
"memories": [],
"stats": {
"total_nodes": len(nodes_dict),
"total_edges": len(edges_list),
"total_memories": total_memories,
},
"current_file": str(graph_file),
"file_size": graph_file.stat().st_size,
"file_modified": datetime.fromtimestamp(graph_file.stat().st_mtime).isoformat(),
}
return graph_data_cache
except Exception as e:
import traceback
traceback.print_exc()
raise HTTPException(status_code=500, detail=f"加载图数据失败: {e}")
@router.get("/", response_class=HTMLResponse)
async def index(request: Request):
"""主页面"""
return templates.TemplateResponse("visualizer.html", {"request": request})
def _format_graph_data_from_manager(memory_manager) -> Dict[str, Any]:
"""从 MemoryManager 提取并格式化图数据"""
if not memory_manager.graph_store:
return {"nodes": [], "edges": [], "memories": [], "stats": {}}
all_memories = memory_manager.graph_store.get_all_memories()
nodes_dict = {}
edges_list = []
memory_info = []
for memory in all_memories:
memory_info.append(
{
"id": memory.id,
"type": memory.memory_type.value,
"importance": memory.importance,
"text": memory.to_text(),
}
)
for node in memory.nodes:
if node.id not in nodes_dict:
nodes_dict[node.id] = {
"id": node.id,
"label": node.content,
"type": node.node_type.value,
"group": node.node_type.name,
"title": f"{node.node_type.value}: {node.content}",
}
for edge in memory.edges:
edges_list.append( # noqa: PERF401
{
"id": edge.id,
"from": edge.source_id,
"to": edge.target_id,
"label": edge.relation,
"arrows": "to",
"memory_id": memory.id,
}
)
stats = memory_manager.get_statistics()
return {
"nodes": list(nodes_dict.values()),
"edges": edges_list,
"memories": memory_info,
"stats": {
"total_nodes": stats.get("total_nodes", 0),
"total_edges": stats.get("total_edges", 0),
"total_memories": stats.get("total_memories", 0),
},
"current_file": "memory_manager (实时数据)",
}
@router.get("/api/graph/full")
async def get_full_graph():
"""获取完整记忆图数据"""
try:
from src.memory_graph.manager_singleton import get_memory_manager
memory_manager = get_memory_manager()
data = {}
if memory_manager and memory_manager._initialized:
data = _format_graph_data_from_manager(memory_manager)
else:
# 如果内存管理器不可用，则从文件加载
data = load_graph_data_from_file()
return JSONResponse(content={"success": True, "data": data})
except Exception as e:
import traceback
traceback.print_exc()
return JSONResponse(content={"success": False, "error": str(e)}, status_code=500)
@router.get("/api/files")
async def list_files_api():
"""列出所有可用的数据文件"""
try:
files = find_available_data_files()
file_list = []
for f in files:
stat = f.stat()
file_list.append(
{
"path": str(f),
"name": f.name,
"size": stat.st_size,
"size_kb": round(stat.st_size / 1024, 2),
"modified": datetime.fromtimestamp(stat.st_mtime).isoformat(),
"modified_readable": datetime.fromtimestamp(stat.st_mtime).strftime("%Y-%m-%d %H:%M:%S"),
"is_current": str(f) == str(current_data_file) if current_data_file else False,
}
)
return JSONResponse(
content={
"success": True,
"files": file_list,
"count": len(file_list),
"current_file": str(current_data_file) if current_data_file else None,
}
)
except Exception as e:
# 增加日志记录
# logger.error(f"列出数据文件失败: {e}", exc_info=True)
return JSONResponse(content={"success": False, "error": str(e)}, status_code=500)
@router.post("/select_file")
async def select_file(request: Request):
"""选择要加载的数据文件"""
global graph_data_cache, current_data_file
try:
data = await request.json()
file_path = data.get("file_path")
if not file_path:
raise HTTPException(status_code=400, detail="未提供文件路径")
file_to_load = Path(file_path)
if not file_to_load.exists():
raise HTTPException(status_code=404, detail=f"文件不存在: {file_path}")
graph_data_cache = None
current_data_file = file_to_load
graph_data = load_graph_data_from_file(file_to_load)
return JSONResponse(
content={
"success": True,
"message": f"已切换到文件: {file_to_load.name}",
"stats": graph_data.get("stats", {}),
}
)
except Exception as e:
return JSONResponse(content={"success": False, "error": str(e)}, status_code=500)
@router.get("/reload")
async def reload_data():
"""重新加载数据"""
global graph_data_cache
graph_data_cache = None
data = load_graph_data_from_file()
return JSONResponse(content={"success": True, "message": "数据已重新加载", "stats": data.get("stats", {})})
@router.get("/api/search")
async def search_memories(q: str, limit: int = 50):
"""搜索记忆"""
try:
from src.memory_graph.manager_singleton import get_memory_manager
memory_manager = get_memory_manager()
results = []
if memory_manager and memory_manager._initialized and memory_manager.graph_store:
# 从 memory_manager 搜索
all_memories = memory_manager.graph_store.get_all_memories()
for memory in all_memories:
if q.lower() in memory.to_text().lower():
results.append(
{
"id": memory.id,
"type": memory.memory_type.value,
"importance": memory.importance,
"text": memory.to_text(),
}
)
else:
# 从文件加载的数据中搜索 (降级方案)
data = load_graph_data_from_file()
for memory in data.get("memories", []):
if q.lower() in memory.get("text", "").lower():
results.append(memory)
return JSONResponse(
content={
"success": True,
"data": {
"results": results[:limit],
"count": len(results),
},
}
)
except Exception as e:
return JSONResponse(content={"success": False, "error": str(e)}, status_code=500)
@router.get("/api/stats")
async def get_statistics():
"""获取统计信息"""
try:
data = load_graph_data_from_file()
node_types = {}
memory_types = {}
for node in data["nodes"]:
node_type = node.get("type", "Unknown")
node_types[node_type] = node_types.get(node_type, 0) + 1
for memory in data.get("memories", []):
mem_type = memory.get("type", "Unknown")
memory_types[mem_type] = memory_types.get(mem_type, 0) + 1
stats = data.get("stats", {})
stats["node_types"] = node_types
stats["memory_types"] = memory_types
return JSONResponse(content={"success": True, "data": stats})
except Exception as e:
return JSONResponse(content={"success": False, "error": str(e)}, status_code=500)

111
src/api/statistic_router.py
View File

@@ -4,10 +4,10 @@ from typing import Any, Literal
from fastapi import APIRouter, HTTPException, Query
from src.common.database.compatibility import db_get
from src.common.database.core.models import LLMUsage
from src.chat.utils.statistic import (
StatisticOutputTask,
)
from src.common.logger import get_logger
from src.config.config import model_config
logger = get_logger("LLM统计API")
@@ -37,108 +37,6 @@ COST_BY_USER = "costs_by_user"
COST_BY_MODEL = "costs_by_model"
COST_BY_MODULE = "costs_by_module"
async def _collect_stats_in_period(start_time: datetime, end_time: datetime) -> dict[str, Any]:
"""在指定时间段内收集LLM使用统计信息"""
records = await db_get(
model_class=LLMUsage,
filters={"timestamp": {"$gte": start_time, "$lt": end_time}},
)
if not records:
return {}
# 创建一个从 model_identifier 到 name 的映射
model_identifier_to_name_map = {model.model_identifier: model.name for model in model_config.models}
stats: dict[str, Any] = {
TOTAL_REQ_CNT: 0,
TOTAL_COST: 0.0,
REQ_CNT_BY_TYPE: defaultdict(int),
REQ_CNT_BY_USER: defaultdict(int),
REQ_CNT_BY_MODEL: defaultdict(int),
REQ_CNT_BY_MODULE: defaultdict(int),
IN_TOK_BY_TYPE: defaultdict(int),
IN_TOK_BY_USER: defaultdict(int),
IN_TOK_BY_MODEL: defaultdict(int),
IN_TOK_BY_MODULE: defaultdict(int),
OUT_TOK_BY_TYPE: defaultdict(int),
OUT_TOK_BY_USER: defaultdict(int),
OUT_TOK_BY_MODEL: defaultdict(int),
OUT_TOK_BY_MODULE: defaultdict(int),
TOTAL_TOK_BY_TYPE: defaultdict(int),
TOTAL_TOK_BY_USER: defaultdict(int),
TOTAL_TOK_BY_MODEL: defaultdict(int),
TOTAL_TOK_BY_MODULE: defaultdict(int),
COST_BY_TYPE: defaultdict(float),
COST_BY_USER: defaultdict(float),
COST_BY_MODEL: defaultdict(float),
COST_BY_MODULE: defaultdict(float),
}
for record in records:
if not isinstance(record, dict):
continue
stats[TOTAL_REQ_CNT] += 1
request_type = record.get("request_type") or "unknown"
user_id = record.get("user_id") or "unknown"
# 从数据库获取的是真实模型名 (model_identifier)
real_model_name = record.get("model_name") or "unknown"
module_name = request_type.split(".")[0] if "." in request_type else request_type
# 尝试通过真实模型名找到配置文件中的模型名
config_model_name = model_identifier_to_name_map.get(real_model_name, real_model_name)
prompt_tokens = record.get("prompt_tokens") or 0
completion_tokens = record.get("completion_tokens") or 0
total_tokens = prompt_tokens + completion_tokens
cost = 0.0
try:
# 使用配置文件中的模型名来获取模型信息
model_info = model_config.get_model_info(config_model_name)
if model_info:
input_cost = (prompt_tokens / 1000000) * model_info.price_in
output_cost = (completion_tokens / 1000000) * model_info.price_out
cost = round(input_cost + output_cost, 6)
except KeyError as e:
logger.info(str(e))
logger.warning(f"模型 '{config_model_name}' (真实名称: '{real_model_name}') 在配置中未找到，成本计算将使用默认值 0.0")
stats[TOTAL_COST] += cost
# 按类型统计
stats[REQ_CNT_BY_TYPE][request_type] += 1
stats[IN_TOK_BY_TYPE][request_type] += prompt_tokens
stats[OUT_TOK_BY_TYPE][request_type] += completion_tokens
stats[TOTAL_TOK_BY_TYPE][request_type] += total_tokens
stats[COST_BY_TYPE][request_type] += cost
# 按用户统计
stats[REQ_CNT_BY_USER][user_id] += 1
stats[IN_TOK_BY_USER][user_id] += prompt_tokens
stats[OUT_TOK_BY_USER][user_id] += completion_tokens
stats[TOTAL_TOK_BY_USER][user_id] += total_tokens
stats[COST_BY_USER][user_id] += cost
# 按模型统计 (使用配置文件中的名称)
stats[REQ_CNT_BY_MODEL][config_model_name] += 1
stats[IN_TOK_BY_MODEL][config_model_name] += prompt_tokens
stats[OUT_TOK_BY_MODEL][config_model_name] += completion_tokens
stats[TOTAL_TOK_BY_MODEL][config_model_name] += total_tokens
stats[COST_BY_MODEL][config_model_name] += cost
# 按模块统计
stats[REQ_CNT_BY_MODULE][module_name] += 1
stats[IN_TOK_BY_MODULE][module_name] += prompt_tokens
stats[OUT_TOK_BY_MODULE][module_name] += completion_tokens
stats[TOTAL_TOK_BY_MODULE][module_name] += total_tokens
stats[COST_BY_MODULE][module_name] += cost
return stats
@router.get("/llm/stats")
async def get_llm_stats(
period_type: Literal[
@@ -179,7 +77,8 @@ async def get_llm_stats(
if start_time is None:
raise HTTPException(status_code=400, detail="无法确定查询的起始时间")
period_stats = await _collect_stats_in_period(start_time, end_time)
stats_data = await StatisticOutputTask._collect_model_request_for_period([("custom", start_time)])
period_stats = stats_data.get("custom", {})
if not period_stats:
return {"period": {"start": start_time.isoformat(), "end": end_time.isoformat()}, "data": {}}

8
tools/memory_visualizer/templates/visualizer.html → src/api/templates/visualizer.html
View File

@@ -658,7 +658,7 @@
try {
document.getElementById('loading').style.display = 'block';
const response = await fetch('/api/graph/full');
const response = await fetch('/visualizer/api/graph/full');
const result = await response.json();
if (result.success) {
@@ -748,7 +748,7 @@
}
try {
const response = await fetch(`/api/search?q=${encodeURIComponent(query)}&limit=50`);
const response = await fetch(`/visualizer/api/search?q=${encodeURIComponent(query)}&limit=50`);
const result = await response.json();
if (result.success) {
@@ -1041,7 +1041,7 @@
// 文件选择功能
async function loadFileList() {
try {
const response = await fetch('/api/files');
const response = await fetch('/visualizer/api/files');
const result = await response.json();
if (result.success) {
@@ -1130,7 +1130,7 @@
document.getElementById('loading').style.display = 'block';
closeFileSelector();
const response = await fetch('/api/select_file', {
const response = await fetch('/visualizer/api/select_file', {
method: 'POST',
headers: {
'Content-Type': 'application/json'

3
src/common/server.py
View File

@@ -2,7 +2,8 @@ import os
import socket
from fastapi import APIRouter, FastAPI
from fastapi.middleware.cors import CORSMiddleware # 新增导入
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from rich.traceback import install
from uvicorn import Config
from uvicorn import Server as UvicornServer

3
src/main.py
View File

@@ -423,15 +423,16 @@ MoFox_Bot(第三方修改版)
# 注册API路由
try:
from src.api.memory_visualizer_router import router as visualizer_router
from src.api.message_router import router as message_router
from src.api.statistic_router import router as llm_statistic_router
self.server.register_router(message_router, prefix="/api")
self.server.register_router(llm_statistic_router, prefix="/api")
self.server.register_router(visualizer_router, prefix="/visualizer")
logger.info("API路由注册成功")
except Exception as e:
logger.error(f"注册API路由失败: {e}")
# 初始化统一调度器
try:
from src.schedule.unified_scheduler import initialize_scheduler

108
tools/memory_visualizer/CHANGELOG.md
View File

@@ -1,108 +0,0 @@
# 🔄 更新日志 - 记忆图可视化工具
## v1.1 - 2025-11-06
### ✨ 新增功能
1. **📂 文件选择器**
- 自动搜索所有可用的记忆图数据文件
- 支持在Web界面中切换不同的数据文件
- 显示文件大小、修改时间等信息
- 高亮显示当前使用的文件
2. **🔍 智能文件搜索**
- 自动查找 `data/memory_graph/graph_store.json`
- 搜索所有备份文件 `graph_store_*.json`
- 搜索 `data/backup/` 目录下的历史数据
- 按修改时间排序,自动使用最新文件
3. **📊 增强的文件信息显示**
- 在侧边栏显示当前文件信息
- 包含文件名、大小、修改时间
- 实时更新,方便追踪
### 🔧 改进
- 更友好的错误提示
- 无数据文件时显示引导信息
- 优化用户体验
### 🎯 使用方法
```bash
# 启动可视化工具
python run_visualizer_simple.py
# 或直接运行
python tools/memory_visualizer/visualizer_simple.py
```
在Web界面中:
1. 点击侧边栏的 "选择文件" 按钮
2. 浏览所有可用的数据文件
3. 点击任意文件切换数据源
4. 图形会自动重新加载
### 📸 新界面预览
侧边栏新增:
```
┌─────────────────────────┐
│ 📂 数据文件 │
│ ┌──────────┬──────────┐ │
│ │ 选择文件 │ 刷新列表 │ │
│ └──────────┴──────────┘ │
│ ┌─────────────────────┐ │
│ │ 📄 graph_store.json │ │
│ │ 大小: 125 KB │ │
│ │ 修改: 2025-11-06 │ │
│ └─────────────────────┘ │
└─────────────────────────┘
```
文件选择对话框:
```
┌────────────────────────────────┐
│ 📂 选择数据文件 [×] │
├────────────────────────────────┤
│ ┌────────────────────────────┐ │
│ │ 📄 graph_store.json [当前] │ │
│ │ 125 KB | 2025-11-06 09:30 │ │
│ └────────────────────────────┘ │
│ ┌────────────────────────────┐ │
│ │ 📄 graph_store_backup.json │ │
│ │ 120 KB | 2025-11-05 18:00 │ │
│ └────────────────────────────┘ │
└────────────────────────────────┘
```
---
## v1.0 - 2025-11-06 (初始版本)
### 🎉 首次发布
- ✅ 基于Vis.js的交互式图形可视化
- ✅ 节点类型颜色分类
- ✅ 搜索和过滤功能
- ✅ 统计信息显示
- ✅ 节点详情查看
- ✅ 数据导出功能
- ✅ 独立版服务器(快速启动)
- ✅ 完整版服务器(实时数据)
---
## 🔮 计划中的功能 (v1.2+)
- [ ] 时间轴视图 - 查看记忆随时间的变化
- [ ] 3D可视化模式
- [ ] 记忆重要性热力图
- [ ] 关系强度可视化
- [ ] 导出为图片/PDF
- [ ] 记忆路径追踪
- [ ] 多文件对比视图
- [ ] 性能优化 - 支持更大规模图形
- [ ] 移动端适配
欢迎提出建议和需求! 🚀

163
tools/memory_visualizer/FILE_ORGANIZATION.md
View File

@@ -1,163 +0,0 @@
# 📁 可视化工具文件整理完成
## ✅ 整理结果
### 新的目录结构
```
tools/memory_visualizer/
├── visualizer.ps1 ⭐ 统一启动脚本（主入口）
├── visualizer_simple.py # 独立版服务器
├── visualizer_server.py # 完整版服务器
├── generate_sample_data.py # 测试数据生成器
├── test_visualizer.py # 测试脚本
├── run_visualizer.py # Python 运行脚本（独立版）
├── run_visualizer_simple.py # Python 运行脚本（简化版）
├── start_visualizer.bat # Windows 批处理启动脚本
├── start_visualizer.ps1 # PowerShell 启动脚本
├── start_visualizer.sh # Linux/Mac 启动脚本
├── requirements.txt # Python 依赖
├── templates/ # HTML 模板
│ └── visualizer.html # 可视化界面
├── docs/ # 文档目录
│ ├── VISUALIZER_README.md
│ ├── VISUALIZER_GUIDE.md
│ └── VISUALIZER_INSTALL_COMPLETE.md
├── README.md # 主说明文档
├── QUICKSTART.md # 快速开始指南
└── CHANGELOG.md # 更新日志
```
### 根目录保留文件
```
项目根目录/
├── visualizer.ps1 # 快捷启动脚本（指向 tools/memory_visualizer/visualizer.ps1）
└── tools/memory_visualizer/ # 所有可视化工具文件
```
## 🚀 使用方法
### 推荐方式：使用统一启动脚本
```powershell
# 在项目根目录
.\visualizer.ps1
# 或在工具目录
cd tools\memory_visualizer
.\visualizer.ps1
```
### 命令行参数
```powershell
# 直接启动独立版（推荐）
.\visualizer.ps1 -Simple
# 启动完整版
.\visualizer.ps1 -Full
# 生成测试数据
.\visualizer.ps1 -Generate
# 运行测试
.\visualizer.ps1 -Test
```
## 📋 整理内容
### 已移动的文件
从项目根目录移动到 `tools/memory_visualizer/`：
1. **脚本文件**
- `generate_sample_data.py`
- `run_visualizer.py`
- `run_visualizer_simple.py`
- `test_visualizer.py`
- `start_visualizer.bat`
- `start_visualizer.ps1`
- `start_visualizer.sh`
- `visualizer.ps1`
2. **文档文件** → `docs/` 子目录
- `VISUALIZER_GUIDE.md`
- `VISUALIZER_INSTALL_COMPLETE.md`
- `VISUALIZER_README.md`
### 已创建的新文件
1. **统一启动脚本**
- `tools/memory_visualizer/visualizer.ps1` - 功能齐全的统一入口
2. **快捷脚本**
- `visualizer.ps1`（根目录）- 快捷方式，指向实际脚本
3. **更新的文档**
- `tools/memory_visualizer/README.md` - 更新为反映新结构
## 🎯 优势
### 整理前的问题
- ❌ 文件散落在根目录
- ❌ 多个启动脚本功能重复
- ❌ 文档分散不便管理
- ❌ 不清楚哪个是主入口
### 整理后的改进
- ✅ 所有文件集中在 `tools/memory_visualizer/`
- ✅ 单一统一的启动脚本 `visualizer.ps1`
- ✅ 文档集中在 `docs/` 子目录
- ✅ 清晰的主入口和快捷方式
- ✅ 更好的可维护性
## 📝 功能对比
### 旧的方式（整理前）
```powershell
# 需要记住多个脚本名称
.\start_visualizer.ps1
.\run_visualizer.py
.\run_visualizer_simple.py
.\generate_sample_data.py
```
### 新的方式（整理后）
```powershell
# 只需要一个统一的脚本
.\visualizer.ps1 # 交互式菜单
.\visualizer.ps1 -Simple # 启动独立版
.\visualizer.ps1 -Generate # 生成数据
.\visualizer.ps1 -Test # 运行测试
```
## 🔧 维护说明
### 添加新功能
1. 在 `tools/memory_visualizer/` 目录下添加新文件
2. 如需启动选项，在 `visualizer.ps1` 中添加新参数
3. 更新 `README.md` 文档
### 更新文档
1. 主文档：`tools/memory_visualizer/README.md`
2. 详细文档：`tools/memory_visualizer/docs/`
## ✅ 测试结果
- ✅ 统一启动脚本正常工作
- ✅ 独立版服务器成功启动（端口 5001）
- ✅ 数据加载成功（725 节点，769 边）
- ✅ Web 界面正常访问
- ✅ 所有文件已整理到位
## 📚 相关文档
- [README](tools/memory_visualizer/README.md) - 主要说明文档
- [QUICKSTART](tools/memory_visualizer/QUICKSTART.md) - 快速开始指南
- [CHANGELOG](tools/memory_visualizer/CHANGELOG.md) - 更新日志
- [详细指南](tools/memory_visualizer/docs/VISUALIZER_GUIDE.md) - 完整使用指南
---
整理完成时间：2025-11-06

279
tools/memory_visualizer/QUICKSTART.md
View File

@@ -1,279 +0,0 @@
# 记忆图可视化工具 - 快速入门指南
## 🎯 方案选择
我为你创建了**两个版本**的可视化工具:
### 1️⃣ 独立版 (推荐 ⭐)
- **文件**: `tools/memory_visualizer/visualizer_simple.py`
- **优点**:
- 直接读取存储文件,无需初始化完整系统
- 启动快速
- 占用资源少
- **适用**: 快速查看已有记忆数据
### 2️⃣ 完整版
- **文件**: `tools/memory_visualizer/visualizer_server.py`
- **优点**:
- 实时数据
- 支持更多功能
- **缺点**:
- 需要完整初始化记忆管理器
- 启动较慢
## 🚀 快速开始
### 步骤 1: 安装依赖
**Windows (PowerShell):**
```powershell
# 依赖会自动检查和安装
.\start_visualizer.ps1
```
**Windows (CMD):**
```cmd
start_visualizer.bat
```
**Linux/Mac:**
```bash
chmod +x start_visualizer.sh
./start_visualizer.sh
```
**手动安装依赖:**
```bash
# 使用虚拟环境
.\.venv\Scripts\python.exe -m pip install flask flask-cors
# 或全局安装
pip install flask flask-cors
```
### 步骤 2: 确保有数据
如果还没有记忆数据,可以:
**选项A**: 运行Bot生成实际数据
```bash
python bot.py
# 与Bot交互一会儿,让它积累一些记忆
```
**选项B**: 生成测试数据 (如果测试脚本可用)
```bash
python test_visualizer.py
# 选择选项 1: 生成测试数据
```
### 步骤 3: 启动可视化服务器
**方式一: 使用启动脚本 (推荐 ⭐)**
Windows PowerShell:
```powershell
.\start_visualizer.ps1
```
Windows CMD:
```cmd
start_visualizer.bat
```
Linux/Mac:
```bash
./start_visualizer.sh
```
**方式二: 手动启动**
使用虚拟环境:
```bash
# Windows
.\.venv\Scripts\python.exe tools/memory_visualizer/visualizer_simple.py
# Linux/Mac
.venv/bin/python tools/memory_visualizer/visualizer_simple.py
```
或使用系统Python:
```bash
python tools/memory_visualizer/visualizer_simple.py
```
服务器将在 http://127.0.0.1:5001 启动
### 步骤 4: 打开浏览器
访问对应的地址,开始探索记忆图! 🎉
## 🎨 界面功能
### 左侧栏
1. **🔍 搜索框**
- 输入关键词搜索相关记忆
- 结果会在图中高亮显示
2. **📊 统计信息**
- 节点总数
- 边总数
- 记忆总数
- 图密度
3. **🎨 节点类型图例**
- 🔴 主体 (SUBJECT) - 记忆的主语
- 🔵 主题 (TOPIC) - 动作或状态
- 🟢 客体 (OBJECT) - 宾语
- 🟠 属性 (ATTRIBUTE) - 延伸属性
- 🟣 值 (VALUE) - 属性的具体值
4. **🔧 过滤器**
- 勾选/取消勾选来显示/隐藏特定类型的节点
- 实时更新图形
5. **ℹ️ 节点信息**
- 点击任意节点查看详细信息
- 显示节点类型、内容、创建时间等
### 右侧主区域
1. **控制按钮**
- 🔄 刷新图形: 重新加载最新数据
- 📐 适应窗口: 自动调整图形大小
- 💾 导出数据: 下载JSON格式的图数据
2. **交互式图形**
- **拖动节点**: 点击并拖动单个节点
- **拖动画布**: 按住空白处拖动整个图形
- **缩放**: 使用鼠标滚轮放大/缩小
- **点击节点**: 查看详细信息
- **物理模拟**: 节点会自动排列,避免重叠
## 🎮 操作技巧
### 查看特定类型的节点
1. 在左侧过滤器中取消勾选不需要的类型
2. 图形会自动更新,只显示选中的类型
### 查找特定记忆
1. 在搜索框输入关键词(如: "小明", "吃饭")
2. 点击"搜索"按钮
3. 相关节点会被选中并自动聚焦
### 整理混乱的图形
1. 点击"适应窗口"按钮
2. 或者刷新页面重新初始化布局
### 导出数据进行分析
1. 点击"导出数据"按钮
2. JSON文件会自动下载
3. 可以用于进一步的数据分析或备份
## 🎯 示例场景
### 场景1: 了解记忆图整体结构
1. 启动可视化工具
2. 观察不同颜色的节点分布
3. 查看统计信息了解数量
4. 使用过滤器逐个类型查看
### 场景2: 追踪特定主题的记忆
1. 在搜索框输入主题关键词(如: "学习")
2. 点击搜索
3. 查看高亮的相关节点
4. 点击节点查看详情
### 场景3: 调试记忆系统
1. 创建一条新记忆
2. 刷新可视化页面
3. 查看新节点和边是否正确创建
4. 验证节点类型和关系
## 🐛 常见问题
### Q: 页面显示空白或没有数据?
**A**:
1. 检查是否有记忆数据: 查看 `data/memory_graph/` 目录
2. 确保记忆系统已启用: 检查 `config/bot_config.toml` 中 `[memory] enable = true`
3. 尝试生成一些测试数据
### Q: 节点太多,看不清楚?
**A**:
1. 使用过滤器只显示某些类型
2. 使用搜索功能定位特定节点
3. 调整浏览器窗口大小,点击"适应窗口"
### Q: 如何更新数据?
**A**:
- **独立版**: 点击"刷新图形"或访问 `/api/reload`
- **完整版**: 点击"刷新图形"会自动加载最新数据
### Q: 端口被占用怎么办?
**A**: 修改启动脚本中的端口号:
```python
run_server(host='127.0.0.1', port=5002, debug=True) # 改为其他端口
```
## 🎨 自定义配置
### 修改节点颜色
编辑 `templates/visualizer.html`,找到:
```javascript
const nodeColors = {
'SUBJECT': '#FF6B6B', // 改为你喜欢的颜色
'TOPIC': '#4ECDC4',
// ...
};
```
### 修改物理引擎参数
在同一文件中找到 `physics` 配置:
```javascript
physics: {
barnesHut: {
gravitationalConstant: -8000, // 调整引力
springLength: 150, // 调整弹簧长度
// ...
}
}
```
### 修改数据加载限制
编辑对应的服务器文件,修改 `get_all_memories()` 的limit参数。
## 📝 文件结构
```
tools/memory_visualizer/
├── README.md # 详细文档
├── requirements.txt # 依赖列表
├── visualizer_server.py # 完整版服务器
├── visualizer_simple.py # 独立版服务器 ⭐
└── templates/
└── visualizer.html # Web界面模板
run_visualizer.py # 快速启动脚本
test_visualizer.py # 测试和演示脚本
```
## 🚀 下一步
现在你可以:
1. ✅ 启动可视化工具查看现有数据
2. ✅ 与Bot交互生成更多记忆
3. ✅ 使用可视化工具验证记忆结构
4. ✅ 根据需要自定义样式和配置
祝你使用愉快! 🎉
---
如有问题,请查看 `tools/memory_visualizer/README.md` 获取更多帮助。

201
tools/memory_visualizer/README.md
View File

@@ -1,201 +0,0 @@
# 🦊 记忆图可视化工具
一个交互式的 Web 可视化工具，用于查看和分析 MoFox Bot 的记忆图结构。
## 📁 目录结构
```
tools/memory_visualizer/
├── visualizer.ps1 # 统一启动脚本（主入口）⭐
├── visualizer_simple.py # 独立版服务器（推荐）
├── visualizer_server.py # 完整版服务器
├── generate_sample_data.py # 测试数据生成器
├── test_visualizer.py # 测试脚本
├── requirements.txt # Python 依赖
├── templates/ # HTML 模板
│ └── visualizer.html # 可视化界面
├── docs/ # 文档目录
│ ├── VISUALIZER_README.md
│ ├── VISUALIZER_GUIDE.md
│ └── VISUALIZER_INSTALL_COMPLETE.md
├── README.md # 本文件
├── QUICKSTART.md # 快速开始指南
└── CHANGELOG.md # 更新日志
```
## 🚀 快速开始
### 方式 1：交互式菜单（推荐）
```powershell
# 在项目根目录运行
.\visualizer.ps1
# 或在工具目录运行
cd tools\memory_visualizer
.\visualizer.ps1
```
### 方式 2：命令行参数
```powershell
# 启动独立版（推荐，快速）
.\visualizer.ps1 -Simple
# 启动完整版（需要 MemoryManager）
.\visualizer.ps1 -Full
# 生成测试数据
.\visualizer.ps1 -Generate
# 运行测试
.\visualizer.ps1 -Test
# 查看帮助
.\visualizer.ps1 -Help
```
## 📊 两个版本的区别
### 独立版（Simple）- 推荐
- ✅ **快速启动**：直接读取数据文件，无需初始化 MemoryManager
- ✅ **轻量级**：只依赖 Flask 和 vis.js
- ✅ **稳定**：不依赖主系统运行状态
- 📌 **端口**：5001
- 📁 **数据源**：`data/memory_graph/*.json`
### 完整版（Full）
- 🔄 **实时数据**：使用 MemoryManager 获取最新数据
- 🔌 **集成**：与主系统深度集成
- ⚡ **功能完整**：支持所有高级功能
- 📌 **端口**：5000
- 📁 **数据源**：MemoryManager
## ✨ 主要功能
1. **交互式图形可视化**
- 🎨 5 种节点类型（主体、主题、客体、属性、值）
- 🔗 完整路径高亮显示
- 🔍 点击节点查看连接关系
- 📐 自动布局和缩放
2. **高级筛选**
- ☑️ 按节点类型筛选
- 🔎 关键词搜索
- 📊 统计信息实时更新
3. **智能高亮**
- 💡 点击节点高亮所有连接路径（递归探索）
- 👻 无关节点变为半透明
- 🎯 自动聚焦到相关子图
4. **物理引擎优化**
- 🚀 智能布局算法
- ⏱️ 自动停止防止持续运行
- 🔄 筛选后自动重新布局
5. **数据管理**
- 📂 多文件选择器
- 💾 导出图形数据
- 🔄 实时刷新
## 🔧 依赖安装
脚本会自动检查并安装依赖，也可以手动安装：
```powershell
# 激活虚拟环境
.\.venv\Scripts\Activate.ps1
# 安装依赖
pip install -r tools/memory_visualizer/requirements.txt
```
**所需依赖：**
- Flask >= 2.3.0
- flask-cors >= 4.0.0
## 📖 使用说明
### 1. 查看记忆图
1. 启动服务器（推荐独立版）
2. 在浏览器打开 http://127.0.0.1:5001
3. 等待数据加载完成
### 2. 探索连接关系
1. **点击节点**：查看与该节点相关的所有连接路径
2. **点击空白处**：恢复所有节点显示
3. **使用筛选器**：按类型过滤节点
### 3. 搜索记忆
1. 在搜索框输入关键词
2. 点击搜索按钮
3. 相关节点会自动高亮
### 4. 查看统计
- 左侧面板显示实时统计信息
- 节点数、边数、记忆数
- 图密度等指标
## 🎨 节点颜色说明
- 🔴 **主体（SUBJECT）**：红色 (#FF6B6B)
- 🔵 **主题（TOPIC）**：青色 (#4ECDC4)
- 🟦 **客体（OBJECT）**：蓝色 (#45B7D1)
- 🟠 **属性（ATTRIBUTE）**：橙色 (#FFA07A)
- 🟢 **值（VALUE）**：绿色 (#98D8C8)
## 🐛 常见问题
### 问题 1：没有数据显示
**解决方案：**
1. 检查 `data/memory_graph/` 目录是否存在数据文件
2. 运行 `.\visualizer.ps1 -Generate` 生成测试数据
3. 确保 Bot 已经运行过并生成了记忆数据
### 问题 2：物理引擎一直运行
**解决方案：**
- 新版本已修复此问题
- 物理引擎会在稳定后自动停止（最多 5 秒）
### 问题 3：筛选后节点排版错乱
**解决方案：**
- 新版本已修复此问题
- 筛选后会自动重新布局
### 问题 4：无法查看完整连接路径
**解决方案：**
- 新版本使用 BFS 算法递归探索所有连接
- 点击节点即可查看完整路径
## 📝 开发说明
### 添加新功能
1. 编辑 `visualizer_simple.py` 或 `visualizer_server.py`
2. 修改 `templates/visualizer.html` 更新界面
3. 更新 `requirements.txt` 添加新依赖
4. 运行测试：`.\visualizer.ps1 -Test`
### 调试
```powershell
# 启动 Flask 调试模式
$env:FLASK_DEBUG = "1"
python tools/memory_visualizer/visualizer_simple.py
```
## 📚 相关文档
- [快速开始指南](QUICKSTART.md)
- [更新日志](CHANGELOG.md)
- [详细使用指南](docs/VISUALIZER_GUIDE.md)
## 🆘 获取帮助
遇到问题？
1. 查看 [常见问题](#常见问题)
2. 运行 `.\visualizer.ps1 -Help` 查看帮助
3. 查看项目文档目录
## 📄 许可证
与 MoFox Bot 主项目相同

163
tools/memory_visualizer/README.md.bak
View File

@@ -1,163 +0,0 @@
# 🦊 MoFox Bot 记忆图可视化工具
这是一个交互式的Web界面,用于可视化和探索MoFox Bot的记忆图结构。
## ✨ 功能特性
- **交互式图形可视化**: 使用Vis.js展示节点和边的关系
- **实时数据**: 直接从记忆管理器读取最新数据
- **节点类型分类**: 不同颜色区分不同类型的节点
- 🔴 主体 (SUBJECT)
- 🔵 主题 (TOPIC)
- 🟢 客体 (OBJECT)
- 🟠 属性 (ATTRIBUTE)
- 🟣 值 (VALUE)
- **搜索功能**: 快速查找相关记忆
- **过滤器**: 按节点类型过滤显示
- **统计信息**: 实时显示图的统计数据
- **节点详情**: 点击节点查看详细信息
- **自由缩放拖动**: 支持图形的交互式操作
- **数据导出**: 导出当前图形数据为JSON
## 🚀 快速开始
### 1. 安装依赖
```bash
pip install flask flask-cors
```
### 2. 启动服务器
在项目根目录运行:
```bash
python tools/memory_visualizer/visualizer_server.py
```
或者使用便捷脚本:
```bash
python run_visualizer.py
```
### 3. 打开浏览器
访问: http://127.0.0.1:5000
## 📊 界面说明
### 主界面布局
```
┌─────────────────────────────────────────────────┐
│ 侧边栏 │ 主内容区 │
│ - 搜索框 │ - 控制按钮 │
│ - 统计信息 │ - 图形显示 │
│ - 节点类型图例 │ │
│ - 过滤器 │ │
│ - 节点详情 │ │
└─────────────────────────────────────────────────┘
```
### 操作说明
- **🔍 搜索**: 在搜索框输入关键词,点击"搜索"按钮查找相关记忆
- **🔄 刷新图形**: 重新加载最新的记忆图数据
- **📐 适应窗口**: 自动调整图形大小以适应窗口
- **💾 导出数据**: 将当前图形数据导出为JSON文件
- **✅ 过滤器**: 勾选/取消勾选不同类型的节点来过滤显示
- **👆 点击节点**: 点击任意节点查看详细信息
- **🖱️ 拖动**: 按住鼠标拖动节点或整个图形
- **🔍 缩放**: 使用鼠标滚轮缩放图形
## 🔧 配置说明
### 修改服务器配置
在 `visualizer_server.py` 的最后:
```python
if __name__ == '__main__':
run_server(
host='127.0.0.1', # 监听地址
port=5000, # 端口号
debug=True # 调试模式
)
```
### API端点
- `GET /` - 主页面
- `GET /api/graph/full` - 获取完整记忆图数据
- `GET /api/memory/<memory_id>` - 获取特定记忆详情
- `GET /api/search?q=<query>&limit=<n>` - 搜索记忆
- `GET /api/stats` - 获取统计信息
## 📝 技术栈
- **后端**: Flask (Python Web框架)
- **前端**:
- Vis.js (图形可视化库)
- 原生JavaScript
- CSS3 (渐变、动画、响应式布局)
- **数据**: 直接从MoFox Bot记忆管理器读取
## 🐛 故障排除
### 问题: 无法启动服务器
**原因**: 记忆系统未启用或配置错误
**解决**: 检查 `config/bot_config.toml` 确保:
```toml
[memory]
enable = true
data_dir = "data/memory_graph"
```
### 问题: 图形显示空白
**原因**: 没有记忆数据
**解决**:
1. 先运行Bot让其生成一些记忆
2. 或者运行测试脚本生成测试数据
### 问题: 节点太多,图形混乱
**解决**:
1. 使用过滤器只显示某些类型的节点
2. 使用搜索功能定位特定记忆
3. 调整物理引擎参数(在visualizer.html中)
## 🎨 自定义样式
修改 `templates/visualizer.html` 中的样式定义:
```javascript
const nodeColors = {
'SUBJECT': '#FF6B6B', // 主体颜色
'TOPIC': '#4ECDC4', // 主题颜色
'OBJECT': '#45B7D1', // 客体颜色
'ATTRIBUTE': '#FFA07A', // 属性颜色
'VALUE': '#98D8C8' // 值颜色
};
```
## 📈 性能优化
对于大型图形(>1000节点):
1. **禁用物理引擎**: 在stabilization完成后自动禁用
2. **限制显示节点**: 使用过滤器或搜索
3. **分页加载**: 修改API使用分页
## 🤝 贡献
欢迎提交Issue和Pull Request!
## 📄 许可
与MoFox Bot主项目相同的许可证

210
tools/memory_visualizer/docs/VISUALIZER_INSTALL_COMPLETE.md
View File

@@ -1,210 +0,0 @@
# ✅ 记忆图可视化工具 - 安装完成
## 🎉 恭喜！可视化工具已成功创建！
---
## 📦 已创建的文件
```
Bot/
├── visualizer.ps1 ⭐⭐⭐ # 统一启动脚本 (推荐使用)
├── start_visualizer.ps1 # 独立版快速启动
├── start_visualizer.bat # CMD版启动脚本
├── generate_sample_data.py # 示例数据生成器
├── VISUALIZER_README.md ⭐ # 快速参考指南
├── VISUALIZER_GUIDE.md # 完整使用指南
└── tools/memory_visualizer/
├── visualizer_simple.py ⭐ # 独立版服务器 (推荐)
├── visualizer_server.py # 完整版服务器
├── README.md # 详细文档
├── QUICKSTART.md # 快速入门
├── CHANGELOG.md # 更新日志
└── templates/
└── visualizer.html ⭐ # 精美Web界面
```
---
## 🚀 立即开始 (3秒)
### 方法 1: 使用统一启动脚本 (最简单 ⭐⭐⭐)
```powershell
.\visualizer.ps1
```
然后按提示选择:
- **1** = 独立版 (推荐，快速)
- **2** = 完整版 (实时数据)
- **3** = 生成示例数据
### 方法 2: 直接启动
```powershell
# 如果还没有数据，先生成
.\.venv\Scripts\python.exe generate_sample_data.py
# 启动可视化
.\start_visualizer.ps1
# 打开浏览器
# http://127.0.0.1:5001
```
---
## 🎨 功能亮点
### ✨ 核心功能
- 🎯 **交互式图形**: 拖动、缩放、点击
- 🎨 **颜色分类**: 5种节点类型自动上色
- 🔍 **智能搜索**: 快速定位相关记忆
- 🔧 **灵活过滤**: 按节点类型筛选
- 📊 **实时统计**: 节点、边、记忆数量
- 💾 **数据导出**: JSON格式导出
### 📂 独立版特色 (推荐)
- ⚡ **秒速启动**: 2秒内完成
- 📁 **文件切换**: 浏览所有历史数据
- 🔄 **自动搜索**: 智能查找数据文件
- 💚 **低资源**: 占用资源极少
### 🔥 完整版特色
- 🔴 **实时数据**: 与Bot同步
- 🔄 **自动更新**: 无需刷新
- 🛠️ **完整功能**: 使用全部API
---
## 📊 界面预览
```
┌─────────────────────────────────────────────────────────┐
│ 侧边栏 │ 主区域 │
│ ┌─────────────────────┐ │ ┌───────────────────────┐ │
│ │ 📂 数据文件 │ │ │ 🔄 📐 💾 控制按钮 │ │
│ │ [选择] [刷新] │ │ └───────────────────────┘ │
│ │ 📄 当前: xxx.json │ │ ┌───────────────────────┐ │
│ └─────────────────────┘ │ │ │ │
│ │ │ 交互式图形可视化 │ │
│ ┌─────────────────────┐ │ │ │ │
│ │ 🔍 搜索记忆 │ │ │ 🔴 主体 🔵 主题 │ │
│ │ [...........] [搜索] │ │ │ 🟢 客体 🟠 属性 │ │
│ └─────────────────────┘ │ │ 🟣 值 │ │
│ │ │ │ │
│ 📊 统计: 12节点 15边 │ │ 可拖动、缩放、点击 │ │
│ │ │ │ │
│ 🎨 节点类型图例 │ └───────────────────────┘ │
│ 🔧 过滤器 │ │
│ ℹ️ 节点信息 │ │
└─────────────────────────────────────────────────────────┘
```
---
## 🎯 快速命令
```powershell
# 统一启动 (推荐)
.\visualizer.ps1
# 生成示例数据
.\.venv\Scripts\python.exe generate_sample_data.py
# 独立版 (端口 5001)
.\start_visualizer.ps1
# 完整版 (端口 5000)
.\.venv\Scripts\python.exe tools/memory_visualizer/visualizer_server.py
```
---
## 📖 文档索引
### 快速参考 (必读 ⭐)
- **VISUALIZER_README.md** - 快速参考卡片
- **VISUALIZER_GUIDE.md** - 完整使用指南
### 详细文档
- **tools/memory_visualizer/README.md** - 技术文档
- **tools/memory_visualizer/QUICKSTART.md** - 快速入门
- **tools/memory_visualizer/CHANGELOG.md** - 版本历史
---
## 💡 使用建议
### 🎯 对于首次使用者
1. 运行 `.\visualizer.ps1`
2. 选择 `3` 生成示例数据
3. 选择 `1` 启动独立版
4. 打开浏览器访问 http://127.0.0.1:5001
5. 开始探索！
### 🔧 对于开发者
1. 运行Bot积累真实数据
2. 启动完整版可视化: `.\visualizer.ps1` → `2`
3. 实时查看记忆图变化
4. 调试和优化
### 📊 对于数据分析
1. 使用独立版查看历史数据
2. 切换不同时期的数据文件
3. 使用搜索和过滤功能
4. 导出数据进行分析
---
## 🐛 常见问题
### Q: 未找到数据文件？
**A**: 运行 `.\visualizer.ps1` 选择 `3` 生成示例数据
### Q: 端口被占用？
**A**: 修改对应服务器文件中的端口号，或关闭占用端口的程序
### Q: 两个版本有什么区别？
**A**:
- **独立版**: 快速，读文件，可切换，推荐日常使用
- **完整版**: 实时，用内存，完整功能，推荐开发调试
### Q: 图形显示混乱？
**A**:
1. 使用过滤器减少节点
2. 点击"适应窗口"
3. 刷新页面
---
## 🎉 开始使用
### 立即启动
```powershell
.\visualizer.ps1
```
### 访问地址
- 独立版: http://127.0.0.1:5001
- 完整版: http://127.0.0.1:5000
---
## 🤝 反馈与支持
如有问题或建议，请查看:
- 📖 `VISUALIZER_GUIDE.md` - 完整使用指南
- 📝 `tools/memory_visualizer/README.md` - 技术文档
---
## 🌟 特别感谢
感谢你使用 MoFox Bot 记忆图可视化工具！
**享受探索记忆图的乐趣！** 🚀🦊
---
_最后更新: 2025-11-06_

159
tools/memory_visualizer/docs/VISUALIZER_README.md
View File

@@ -1,159 +0,0 @@
# 🎯 记忆图可视化工具 - 快速参考
## 🚀 快速启动
### 推荐方式 (交互式菜单)
```powershell
.\visualizer.ps1
```
然后选择:
- **选项 1**: 独立版 (快速，推荐) ⭐
- **选项 2**: 完整版 (实时数据)
- **选项 3**: 生成示例数据
---
## 📋 各版本对比
| 特性 | 独立版 ⭐ | 完整版 |
|------|---------|--------|
| **启动速度** | 🚀 快速 (2秒) | ⏱️ 较慢 (5-10秒) |
| **数据源** | 📂 文件 | 💾 内存 (实时) |
| **文件切换** | ✅ 支持 | ❌ 不支持 |
| **资源占用** | 💚 低 | 💛 中等 |
| **端口** | 5001 | 5000 |
| **适用场景** | 查看历史数据、调试 | 实时监控、开发 |
---
## 🔧 手动启动命令
### 独立版 (推荐)
```powershell
# Windows
.\start_visualizer.ps1
# 或直接运行
.\.venv\Scripts\python.exe tools/memory_visualizer/visualizer_simple.py
```
访问: http://127.0.0.1:5001
### 完整版
```powershell
.\.venv\Scripts\python.exe tools/memory_visualizer/visualizer_server.py
```
访问: http://127.0.0.1:5000
### 生成示例数据
```powershell
.\.venv\Scripts\python.exe generate_sample_data.py
```
---
## 📊 功能一览
### 🎨 可视化功能
- ✅ 交互式图形 (拖动、缩放、点击)
- ✅ 节点类型颜色分类
- ✅ 实时搜索和过滤
- ✅ 统计信息展示
- ✅ 节点详情查看
### 📂 数据管理
- ✅ 自动搜索数据文件
- ✅ 多文件切换 (独立版)
- ✅ 数据导出 (JSON格式)
- ✅ 文件信息显示
---
## 🎯 使用场景
### 1️⃣ 首次使用
```powershell
# 1. 生成示例数据
.\visualizer.ps1
# 选择: 3
# 2. 启动可视化
.\visualizer.ps1
# 选择: 1
# 3. 打开浏览器
# 访问: http://127.0.0.1:5001
```
### 2️⃣ 查看实际数据
```powershell
# 先运行Bot生成记忆
# 然后启动可视化
.\visualizer.ps1
# 选择: 1 (独立版) 或 2 (完整版)
```
### 3️⃣ 调试记忆系统
```powershell
# 使用完整版，实时查看变化
.\visualizer.ps1
# 选择: 2
```
---
## 🐛 故障排除
### ❌ 问题: 未找到数据文件
**解决**:
```powershell
.\visualizer.ps1
# 选择 3 生成示例数据
```
### ❌ 问题: 端口被占用
**解决**:
- 独立版: 修改 `visualizer_simple.py` 中的 `port=5001`
- 完整版: 修改 `visualizer_server.py` 中的 `port=5000`
### ❌ 问题: 数据加载失败
**可能原因**:
- 数据文件格式不正确
- 文件损坏
**解决**:
1. 检查 `data/memory_graph/` 目录
2. 重新生成示例数据
3. 查看终端错误信息
---
## 📚 相关文档
- **完整指南**: `VISUALIZER_GUIDE.md`
- **快速入门**: `tools/memory_visualizer/QUICKSTART.md`
- **详细文档**: `tools/memory_visualizer/README.md`
- **更新日志**: `tools/memory_visualizer/CHANGELOG.md`
---
## 💡 提示
1. **首次使用**: 先生成示例数据 (选项 3)
2. **查看历史**: 使用独立版，可以切换不同数据文件
3. **实时监控**: 使用完整版，与Bot同时运行
4. **性能优化**: 大型图使用过滤器和搜索
5. **快捷键**:
- `Ctrl + 滚轮`: 缩放
- 拖动空白: 移动画布
- 点击节点: 查看详情
---
## 🎉 开始探索！
```powershell
.\visualizer.ps1
```
享受你的记忆图之旅！🚀🦊

9
tools/memory_visualizer/requirements.txt
View File

@@ -1,9 +0,0 @@
# 记忆图可视化工具依赖
# Web框架
flask>=2.3.0
flask-cors>=4.0.0
# 其他依赖由主项目提供
# - src.memory_graph
# - src.config

38
tools/memory_visualizer/run_visualizer.py
View File

@@ -1,38 +0,0 @@
#!/usr/bin/env python3
"""
记忆图可视化工具启动脚本
快速启动记忆图可视化Web服务器
"""
import sys
from pathlib import Path
# 添加项目根目录到路径
project_root = Path(__file__).parent
sys.path.insert(0, str(project_root))
from tools.memory_visualizer.visualizer_server import run_server
if __name__ == '__main__':
print("=" * 60)
print("🦊 MoFox Bot - 记忆图可视化工具")
print("=" * 60)
print()
print("📊 启动可视化服务器...")
print("🌐 访问地址: http://127.0.0.1:5000")
print("⏹️ 按 Ctrl+C 停止服务器")
print()
print("=" * 60)
try:
run_server(
host='127.0.0.1',
port=5000,
debug=True
)
except KeyboardInterrupt:
print("\n\n👋 服务器已停止")
except Exception as e:
print(f"\n❌ 启动失败: {e}")
sys.exit(1)

39
tools/memory_visualizer/run_visualizer_simple.py
View File

@@ -1,39 +0,0 @@
"""
快速启动脚本 - 记忆图可视化工具 (独立版)
使用说明:
1. 直接运行此脚本启动可视化服务器
2. 工具会自动搜索可用的数据文件
3. 如果找到多个文件,会使用最新的文件
4. 你也可以在Web界面中选择其他文件
"""
import sys
from pathlib import Path
# 添加项目根目录
project_root = Path(__file__).parent
sys.path.insert(0, str(project_root))
if __name__ == '__main__':
print("=" * 70)
print("🦊 MoFox Bot - 记忆图可视化工具 (独立版)")
print("=" * 70)
print()
print("✨ 特性:")
print(" • 自动搜索可用的数据文件")
print(" • 支持在Web界面中切换文件")
print(" • 快速启动,无需完整初始化")
print()
print("=" * 70)
try:
from tools.memory_visualizer.visualizer_simple import run_server
run_server(host='127.0.0.1', port=5001, debug=True)
except KeyboardInterrupt:
print("\n\n👋 服务器已停止")
except Exception as e:
print(f"\n❌ 启动失败: {e}")
import traceback
traceback.print_exc()
sys.exit(1)

53
tools/memory_visualizer/start_visualizer.bat
View File

@@ -1,53 +0,0 @@
@echo off
REM 记忆图可视化工具启动脚本 - CMD版本
echo ======================================================================
echo 🦊 MoFox Bot - 记忆图可视化工具
echo ======================================================================
echo.
REM 检查虚拟环境
set VENV_PYTHON=.venv\Scripts\python.exe
if not exist "%VENV_PYTHON%" (
echo ❌ 未找到虚拟环境: %VENV_PYTHON%
echo.
echo 请先创建虚拟环境:
echo python -m venv .venv
echo .venv\Scripts\activate.bat
echo pip install -r requirements.txt
echo.
exit /b 1
)
echo ✅ 使用虚拟环境: %VENV_PYTHON%
echo.
REM 检查依赖
echo 🔍 检查依赖...
"%VENV_PYTHON%" -c "import flask; import flask_cors" 2>nul
if errorlevel 1 (
echo ⚠️ 缺少依赖，正在安装...
"%VENV_PYTHON%" -m pip install flask flask-cors --quiet
if errorlevel 1 (
echo ❌ 安装依赖失败
exit /b 1
)
echo ✅ 依赖安装完成
)
echo ✅ 依赖检查完成
echo.
REM 显示信息
echo 📊 启动可视化服务器...
echo 🌐 访问地址: http://127.0.0.1:5001
echo ⏹️ 按 Ctrl+C 停止服务器
echo.
echo ======================================================================
echo.
REM 启动服务器
"%VENV_PYTHON%" "tools\memory_visualizer\visualizer_simple.py"
echo.
echo 👋 服务器已停止

65
tools/memory_visualizer/start_visualizer.ps1
View File

@@ -1,65 +0,0 @@
#!/usr/bin/env pwsh
# 记忆图可视化工具启动脚本 - PowerShell版本
Write-Host "=" -NoNewline -ForegroundColor Cyan
Write-Host ("=" * 69) -ForegroundColor Cyan
Write-Host "🦊 MoFox Bot - 记忆图可视化工具" -ForegroundColor Yellow
Write-Host "=" -NoNewline -ForegroundColor Cyan
Write-Host ("=" * 69) -ForegroundColor Cyan
Write-Host ""
# 检查虚拟环境
$venvPath = ".venv\Scripts\python.exe"
if (-not (Test-Path $venvPath)) {
Write-Host "❌ 未找到虚拟环境: $venvPath" -ForegroundColor Red
Write-Host ""
Write-Host "请先创建虚拟环境:" -ForegroundColor Yellow
Write-Host " python -m venv .venv" -ForegroundColor Cyan
Write-Host " .\.venv\Scripts\Activate.ps1" -ForegroundColor Cyan
Write-Host " pip install -r requirements.txt" -ForegroundColor Cyan
Write-Host ""
exit 1
}
Write-Host "✅ 使用虚拟环境: $venvPath" -ForegroundColor Green
Write-Host ""
# 检查依赖
Write-Host "🔍 检查依赖..." -ForegroundColor Cyan
& $venvPath -c "import flask; import flask_cors" 2>$null
if ($LASTEXITCODE -ne 0) {
Write-Host "⚠️ 缺少依赖，正在安装..." -ForegroundColor Yellow
& $venvPath -m pip install flask flask-cors --quiet
if ($LASTEXITCODE -ne 0) {
Write-Host "❌ 安装依赖失败" -ForegroundColor Red
exit 1
}
Write-Host "✅ 依赖安装完成" -ForegroundColor Green
}
Write-Host "✅ 依赖检查完成" -ForegroundColor Green
Write-Host ""
# 显示信息
Write-Host "📊 启动可视化服务器..." -ForegroundColor Cyan
Write-Host "🌐 访问地址: " -NoNewline -ForegroundColor White
Write-Host "http://127.0.0.1:5001" -ForegroundColor Blue
Write-Host "⏹️ 按 Ctrl+C 停止服务器" -ForegroundColor Yellow
Write-Host ""
Write-Host "=" -NoNewline -ForegroundColor Cyan
Write-Host ("=" * 69) -ForegroundColor Cyan
Write-Host ""
# 启动服务器
try {
& $venvPath "tools\memory_visualizer\visualizer_simple.py"
}
catch {
Write-Host ""
Write-Host "❌ 启动失败: $_" -ForegroundColor Red
exit 1
}
finally {
Write-Host ""
Write-Host "👋 服务器已停止" -ForegroundColor Yellow
}

53
tools/memory_visualizer/start_visualizer.sh
View File

@@ -1,53 +0,0 @@
#!/bin/bash
# 记忆图可视化工具启动脚本 - Bash版本 (Linux/Mac)
echo "======================================================================"
echo "🦊 MoFox Bot - 记忆图可视化工具"
echo "======================================================================"
echo ""
# 检查虚拟环境
VENV_PYTHON=".venv/bin/python"
if [ ! -f "$VENV_PYTHON" ]; then
echo "❌ 未找到虚拟环境: $VENV_PYTHON"
echo ""
echo "请先创建虚拟环境:"
echo " python -m venv .venv"
echo " source .venv/bin/activate"
echo " pip install -r requirements.txt"
echo ""
exit 1
fi
echo "✅ 使用虚拟环境: $VENV_PYTHON"
echo ""
# 检查依赖
echo "🔍 检查依赖..."
$VENV_PYTHON -c "import flask; import flask_cors" 2>/dev/null
if [ $? -ne 0 ]; then
echo "⚠️ 缺少依赖，正在安装..."
$VENV_PYTHON -m pip install flask flask-cors --quiet
if [ $? -ne 0 ]; then
echo "❌ 安装依赖失败"
exit 1
fi
echo "✅ 依赖安装完成"
fi
echo "✅ 依赖检查完成"
echo ""
# 显示信息
echo "📊 启动可视化服务器..."
echo "🌐 访问地址: http://127.0.0.1:5001"
echo "⏹️ 按 Ctrl+C 停止服务器"
echo ""
echo "======================================================================"
echo ""
# 启动服务器
$VENV_PYTHON "tools/memory_visualizer/visualizer_simple.py"
echo ""
echo "👋 服务器已停止"

59
tools/memory_visualizer/visualizer.ps1
View File

@@ -1,59 +0,0 @@
# 记忆图可视化工具统一启动脚本
param(
[switch]$Simple,
[switch]$Full,
[switch]$Generate,
[switch]$Test
)
$ScriptDir = Split-Path -Parent $MyInvocation.MyCommand.Path
$ProjectRoot = Split-Path -Parent (Split-Path -Parent $ScriptDir)
Set-Location $ProjectRoot
function Get-Python {
$paths = @(".venv\Scripts\python.exe", "venv\Scripts\python.exe")
foreach ($p in $paths) {
if (Test-Path $p) { return $p }
}
return $null
}
$python = Get-Python
if (-not $python) {
Write-Host "ERROR: Virtual environment not found" -ForegroundColor Red
exit 1
}
if ($Simple) {
Write-Host "Starting Simple Server on http://127.0.0.1:5001" -ForegroundColor Green
& $python "$ScriptDir\visualizer_simple.py"
}
elseif ($Full) {
Write-Host "Starting Full Server on http://127.0.0.1:5000" -ForegroundColor Green
& $python "$ScriptDir\visualizer_server.py"
}
elseif ($Generate) {
& $python "$ScriptDir\generate_sample_data.py"
}
elseif ($Test) {
& $python "$ScriptDir\test_visualizer.py"
}
else {
Write-Host "MoFox Bot - Memory Graph Visualizer" -ForegroundColor Cyan
Write-Host ""
Write-Host "[1] Start Simple Server (Recommended)"
Write-Host "[2] Start Full Server"
Write-Host "[3] Generate Test Data"
Write-Host "[4] Run Tests"
Write-Host "[Q] Quit"
Write-Host ""
$choice = Read-Host "Select"
switch ($choice) {
"1" { & $python "$ScriptDir\visualizer_simple.py" }
"2" { & $python "$ScriptDir\visualizer_server.py" }
"3" { & $python "$ScriptDir\generate_sample_data.py" }
"4" { & $python "$ScriptDir\test_visualizer.py" }
default { exit 0 }
}
}

335
tools/memory_visualizer/visualizer_server.py
View File

@@ -1,335 +0,0 @@
"""
记忆图可视化服务器
提供 Web API 用于可视化记忆图数据
"""
import asyncio
import logging
# 添加项目根目录到 Python 路径
import sys
from datetime import datetime
from pathlib import Path
from typing import Optional
from flask import Flask, jsonify, render_template, request
from flask_cors import CORS
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root))
from src.memory_graph.manager import MemoryManager
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = Flask(__name__)
CORS(app) # 允许跨域请求
# 全局记忆管理器
memory_manager: Optional[MemoryManager] = None
def init_memory_manager():
"""初始化记忆管理器"""
global memory_manager
if memory_manager is None:
try:
memory_manager = MemoryManager()
# 在新的事件循环中初始化
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(memory_manager.initialize())
logger.info("记忆管理器初始化成功")
except Exception as e:
logger.error(f"初始化记忆管理器失败: {e}")
raise
@app.route("/")
def index():
"""主页面"""
return render_template("visualizer.html")
@app.route("/api/graph/full")
def get_full_graph():
"""
获取完整记忆图数据
返回所有节点和边，格式化为前端可用的结构
"""
try:
if memory_manager is None:
init_memory_manager()
# 获取所有记忆
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# 获取所有记忆
all_memories = memory_manager.graph_store.get_all_memories()
# 构建节点和边数据
nodes_dict = {} # {node_id: node_data}
edges_dict = {} # {edge_id: edge_data} - 使用字典去重
memory_info = []
for memory in all_memories:
# 添加记忆信息
memory_info.append(
{
"id": memory.id,
"type": memory.memory_type.value,
"importance": memory.importance,
"activation": memory.activation,
"status": memory.status.value,
"created_at": memory.created_at.isoformat(),
"text": memory.to_text(),
"access_count": memory.access_count,
}
)
# 处理节点
for node in memory.nodes:
if node.id not in nodes_dict:
nodes_dict[node.id] = {
"id": node.id,
"label": node.content,
"type": node.node_type.value,
"group": node.node_type.name, # 用于颜色分组
"title": f"{node.node_type.value}: {node.content}",
"metadata": node.metadata,
"created_at": node.created_at.isoformat(),
}
# 处理边 - 使用字典自动去重
for edge in memory.edges:
edge_id = edge.id
# 如果ID已存在，生成唯一ID
counter = 1
original_edge_id = edge_id
while edge_id in edges_dict:
edge_id = f"{original_edge_id}_{counter}"
counter += 1
edges_dict[edge_id] = {
"id": edge_id,
"from": edge.source_id,
"to": edge.target_id,
"label": edge.relation,
"type": edge.edge_type.value,
"importance": edge.importance,
"title": f"{edge.edge_type.value}: {edge.relation}",
"arrows": "to",
"memory_id": memory.id,
}
nodes_list = list(nodes_dict.values())
edges_list = list(edges_dict.values())
return jsonify(
{
"success": True,
"data": {
"nodes": nodes_list,
"edges": edges_list,
"memories": memory_info,
"stats": {
"total_nodes": len(nodes_list),
"total_edges": len(edges_list),
"total_memories": len(all_memories),
},
},
}
)
except Exception as e:
logger.error(f"获取图数据失败: {e}", exc_info=True)
return jsonify({"success": False, "error": str(e)}), 500
@app.route("/api/memory/<memory_id>")
def get_memory_detail(memory_id: str):
"""
获取特定记忆的详细信息
Args:
memory_id: 记忆ID
"""
try:
if memory_manager is None:
init_memory_manager()
memory = memory_manager.graph_store.get_memory_by_id(memory_id)
if memory is None:
return jsonify({"success": False, "error": "记忆不存在"}), 404
return jsonify({"success": True, "data": memory.to_dict()})
except Exception as e:
logger.error(f"获取记忆详情失败: {e}", exc_info=True)
return jsonify({"success": False, "error": str(e)}), 500
@app.route("/api/search")
def search_memories():
"""
搜索记忆
Query参数:
- q: 搜索关键词
- type: 记忆类型过滤
- limit: 返回数量限制
"""
try:
if memory_manager is None:
init_memory_manager()
query = request.args.get("q", "")
memory_type = request.args.get("type", None)
limit = int(request.args.get("limit", 50))
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# 执行搜索
results = loop.run_until_complete(memory_manager.search_memories(query=query, top_k=limit))
# 构建返回数据
memories = []
for memory in results:
memories.append(
{
"id": memory.id,
"text": memory.to_text(),
"type": memory.memory_type.value,
"importance": memory.importance,
"created_at": memory.created_at.isoformat(),
}
)
return jsonify(
{
"success": True,
"data": {
"results": memories,
"count": len(memories),
},
}
)
except Exception as e:
logger.error(f"搜索失败: {e}", exc_info=True)
return jsonify({"success": False, "error": str(e)}), 500
@app.route("/api/stats")
def get_statistics():
"""
获取记忆图统计信息
"""
try:
if memory_manager is None:
init_memory_manager()
# 获取统计信息
all_memories = memory_manager.graph_store.get_all_memories()
all_nodes = set()
all_edges = 0
for memory in all_memories:
for node in memory.nodes:
all_nodes.add(node.id)
all_edges += len(memory.edges)
stats = {
"total_memories": len(all_memories),
"total_nodes": len(all_nodes),
"total_edges": all_edges,
"node_types": {},
"memory_types": {},
}
# 统计节点类型分布
for memory in all_memories:
mem_type = memory.memory_type.value
stats["memory_types"][mem_type] = stats["memory_types"].get(mem_type, 0) + 1
for node in memory.nodes:
node_type = node.node_type.value
stats["node_types"][node_type] = stats["node_types"].get(node_type, 0) + 1
return jsonify({"success": True, "data": stats})
except Exception as e:
logger.error(f"获取统计信息失败: {e}", exc_info=True)
return jsonify({"success": False, "error": str(e)}), 500
@app.route("/api/files")
def list_files():
"""
列出所有可用的数据文件
注意: 完整版服务器直接使用内存中的数据，不支持文件切换
"""
try:
from pathlib import Path
data_dir = Path("data/memory_graph")
files = []
if data_dir.exists():
for f in data_dir.glob("*.json"):
stat = f.stat()
files.append(
{
"path": str(f),
"name": f.name,
"size": stat.st_size,
"size_kb": round(stat.st_size / 1024, 2),
"modified": datetime.fromtimestamp(stat.st_mtime).isoformat(),
"modified_readable": datetime.fromtimestamp(stat.st_mtime).strftime("%Y-%m-%d %H:%M:%S"),
"is_current": True, # 完整版始终使用内存数据
}
)
return jsonify(
{
"success": True,
"files": files,
"count": len(files),
"current_file": "memory_manager (实时数据)",
"note": "完整版服务器使用实时内存数据，如需切换文件请使用独立版服务器",
}
)
except Exception as e:
logger.error(f"获取文件列表失败: {e}", exc_info=True)
return jsonify({"success": False, "error": str(e)}), 500
@app.route("/api/reload")
def reload_data():
"""
重新加载数据
"""
return jsonify({"success": True, "message": "完整版服务器使用实时数据，无需重新加载", "note": "数据始终是最新的"})
def run_server(host: str = "127.0.0.1", port: int = 5000, debug: bool = False):
"""
启动可视化服务器
Args:
host: 服务器地址
port: 端口号
debug: 是否开启调试模式
"""
logger.info(f"启动记忆图可视化服务器: http://{host}:{port}")
app.run(host=host, port=port, debug=debug)
if __name__ == "__main__":
run_server(debug=True)

480
tools/memory_visualizer/visualizer_simple.py
View File

@@ -1,480 +0,0 @@
"""
记忆图可视化 - 独立版本
直接从存储的数据文件生成可视化,无需启动完整的记忆管理器
"""
import orjson
import sys
from pathlib import Path
from datetime import datetime
from typing import Any, Dict, List, Set
from pathlib import Path
from typing import Any, Dict, List, Optional, Set
# 添加项目根目录
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root))
from flask import Flask, jsonify, render_template_string, request, send_from_directory
from flask_cors import CORS
app = Flask(__name__)
CORS(app)
# 数据缓存
graph_data_cache = None
data_dir = project_root / "data" / "memory_graph"
current_data_file = None # 当前选择的数据文件
def find_available_data_files() -> List[Path]:
"""查找所有可用的记忆图数据文件"""
files = []
if not data_dir.exists():
return files
# 查找多种可能的文件名
possible_files = [
"graph_store.json",
"memory_graph.json",
"graph_data.json",
]
for filename in possible_files:
file_path = data_dir / filename
if file_path.exists():
files.append(file_path)
# 查找所有备份文件
for pattern in ["graph_store_*.json", "memory_graph_*.json", "graph_data_*.json"]:
for backup_file in data_dir.glob(pattern):
if backup_file not in files:
files.append(backup_file)
# 查找backups子目录
backups_dir = data_dir / "backups"
if backups_dir.exists():
for backup_file in backups_dir.glob("**/*.json"):
if backup_file not in files:
files.append(backup_file)
# 查找data/backup目录
backup_dir = data_dir.parent / "backup"
if backup_dir.exists():
for backup_file in backup_dir.glob("**/graph_*.json"):
if backup_file not in files:
files.append(backup_file)
for backup_file in backup_dir.glob("**/memory_*.json"):
if backup_file not in files:
files.append(backup_file)
return sorted(files, key=lambda f: f.stat().st_mtime, reverse=True)
def load_graph_data(file_path: Optional[Path] = None) -> Dict[str, Any]:
"""从磁盘加载图数据"""
global graph_data_cache, current_data_file
# 如果指定了新文件，清除缓存
if file_path is not None and file_path != current_data_file:
graph_data_cache = None
current_data_file = file_path
if graph_data_cache is not None:
return graph_data_cache
try:
# 确定要加载的文件
if current_data_file is not None:
graph_file = current_data_file
else:
# 尝试查找可用的数据文件
available_files = find_available_data_files()
if not available_files:
print(f"⚠️ 未找到任何图数据文件")
print(f"📂 搜索目录: {data_dir}")
return {
"nodes": [],
"edges": [],
"memories": [],
"stats": {"total_nodes": 0, "total_edges": 0, "total_memories": 0},
"error": "未找到数据文件",
"available_files": []
}
# 使用最新的文件
graph_file = available_files[0]
current_data_file = graph_file
print(f"📂 自动选择最新文件: {graph_file}")
if not graph_file.exists():
print(f"⚠️ 图数据文件不存在: {graph_file}")
return {
"nodes": [],
"edges": [],
"memories": [],
"stats": {"total_nodes": 0, "total_edges": 0, "total_memories": 0},
"error": f"文件不存在: {graph_file}"
}
print(f"📂 加载图数据: {graph_file}")
with open(graph_file, 'r', encoding='utf-8') as f:
data = orjson.loads(f.read())
# 解析数据
nodes_dict = {}
edges_list = []
memory_info = []
# 实际文件格式是 {nodes: [], edges: [], metadata: {}}
# 不是 {memories: [{nodes: [], edges: []}]}
nodes = data.get("nodes", [])
edges = data.get("edges", [])
metadata = data.get("metadata", {})
print(f"✅ 找到 {len(nodes)} 个节点, {len(edges)} 条边")
# 处理节点
for node in nodes:
node_id = node.get('id', '')
if node_id and node_id not in nodes_dict:
memory_ids = node.get('metadata', {}).get('memory_ids', [])
nodes_dict[node_id] = {
'id': node_id,
'label': node.get('content', ''),
'type': node.get('node_type', ''),
'group': extract_group_from_type(node.get('node_type', '')),
'title': f"{node.get('node_type', '')}: {node.get('content', '')}",
'metadata': node.get('metadata', {}),
'created_at': node.get('created_at', ''),
'memory_ids': memory_ids,
}
# 处理边 - 使用集合去重，避免重复的边ID
existing_edge_ids = set()
for edge in edges:
# 边的ID字段可能是 'id' 或 'edge_id'
edge_id = edge.get('edge_id') or edge.get('id', '')
# 如果ID为空或已存在，跳过这条边
if not edge_id or edge_id in existing_edge_ids:
continue
existing_edge_ids.add(edge_id)
memory_id = edge.get('metadata', {}).get('memory_id', '')
# 注意: GraphStore 保存的格式使用 'source'/'target', 不是 'source_id'/'target_id'
edges_list.append({
'id': edge_id,
'from': edge.get('source', edge.get('source_id', '')),
'to': edge.get('target', edge.get('target_id', '')),
'label': edge.get('relation', ''),
'type': edge.get('edge_type', ''),
'importance': edge.get('importance', 0.5),
'title': f"{edge.get('edge_type', '')}: {edge.get('relation', '')}",
'arrows': 'to',
'memory_id': memory_id,
})
# 从元数据中获取统计信息
stats = metadata.get('statistics', {})
total_memories = stats.get('total_memories', 0)
# TODO: 如果需要记忆详细信息,需要从其他地方加载
# 目前只有节点和边的数据
graph_data_cache = {
'nodes': list(nodes_dict.values()),
'edges': edges_list,
'memories': memory_info, # 空列表,因为文件中没有记忆详情
'stats': {
'total_nodes': len(nodes_dict),
'total_edges': len(edges_list),
'total_memories': total_memories,
},
'current_file': str(graph_file),
'file_size': graph_file.stat().st_size,
'file_modified': datetime.fromtimestamp(graph_file.stat().st_mtime).isoformat(),
}
print(f"📊 统计: {len(nodes_dict)} 个节点, {len(edges_list)} 条边, {total_memories} 条记忆")
print(f"📄 数据文件: {graph_file} ({graph_file.stat().st_size / 1024:.2f} KB)")
return graph_data_cache
except Exception as e:
print(f"❌ 加载失败: {e}")
import traceback
traceback.print_exc()
return {"nodes": [], "edges": [], "memories": [], "stats": {}}
def extract_group_from_type(node_type: str) -> str:
"""从节点类型提取分组名"""
# 假设类型格式为 "主体" 或 "SUBJECT"
type_mapping = {
'主体': 'SUBJECT',
'主题': 'TOPIC',
'客体': 'OBJECT',
'属性': 'ATTRIBUTE',
'值': 'VALUE',
}
return type_mapping.get(node_type, node_type)
def generate_memory_text(memory: Dict[str, Any]) -> str:
"""生成记忆的文本描述"""
try:
nodes = {n['id']: n for n in memory.get('nodes', [])}
edges = memory.get('edges', [])
subject_id = memory.get('subject_id', '')
if not subject_id or subject_id not in nodes:
return f"[记忆 {memory.get('id', '')[:8]}]"
parts = [nodes[subject_id]['content']]
# 找主题节点
for edge in edges:
if edge.get('edge_type') == '记忆类型' and edge.get('source_id') == subject_id:
topic_id = edge.get('target_id', '')
if topic_id in nodes:
parts.append(nodes[topic_id]['content'])
# 找客体
for e2 in edges:
if e2.get('edge_type') == '核心关系' and e2.get('source_id') == topic_id:
obj_id = e2.get('target_id', '')
if obj_id in nodes:
parts.append(f"{e2.get('relation', '')} {nodes[obj_id]['content']}")
break
break
return " ".join(parts)
except Exception:
return f"[记忆 {memory.get('id', '')[:8]}]"
# 使用内嵌的HTML模板(与之前相同)
HTML_TEMPLATE = open(project_root / "tools" / "memory_visualizer" / "templates" / "visualizer.html", 'r', encoding='utf-8').read()
@app.route('/')
def index():
"""主页面"""
return render_template_string(HTML_TEMPLATE)
@app.route('/api/graph/full')
def get_full_graph():
"""获取完整记忆图数据"""
try:
data = load_graph_data()
return jsonify({
'success': True,
'data': data
})
except Exception as e:
return jsonify({
'success': False,
'error': str(e)
}), 500
@app.route('/api/memory/<memory_id>')
def get_memory_detail(memory_id: str):
"""获取记忆详情"""
try:
data = load_graph_data()
memory = next((m for m in data['memories'] if m['id'] == memory_id), None)
if memory is None:
return jsonify({
'success': False,
'error': '记忆不存在'
}), 404
return jsonify({
'success': True,
'data': memory
})
except Exception as e:
return jsonify({
'success': False,
'error': str(e)
}), 500
@app.route('/api/search')
def search_memories():
"""搜索记忆"""
try:
query = request.args.get('q', '').lower()
limit = int(request.args.get('limit', 50))
data = load_graph_data()
# 简单的文本匹配搜索
results = []
for memory in data['memories']:
text = memory.get('text', '').lower()
if query in text:
results.append(memory)
return jsonify({
'success': True,
'data': {
'results': results[:limit],
'count': len(results),
}
})
except Exception as e:
return jsonify({
'success': False,
'error': str(e)
}), 500
@app.route('/api/stats')
def get_statistics():
"""获取统计信息"""
try:
data = load_graph_data()
# 扩展统计信息
node_types = {}
memory_types = {}
for node in data['nodes']:
node_type = node.get('type', 'Unknown')
node_types[node_type] = node_types.get(node_type, 0) + 1
for memory in data['memories']:
mem_type = memory.get('type', 'Unknown')
memory_types[mem_type] = memory_types.get(mem_type, 0) + 1
stats = data.get('stats', {})
stats['node_types'] = node_types
stats['memory_types'] = memory_types
return jsonify({
'success': True,
'data': stats
})
except Exception as e:
return jsonify({
'success': False,
'error': str(e)
}), 500
@app.route('/api/reload')
def reload_data():
"""重新加载数据"""
global graph_data_cache
graph_data_cache = None
data = load_graph_data()
return jsonify({
'success': True,
'message': '数据已重新加载',
'stats': data.get('stats', {})
})
@app.route('/api/files')
def list_files():
"""列出所有可用的数据文件"""
try:
files = find_available_data_files()
file_list = []
for f in files:
stat = f.stat()
file_list.append({
'path': str(f),
'name': f.name,
'size': stat.st_size,
'size_kb': round(stat.st_size / 1024, 2),
'modified': datetime.fromtimestamp(stat.st_mtime).isoformat(),
'modified_readable': datetime.fromtimestamp(stat.st_mtime).strftime('%Y-%m-%d %H:%M:%S'),
'is_current': str(f) == str(current_data_file) if current_data_file else False
})
return jsonify({
'success': True,
'files': file_list,
'count': len(file_list),
'current_file': str(current_data_file) if current_data_file else None
})
except Exception as e:
return jsonify({
'success': False,
'error': str(e)
}), 500
@app.route('/api/select_file', methods=['POST'])
def select_file():
"""选择要加载的数据文件"""
global graph_data_cache, current_data_file
try:
data = request.get_json()
file_path = data.get('file_path')
if not file_path:
return jsonify({
'success': False,
'error': '未提供文件路径'
}), 400
file_path = Path(file_path)
if not file_path.exists():
return jsonify({
'success': False,
'error': f'文件不存在: {file_path}'
}), 404
# 清除缓存并加载新文件
graph_data_cache = None
current_data_file = file_path
graph_data = load_graph_data(file_path)
return jsonify({
'success': True,
'message': f'已切换到文件: {file_path.name}',
'stats': graph_data.get('stats', {})
})
except Exception as e:
return jsonify({
'success': False,
'error': str(e)
}), 500
def run_server(host: str = '127.0.0.1', port: int = 5001, debug: bool = False):
"""启动服务器"""
print("=" * 60)
print("🦊 MoFox Bot - 记忆图可视化工具 (独立版)")
print("=" * 60)
print(f"📂 数据目录: {data_dir}")
print(f"🌐 访问地址: http://{host}:{port}")
print("⏹️ 按 Ctrl+C 停止服务器")
print("=" * 60)
print()
# 预加载数据
load_graph_data()
app.run(host=host, port=port, debug=debug)
if __name__ == '__main__':
try:
run_server(debug=True)
except KeyboardInterrupt:
print("\n\n👋 服务器已停止")
except Exception as e:
print(f"\n❌ 启动失败: {e}")
sys.exit(1)

16
visualizer.ps1
View File

@@ -1,16 +0,0 @@
#!/usr/bin/env pwsh
# ======================================================================
# 记忆图可视化工具 - 快捷启动脚本
# ======================================================================
# 此脚本是快捷方式，实际脚本位于 tools/memory_visualizer/ 目录
# ======================================================================
$visualizerScript = Join-Path $PSScriptRoot "tools\memory_visualizer\visualizer.ps1"
if (Test-Path $visualizerScript) {
& $visualizerScript @args
} else {
Write-Host "❌ 错误：找不到可视化工具脚本" -ForegroundColor Red
Write-Host " 预期位置: $visualizerScript" -ForegroundColor Yellow
exit 1
}