diff --git a/PRD.md b/PRD.md new file mode 100644 index 00000000..834431c0 --- /dev/null +++ b/PRD.md @@ -0,0 +1,243 @@ +# Foresight 先见之明 — 产品需求文档 (PRD) + +## 1. 产品概述 + +Foresight（先见之明）是一个基于知识图谱和 LLM 的社交媒体舆情模拟平台。用户上传文档资料，系统自动构建知识图谱、生成虚拟 Agent 画像，并模拟社交媒体上的传播与互动行为，最终生成分析报告。 + +**核心价值**：在事件发生前预判舆论走向，帮助品牌、政府、机构提前制定应对策略。 + +**上游项目**：基于 [MiroFish](https://github.com/666ghj/MiroFish) v0.1.2 二次开发。 + +--- + +## 2. 目标用户 + +| 用户类型 | 使用场景 | +|----------|----------| +| 品牌公关团队 | 新品发布前预判舆论反应 | +| 政府舆情分析师 | 政策出台前模拟民意 | +| 内容创作者 | 预测爆款视频的传播路径 | +| 研究人员 | 社交网络传播行为研究 | + +--- + +## 3. 系统架构 + +``` +用户浏览器 + │ + ├── 前端 (Vue 3 + Vite) + │ 部署: 腾讯云 COS + CDN + │ 域名: foresight.yizhou.chat + │ + └── 后端 (Flask + Python) + 端口: 5001 + │ + ├── LLM API (MiniMax M2.7 Highspeed) + │ 用途: 本体生成、画像生成、配置生成、报告生成 + │ + └── Zep Cloud API + 用途: 知识图谱存储、搜索、记忆更新 +``` + +--- + +## 4. 核心功能流程（5 步流水线） + +### Step 1: 图谱构建 + +**输入**: 用户上传文档（PDF/MD/TXT）+ 模拟需求描述 + +**流程**: +1. 文档解析 → 文本提取 +2. LLM 分析文档 → 生成本体（10 个实体类型 + 6-10 个关系类型） +3. 文本分块 → 批量导入 Zep → 构建知识图谱 +4. 返回图谱可视化（节点 + 边） + +**API 端点**: +- `POST /api/graph/ontology/generate` — 本体生成 +- `POST /api/graph/build` — 图谱构建 +- `GET /api/graph/task/` — 构建进度查询 + +**Token 消耗**: +| 服务 | 小文档 (10 实体) | 大文档 (50 实体) | +|------|------------------|------------------| +| LLM (本体生成) | 3K-8K | 5K-12K | +| Zep (图谱构建) | 5K-10K | 20K-50K | + +### Step 2: 环境配置 + +**输入**: 已构建的知识图谱 + +**流程**: +1. 从 Zep 读取图谱实体和关系 +2. 按实体类型筛选，为每个实体生成 Agent 画像（LLM） +3. 生成模拟配置：时间线、事件、Agent 活动参数、平台配置 + +**API 端点**: +- `POST /api/simulation/prepare` — 准备模拟环境 + +**Token 消耗**: +| 服务 | 小规模 (10 实体) | 大规模 (50 实体) | +|------|------------------|------------------| +| LLM (画像生成) | 20K-40K | 50K-100K | +| LLM (配置生成) | 20K-35K | 40K-50K | +| Zep (实体读取) | 1K-2K | 5K-10K | + +### Step 3: 模拟运行 + +**输入**: Agent 画像 + 模拟配置 + +**流程**: +1. 创建虚拟社交平台环境 +2. Agent 按配置执行社交行为（发帖、评论、转发、点赞等） +3. 实时记录互动日志 +4. 可选：将 Agent 行为写回 Zep 图谱（记忆更新） + +**API 端点**: +- `POST /api/simulation/run` — 启动模拟 +- `GET /api/simulation/status/` — 查询状态 +- `GET /api/simulation/history` — 历史记录 + +**Token 消耗**: +| 服务 | 说明 | +|------|------| +| Zep (记忆更新，可选) | 每个 Agent 动作 100-300 tokens，大规模模拟可达 1M+ | + +### Step 4: 报告生成 + +**输入**: 模拟结果 + 知识图谱 + +**流程**: +1. LLM 规划报告大纲（5 个章节） +2. 每个章节使用 ReACT 循环（推理→工具调用→生成） +3. 工具调用包括：图谱搜索（InsightForge/Panorama）、节点详情查询 +4. 最终输出结构化分析报告 + +**API 端点**: +- `POST /api/report/generate` — 生成报告 + +**Token 消耗**: +| 服务 | 小规模 | 大规模 | +|------|--------|--------| +| LLM (ReACT 多轮) | 50K-80K | 100K-150K | +| Zep (图谱搜索) | 5K-10K | 10K-20K | + +> 报告生成是整个流水线中**Token 消耗最大**的环节。 + +### Step 5: 交互问答 + +**输入**: 用户问题 + +**流程**: +1. 用户自由提问 +2. 系统结合图谱搜索 + LLM 回答 + +**API 端点**: +- `POST /api/report/chat` — 实时问答 + +**Token 消耗**: 每条消息 2K-4K tokens (LLM + Zep) + +--- + +## 5. Token 消耗总览 + +### 单次完整流水线估算 + +| 阶段 | 主要 API | 10 实体 | 50 实体 | +|------|----------|---------|---------| +| 图谱构建 | LLM + Zep | 8K-18K | 25K-62K | +| 环境配置 | LLM + Zep | 41K-77K | 95K-160K | +| 模拟运行 | Zep (可选) | 0-100K | 0-1M+ | +| 报告生成 | LLM + Zep | 55K-90K | 110K-170K | +| **合计 (不含模拟记忆)** | | **~100K-185K** | **~230K-392K** | + +### API 费用构成 + +| 外部服务 | 用途 | 计费方式 | +|----------|------|----------| +| **MiniMax M2.7 Highspeed** | 所有 LLM 推理（本体/画像/配置/报告/问答） | 按 token 计费 | +| **Zep Cloud** | 知识图谱（存储/搜索/记忆更新） | 按 API 调用计费 | + +--- + +## 6. 技术栈 + +### 前端 +| 技术 | 版本 | 用途 | +|------|------|------| +| Vue 3 | 3.x | UI 框架 | +| Vite | 7.x | 构建工具 | +| D3.js / Force Graph | - | 图谱可视化 | + +### 后端 +| 技术 | 版本 | 用途 | +|------|------|------| +| Python | 3.x | 运行时 | +| Flask | - | Web 框架 | +| OpenAI SDK | - | LLM 客户端（兼容 MiniMax） | +| zep-cloud | 3.13.0 | Zep 知识图谱 SDK | + +### 部署 +| 组件 | 平台 | 说明 | +|------|------|------| +| 前端静态文件 | 腾讯云 COS + CDN | foresight.yizhou.chat | +| SSL 证书 | Let's Encrypt | 通过 acme.sh 签发 | +| 后端 API | 待部署 | 需要云服务器运行 Flask | + +--- + +## 7. 配置项 + +```env +# LLM 配置 +LLM_API_KEY= +LLM_BASE_URL=https://api.minimax.chat/v1 +LLM_MODEL_NAME=MiniMax-M2.5 + +# Zep 配置 +ZEP_API_KEY= + +# 服务端口 +FLASK_PORT=5001 +``` + +--- + +## 8. 当前状态与待办 + +### 已完成 +- [x] 前端 UI（Vue 3，支持明暗主题） +- [x] 品牌迁移（MiroFish → Foresight 先见之明） +- [x] 前端部署（腾讯云 COS + CDN + HTTPS） +- [x] Dark Mode 全屏响应式布局 +- [x] 国际化支持（中/英） + +### 待完成 +- [ ] **后端云部署**：当前后端只能在本地运行（localhost:5001），需部署到腾讯云 CVM 或轻量服务器 +- [ ] **前端 API 地址配置**：设置 `VITE_API_BASE_URL` 指向云端后端 +- [ ] **图谱生成功能验证**：端到端测试完整流水线 +- [ ] **模拟结果持久化**：当前模拟结果存在内存中，需接入数据库 +- [ ] **用户认证**：多用户场景下的身份管理 + +--- + +## 9. 关键文件索引 + +| 路径 | 用途 | +|------|------| +| `frontend/src/api/index.js` | API 客户端配置（baseURL） | +| `frontend/src/views/Process.vue` | 图谱构建主界面 | +| `frontend/src/views/SimulationView.vue` | 模拟运行界面 | +| `frontend/src/views/ReportView.vue` | 报告查看界面 | +| `backend/run.py` | 后端入口 | +| `backend/app/api/graph.py` | 图谱相关 API 端点 | +| `backend/app/api/simulation.py` | 模拟相关 API 端点 | +| `backend/app/api/report.py` | 报告相关 API 端点 | +| `backend/app/services/ontology_generator.py` | 本体生成服务（LLM） | +| `backend/app/services/graph_builder.py` | 图谱构建服务（Zep） | +| `backend/app/services/oasis_profile_generator.py` | Agent 画像生成（LLM + Zep） | +| `backend/app/services/simulation_config_generator.py` | 模拟配置生成（LLM） | +| `backend/app/services/report_agent.py` | 报告生成（ReACT，LLM + Zep） | +| `backend/app/utils/llm_client.py` | LLM 客户端封装 | +| `.env` | API Keys 和配置 | diff --git a/backend/app/config.py b/backend/app/config.py index 7d7e76b4..195762d9 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -32,8 +32,13 @@ class Config: LLM_BASE_URL = os.environ.get('LLM_BASE_URL', 'https://api.openai.com/v1') LLM_MODEL_NAME = os.environ.get('LLM_MODEL_NAME', 'gpt-4o-mini') - # Zep配置 - ZEP_API_KEY = os.environ.get('ZEP_API_KEY') + # Neo4j / Graphiti 配置 + NEO4J_URI = os.environ.get('NEO4J_URI', 'bolt://localhost:7687') + NEO4J_USER = os.environ.get('NEO4J_USER', 'neo4j') + NEO4J_PASSWORD = os.environ.get('NEO4J_PASSWORD', 'foresight2026') + + # 兼容旧配置：ZEP_API_KEY 不再需要 + ZEP_API_KEY = os.environ.get('ZEP_API_KEY', 'deprecated') # 文件上传配置 MAX_CONTENT_LENGTH = 50 * 1024 * 1024 # 50MB @@ -69,7 +74,7 @@ class Config: errors = [] if not cls.LLM_API_KEY: errors.append("LLM_API_KEY 未配置") - if not cls.ZEP_API_KEY: - errors.append("ZEP_API_KEY 未配置") + if not cls.NEO4J_PASSWORD: + errors.append("NEO4J_PASSWORD 未配置") return errors diff --git a/backend/app/services/graph_builder.py b/backend/app/services/graph_builder.py index 715b9b68..3c1c10fc 100644 --- a/backend/app/services/graph_builder.py +++ b/backend/app/services/graph_builder.py @@ -1,6 +1,6 @@ """ 图谱构建服务 -接口2：使用Zep API构建Standalone Graph +接口2：使用 Graphiti + Neo4j 构建知识图谱（替代 Zep Cloud） """ import os @@ -10,12 +10,9 @@ import threading from typing import Dict, Any, List, Optional, Callable from dataclasses import dataclass -from zep_cloud.client import Zep -from zep_cloud import EpisodeData, EntityEdgeSourceTarget - from ..config import Config from ..models.task import TaskManager, TaskStatus -from ..utils.zep_paging import fetch_all_nodes, fetch_all_edges +from .graphiti_client import GraphitiClient from .text_processor import TextProcessor from ..utils.locale import t, get_locale, set_locale @@ -27,7 +24,7 @@ class GraphInfo: node_count: int edge_count: int entity_types: List[str] - + def to_dict(self) -> Dict[str, Any]: return { "graph_id": self.graph_id, @@ -40,17 +37,14 @@ class GraphInfo: class GraphBuilderService: """ 图谱构建服务 - 负责调用Zep API构建知识图谱 + 使用 Graphiti + Neo4j 构建知识图谱 """ - + def __init__(self, api_key: Optional[str] = None): - self.api_key = api_key or Config.ZEP_API_KEY - if not self.api_key: - raise ValueError("ZEP_API_KEY 未配置") - - self.client = Zep(api_key=self.api_key) + # api_key kept for interface compatibility but unused + self.client = GraphitiClient.get_instance() self.task_manager = TaskManager() - + def build_graph_async( self, text: str, @@ -60,21 +54,7 @@ class GraphBuilderService: chunk_overlap: int = 50, batch_size: int = 3 ) -> str: - """ - 异步构建图谱 - - Args: - text: 输入文本 - ontology: 本体定义（来自接口1的输出） - graph_name: 图谱名称 - chunk_size: 文本块大小 - chunk_overlap: 块重叠大小 - batch_size: 每批发送的块数量 - - Returns: - 任务ID - """ - # 创建任务 + """异步构建图谱，返回任务ID""" task_id = self.task_manager.create_task( task_type="graph_build", metadata={ @@ -83,20 +63,18 @@ class GraphBuilderService: "text_length": len(text), } ) - - # Capture locale before spawning background thread + current_locale = get_locale() - # 在后台线程中执行构建 thread = threading.Thread( target=self._build_graph_worker, args=(task_id, text, ontology, graph_name, chunk_size, chunk_overlap, batch_size, current_locale) ) thread.daemon = True thread.start() - + return task_id - + def _build_graph_worker( self, task_id: str, @@ -117,23 +95,23 @@ class GraphBuilderService: progress=5, message=t('progress.startBuildingGraph') ) - - # 1. 创建图谱 - graph_id = self.create_graph(graph_name) + + # 1. 创建图谱（分配 group_id） + graph_id = self.client.create_graph(graph_name) self.task_manager.update_task( task_id, progress=10, message=t('progress.graphCreated', graphId=graph_id) ) - - # 2. 设置本体 - self.set_ontology(graph_id, ontology) + + # 2. 准备本体类型（Graphiti 使用 Pydantic 模型） + entity_types, edge_types = self._prepare_ontology(ontology) self.task_manager.update_task( task_id, progress=15, message=t('progress.ontologySet') ) - + # 3. 文本分块 chunks = TextProcessor.split_text(text, chunk_size, chunk_overlap) total_chunks = len(chunks) @@ -142,155 +120,113 @@ class GraphBuilderService: progress=20, message=t('progress.textSplit', count=total_chunks) ) - - # 4. 分批发送数据 - episode_uuids = self.add_text_batches( - graph_id, chunks, batch_size, - lambda msg, prog: self.task_manager.update_task( + + # 4. 逐个添加 episode（Graphiti 同步处理，不需要轮询） + def progress_cb(msg, prog): + self.task_manager.update_task( task_id, - progress=20 + int(prog * 0.4), # 20-60% + progress=20 + int(prog * 0.65), # 20-85% message=msg ) + + self.client.add_episodes_batch( + graph_id=graph_id, + texts=chunks, + source_description=graph_name, + entity_types=entity_types, + edge_types=edge_types, + progress_callback=progress_cb, ) - - # 5. 等待Zep处理完成 - self.task_manager.update_task( - task_id, - progress=60, - message=t('progress.waitingZepProcess') - ) - - self._wait_for_episodes( - episode_uuids, - lambda msg, prog: self.task_manager.update_task( - task_id, - progress=60 + int(prog * 0.3), # 60-90% - message=msg - ) - ) - - # 6. 获取图谱信息 + + # 5. 获取图谱信息 self.task_manager.update_task( task_id, progress=90, message=t('progress.fetchingGraphInfo') ) - + graph_info = self._get_graph_info(graph_id) - + # 完成 self.task_manager.complete_task(task_id, { "graph_id": graph_id, "graph_info": graph_info.to_dict(), "chunks_processed": total_chunks, }) - + except Exception as e: import traceback error_msg = f"{str(e)}\n{traceback.format_exc()}" self.task_manager.fail_task(task_id, error_msg) - + def create_graph(self, name: str) -> str: - """创建Zep图谱（公开方法）""" - graph_id = f"foresight_{uuid.uuid4().hex[:16]}" - - self.client.graph.create( - graph_id=graph_id, - name=name, - description="Foresight Social Simulation Graph" - ) - - return graph_id - + """创建图谱（公开方法）""" + return self.client.create_graph(name) + def set_ontology(self, graph_id: str, ontology: Dict[str, Any]): - """设置图谱本体（公开方法）""" - import warnings - from typing import Optional - from pydantic import Field - from zep_cloud.external_clients.ontology import EntityModel, EntityText, EdgeModel - - # 抑制 Pydantic v2 关于 Field(default=None) 的警告 - # 这是 Zep SDK 要求的用法，警告来自动态类创建，可以安全忽略 - warnings.filterwarnings('ignore', category=UserWarning, module='pydantic') - - # Zep 保留名称，不能作为属性名 - RESERVED_NAMES = {'uuid', 'name', 'group_id', 'name_embedding', 'summary', 'created_at'} - - def safe_attr_name(attr_name: str) -> str: - """将保留名称转换为安全名称""" - if attr_name.lower() in RESERVED_NAMES: - return f"entity_{attr_name}" - return attr_name - - # 动态创建实体类型 + """设置图谱本体（Graphiti 在 add_episode 时传入，此方法保留兼容性）""" + # Graphiti doesn't have a separate set_ontology step; + # ontology is passed per add_episode call. + # Store it for later use. + self._stored_ontology = ontology + + def _prepare_ontology(self, ontology: Dict[str, Any]) -> tuple: + """ + 将前端本体格式转换为 Graphiti Pydantic 模型格式 + + Returns: + (entity_types dict, edge_types dict) for Graphiti + """ + from pydantic import BaseModel, Field + from typing import Optional as Opt + entity_types = {} for entity_def in ontology.get("entity_types", []): name = entity_def["name"] description = entity_def.get("description", f"A {name} entity.") - - # 创建属性字典和类型注解（Pydantic v2 需要） - attrs = {"__doc__": description} + + # Build Pydantic model dynamically + attrs = {} annotations = {} - for attr_def in entity_def.get("attributes", []): - attr_name = safe_attr_name(attr_def["name"]) # 使用安全名称 + attr_name = attr_def["name"] + # Skip reserved names + if attr_name.lower() in {'uuid', 'name', 'group_id', 'name_embedding', 'summary', 'created_at'}: + attr_name = f"entity_{attr_name}" attr_desc = attr_def.get("description", attr_name) - # Zep API 需要 Field 的 description，这是必需的 attrs[attr_name] = Field(description=attr_desc, default=None) - annotations[attr_name] = Optional[EntityText] # 类型注解 - + annotations[attr_name] = Opt[str] + attrs["__annotations__"] = annotations - - # 动态创建类 - entity_class = type(name, (EntityModel,), attrs) + attrs["__doc__"] = description + entity_class = type(name, (BaseModel,), attrs) entity_class.__doc__ = description entity_types[name] = entity_class - - # 动态创建边类型 - edge_definitions = {} + + edge_types = {} for edge_def in ontology.get("edge_types", []): name = edge_def["name"] description = edge_def.get("description", f"A {name} relationship.") - - # 创建属性字典和类型注解 - attrs = {"__doc__": description} + + attrs = {} annotations = {} - for attr_def in edge_def.get("attributes", []): - attr_name = safe_attr_name(attr_def["name"]) # 使用安全名称 + attr_name = attr_def["name"] + if attr_name.lower() in {'uuid', 'name', 'group_id', 'name_embedding', 'summary', 'created_at'}: + attr_name = f"edge_{attr_name}" attr_desc = attr_def.get("description", attr_name) - # Zep API 需要 Field 的 description，这是必需的 attrs[attr_name] = Field(description=attr_desc, default=None) - annotations[attr_name] = Optional[str] # 边属性用str类型 - + annotations[attr_name] = Opt[str] + attrs["__annotations__"] = annotations - - # 动态创建类 + attrs["__doc__"] = description class_name = ''.join(word.capitalize() for word in name.split('_')) - edge_class = type(class_name, (EdgeModel,), attrs) + edge_class = type(class_name, (BaseModel,), attrs) edge_class.__doc__ = description - - # 构建source_targets - source_targets = [] - for st in edge_def.get("source_targets", []): - source_targets.append( - EntityEdgeSourceTarget( - source=st.get("source", "Entity"), - target=st.get("target", "Entity") - ) - ) - - if source_targets: - edge_definitions[name] = (edge_class, source_targets) - - # 调用Zep API设置本体 - if entity_types or edge_definitions: - self.client.graph.set_ontology( - graph_ids=[graph_id], - entities=entity_types if entity_types else None, - edges=edge_definitions if edge_definitions else None, - ) - + edge_types[name] = edge_class + + return entity_types if entity_types else None, edge_types if edge_types else None + def add_text_batches( self, graph_id: str, @@ -298,122 +234,25 @@ class GraphBuilderService: batch_size: int = 3, progress_callback: Optional[Callable] = None ) -> List[str]: - """分批添加文本到图谱，返回所有 episode 的 uuid 列表""" - episode_uuids = [] - total_chunks = len(chunks) - - for i in range(0, total_chunks, batch_size): - batch_chunks = chunks[i:i + batch_size] - batch_num = i // batch_size + 1 - total_batches = (total_chunks + batch_size - 1) // batch_size - - if progress_callback: - progress = (i + len(batch_chunks)) / total_chunks - progress_callback( - t('progress.sendingBatch', current=batch_num, total=total_batches, chunks=len(batch_chunks)), - progress - ) - - # 构建episode数据 - episodes = [ - EpisodeData(data=chunk, type="text") - for chunk in batch_chunks - ] - - # 发送到Zep - try: - batch_result = self.client.graph.add_batch( - graph_id=graph_id, - episodes=episodes - ) - - # 收集返回的 episode uuid - if batch_result and isinstance(batch_result, list): - for ep in batch_result: - ep_uuid = getattr(ep, 'uuid_', None) or getattr(ep, 'uuid', None) - if ep_uuid: - episode_uuids.append(ep_uuid) - - # 避免请求过快 - time.sleep(1) - - except Exception as e: - if progress_callback: - progress_callback(t('progress.batchFailed', batch=batch_num, error=str(e)), 0) - raise - - return episode_uuids - - def _wait_for_episodes( - self, - episode_uuids: List[str], - progress_callback: Optional[Callable] = None, - timeout: int = 600 - ): - """等待所有 episode 处理完成（通过查询每个 episode 的 processed 状态）""" - if not episode_uuids: - if progress_callback: - progress_callback(t('progress.noEpisodesWait'), 1.0) - return - - start_time = time.time() - pending_episodes = set(episode_uuids) - completed_count = 0 - total_episodes = len(episode_uuids) - - if progress_callback: - progress_callback(t('progress.waitingEpisodes', count=total_episodes), 0) - - while pending_episodes: - if time.time() - start_time > timeout: - if progress_callback: - progress_callback( - t('progress.episodesTimeout', completed=completed_count, total=total_episodes), - completed_count / total_episodes - ) - break - - # 检查每个 episode 的处理状态 - for ep_uuid in list(pending_episodes): - try: - episode = self.client.graph.episode.get(uuid_=ep_uuid) - is_processed = getattr(episode, 'processed', False) - - if is_processed: - pending_episodes.remove(ep_uuid) - completed_count += 1 - - except Exception as e: - # 忽略单个查询错误，继续 - pass - - elapsed = int(time.time() - start_time) - if progress_callback: - progress_callback( - t('progress.zepProcessing', completed=completed_count, total=total_episodes, pending=len(pending_episodes), elapsed=elapsed), - completed_count / total_episodes if total_episodes > 0 else 0 - ) - - if pending_episodes: - time.sleep(3) # 每3秒检查一次 - - if progress_callback: - progress_callback(t('progress.processingComplete', completed=completed_count, total=total_episodes), 1.0) - + """分批添加文本到图谱，返回 episode IDs""" + self.client.add_episodes_batch( + graph_id=graph_id, + texts=chunks, + progress_callback=progress_callback, + ) + # Graphiti doesn't return episode UUIDs the same way + return [f"ep_{i}" for i in range(len(chunks))] + def _get_graph_info(self, graph_id: str) -> GraphInfo: """获取图谱信息""" - # 获取节点（分页） - nodes = fetch_all_nodes(self.client, graph_id) + nodes = self.client.get_all_nodes(graph_id) + edges = self.client.get_all_edges(graph_id) - # 获取边（分页） - edges = fetch_all_edges(self.client, graph_id) - - # 统计实体类型 entity_types = set() for node in nodes: if node.labels: for label in node.labels: - if label not in ["Entity", "Node"]: + if label not in ["Entity", "Node", "__Entity__"]: entity_types.add(label) return GraphInfo( @@ -422,76 +261,46 @@ class GraphBuilderService: edge_count=len(edges), entity_types=list(entity_types) ) - - def get_graph_data(self, graph_id: str) -> Dict[str, Any]: - """ - 获取完整图谱数据（包含详细信息） - - Args: - graph_id: 图谱ID - - Returns: - 包含nodes和edges的字典，包括时间信息、属性等详细数据 - """ - nodes = fetch_all_nodes(self.client, graph_id) - edges = fetch_all_edges(self.client, graph_id) - # 创建节点映射用于获取节点名称 + def get_graph_data(self, graph_id: str) -> Dict[str, Any]: + """获取完整图谱数据""" + nodes = self.client.get_all_nodes(graph_id) + edges = self.client.get_all_edges(graph_id) + node_map = {} for node in nodes: node_map[node.uuid_] = node.name or "" - + nodes_data = [] for node in nodes: - # 获取创建时间 - created_at = getattr(node, 'created_at', None) - if created_at: - created_at = str(created_at) - nodes_data.append({ "uuid": node.uuid_, "name": node.name, "labels": node.labels or [], "summary": node.summary or "", "attributes": node.attributes or {}, - "created_at": created_at, + "created_at": node.created_at, }) - + edges_data = [] for edge in edges: - # 获取时间信息 - created_at = getattr(edge, 'created_at', None) - valid_at = getattr(edge, 'valid_at', None) - invalid_at = getattr(edge, 'invalid_at', None) - expired_at = getattr(edge, 'expired_at', None) - - # 获取 episodes - episodes = getattr(edge, 'episodes', None) or getattr(edge, 'episode_ids', None) - if episodes and not isinstance(episodes, list): - episodes = [str(episodes)] - elif episodes: - episodes = [str(e) for e in episodes] - - # 获取 fact_type - fact_type = getattr(edge, 'fact_type', None) or edge.name or "" - edges_data.append({ "uuid": edge.uuid_, "name": edge.name or "", "fact": edge.fact or "", - "fact_type": fact_type, + "fact_type": edge.name or "", "source_node_uuid": edge.source_node_uuid, "target_node_uuid": edge.target_node_uuid, "source_node_name": node_map.get(edge.source_node_uuid, ""), "target_node_name": node_map.get(edge.target_node_uuid, ""), "attributes": edge.attributes or {}, - "created_at": str(created_at) if created_at else None, - "valid_at": str(valid_at) if valid_at else None, - "invalid_at": str(invalid_at) if invalid_at else None, - "expired_at": str(expired_at) if expired_at else None, - "episodes": episodes or [], + "created_at": edge.created_at, + "valid_at": edge.valid_at, + "invalid_at": edge.invalid_at, + "expired_at": edge.expired_at, + "episodes": edge.episodes or [], }) - + return { "graph_id": graph_id, "nodes": nodes_data, @@ -499,8 +308,7 @@ class GraphBuilderService: "node_count": len(nodes_data), "edge_count": len(edges_data), } - + def delete_graph(self, graph_id: str): """删除图谱""" - self.client.graph.delete(graph_id=graph_id) - + self.client.delete_graph(graph_id) diff --git a/backend/app/services/graphiti_client.py b/backend/app/services/graphiti_client.py new file mode 100644 index 00000000..1f615c1a --- /dev/null +++ b/backend/app/services/graphiti_client.py @@ -0,0 +1,371 @@ +""" +Graphiti 知识图谱客户端 +替代 Zep Cloud，使用自托管的 Graphiti + Neo4j + +Graphiti 是 async 库，本模块提供同步包装供 Flask 使用。 +""" + +import asyncio +import uuid +import time +from typing import Dict, Any, List, Optional +from datetime import datetime, timezone +from dataclasses import dataclass + +from neo4j import GraphDatabase + +from ..config import Config +from ..utils.logger import get_logger + +logger = get_logger('foresight.graphiti_client') + +# Global event loop for async bridge +_loop: Optional[asyncio.AbstractEventLoop] = None + + +def _get_loop() -> asyncio.AbstractEventLoop: + """Get or create a dedicated event loop for Graphiti async calls.""" + global _loop + if _loop is None or _loop.is_closed(): + _loop = asyncio.new_event_loop() + return _loop + + +def _run_async(coro): + """Run an async coroutine synchronously.""" + loop = _get_loop() + return loop.run_until_complete(coro) + + +@dataclass +class GraphitiNode: + """Node data from Neo4j, compatible with Zep node format.""" + uuid_: str + name: str + labels: List[str] + summary: str + attributes: Dict[str, Any] + created_at: Optional[str] = None + + @property + def uuid(self): + return self.uuid_ + + +@dataclass +class GraphitiEdge: + """Edge data from Neo4j, compatible with Zep edge format.""" + uuid_: str + name: str + fact: str + source_node_uuid: str + target_node_uuid: str + attributes: Dict[str, Any] + created_at: Optional[str] = None + valid_at: Optional[str] = None + invalid_at: Optional[str] = None + expired_at: Optional[str] = None + episodes: Optional[List[str]] = None + + @property + def uuid(self): + return self.uuid_ + + +class GraphitiClient: + """ + Graphiti + Neo4j 知识图谱客户端 + + 提供与原 Zep Cloud 兼容的接口: + - create_graph / delete_graph + - add_episodes (文本导入) + - search (语义搜索) + - get_all_nodes / get_all_edges + - get_node / get_node_edges + """ + + _instance: Optional['GraphitiClient'] = None + _graphiti = None + _initialized = False + + def __init__( + self, + neo4j_uri: Optional[str] = None, + neo4j_user: Optional[str] = None, + neo4j_password: Optional[str] = None, + ): + self.neo4j_uri = neo4j_uri or Config.NEO4J_URI + self.neo4j_user = neo4j_user or Config.NEO4J_USER + self.neo4j_password = neo4j_password or Config.NEO4J_PASSWORD + + # Neo4j driver for direct queries + self._driver = GraphDatabase.driver( + self.neo4j_uri, + auth=(self.neo4j_user, self.neo4j_password), + ) + logger.info(f"GraphitiClient initialized: {self.neo4j_uri}") + + def _ensure_graphiti(self): + """Lazy-init Graphiti (imports are heavy).""" + if self._graphiti is not None: + return + + from graphiti_core import Graphiti + from graphiti_core.llm_client.openai_generic_client import OpenAIGenericClient + from graphiti_core.llm_client.config import LLMConfig + + llm_config = LLMConfig( + api_key=Config.LLM_API_KEY, + model=Config.LLM_MODEL_NAME, + small_model=Config.LLM_MODEL_NAME, + base_url=Config.LLM_BASE_URL, + ) + llm_client = OpenAIGenericClient(config=llm_config) + + self._graphiti = Graphiti( + self.neo4j_uri, + self.neo4j_user, + self.neo4j_password, + llm_client=llm_client, + ) + # Build indices + _run_async(self._graphiti.build_indices_and_constraints()) + self._initialized = True + logger.info("Graphiti core initialized with indices") + + @classmethod + def get_instance(cls) -> 'GraphitiClient': + """Singleton accessor.""" + if cls._instance is None: + cls._instance = cls() + return cls._instance + + # ========== Graph CRUD ========== + + def create_graph(self, name: str) -> str: + """Create a logical graph (just returns a group_id, Neo4j doesn't need explicit creation).""" + graph_id = f"foresight_{uuid.uuid4().hex[:16]}" + logger.info(f"Created graph group: {graph_id} ({name})") + return graph_id + + def delete_graph(self, graph_id: str): + """Delete all nodes and edges belonging to a graph group.""" + with self._driver.session() as session: + # Delete edges first, then nodes + session.run( + "MATCH (a)-[r]-(b) WHERE r.group_id = $gid DELETE r", + gid=graph_id, + ) + session.run( + "MATCH (n) WHERE n.group_id = $gid DETACH DELETE n", + gid=graph_id, + ) + logger.info(f"Deleted graph: {graph_id}") + + # ========== Episode ingestion ========== + + def add_episode( + self, + graph_id: str, + text: str, + source_description: str = "document", + entity_types: Optional[Dict] = None, + edge_types: Optional[Dict] = None, + ): + """Add a single text episode to the graph.""" + self._ensure_graphiti() + + kwargs = { + "name": f"episode_{uuid.uuid4().hex[:8]}", + "episode_body": text, + "source_description": source_description, + "reference_time": datetime.now(timezone.utc), + "group_id": graph_id, + } + if entity_types: + kwargs["entity_types"] = entity_types + if edge_types: + kwargs["edge_types"] = edge_types + + from graphiti_core.nodes import EpisodeType + kwargs["source"] = EpisodeType.text + + _run_async(self._graphiti.add_episode(**kwargs)) + + def add_episodes_batch( + self, + graph_id: str, + texts: List[str], + source_description: str = "document", + entity_types: Optional[Dict] = None, + edge_types: Optional[Dict] = None, + progress_callback=None, + ): + """Add multiple text episodes sequentially (Graphiti processes one at a time).""" + total = len(texts) + for i, text in enumerate(texts): + if progress_callback: + progress_callback( + f"Processing episode {i + 1}/{total}", + (i + 1) / total, + ) + try: + self.add_episode( + graph_id, text, source_description, + entity_types, edge_types, + ) + except Exception as e: + logger.error(f"Failed to add episode {i + 1}/{total}: {e}") + raise + # Small delay to avoid rate limiting + if i < total - 1: + time.sleep(0.5) + + # ========== Search ========== + + def search( + self, + query: str, + graph_id: Optional[str] = None, + limit: int = 10, + ) -> List[Dict[str, Any]]: + """Search the graph for relevant edges/facts.""" + self._ensure_graphiti() + + kwargs = {"query": query, "num_results": limit} + if graph_id: + kwargs["group_ids"] = [graph_id] + + results = _run_async(self._graphiti.search(**kwargs)) + + facts = [] + for edge in results: + facts.append({ + "uuid": str(getattr(edge, 'uuid', '')), + "fact": getattr(edge, 'fact', ''), + "name": getattr(edge, 'name', ''), + "source_node_uuid": str(getattr(edge, 'source_node_uuid', '')), + "target_node_uuid": str(getattr(edge, 'target_node_uuid', '')), + }) + return facts + + # ========== Node/Edge queries (direct Neo4j) ========== + + def get_all_nodes(self, graph_id: str, limit: int = 2000) -> List[GraphitiNode]: + """Get all entity nodes for a graph group.""" + with self._driver.session() as session: + result = session.run( + """ + MATCH (n:Entity) + WHERE n.group_id = $gid + RETURN n, labels(n) as labels + ORDER BY n.name + LIMIT $limit + """, + gid=graph_id, + limit=limit, + ) + nodes = [] + for record in result: + n = record["n"] + raw_labels = record["labels"] + # Filter internal labels + labels = [l for l in raw_labels if l not in ("__Entity__",)] + nodes.append(GraphitiNode( + uuid_=str(n.get("uuid", n.element_id)), + name=n.get("name", ""), + labels=labels, + summary=n.get("summary", ""), + attributes=dict(n) if n else {}, + created_at=str(n.get("created_at", "")) if n.get("created_at") else None, + )) + return nodes + + def get_all_edges(self, graph_id: str) -> List[GraphitiEdge]: + """Get all edges for a graph group.""" + with self._driver.session() as session: + result = session.run( + """ + MATCH (a)-[r:RELATES_TO]->(b) + WHERE r.group_id = $gid + RETURN r, a.uuid as source_uuid, b.uuid as target_uuid + """, + gid=graph_id, + ) + edges = [] + for record in result: + r = record["r"] + edges.append(GraphitiEdge( + uuid_=str(r.get("uuid", r.element_id)), + name=r.get("name", ""), + fact=r.get("fact", ""), + source_node_uuid=str(record["source_uuid"] or ""), + target_node_uuid=str(record["target_uuid"] or ""), + attributes=dict(r) if r else {}, + created_at=str(r.get("created_at", "")) if r.get("created_at") else None, + valid_at=str(r.get("valid_at", "")) if r.get("valid_at") else None, + invalid_at=str(r.get("invalid_at", "")) if r.get("invalid_at") else None, + expired_at=str(r.get("expired_at", "")) if r.get("expired_at") else None, + )) + return edges + + def get_node(self, node_uuid: str) -> Optional[GraphitiNode]: + """Get a single node by UUID.""" + with self._driver.session() as session: + result = session.run( + """ + MATCH (n:Entity {uuid: $uuid}) + RETURN n, labels(n) as labels + """, + uuid=node_uuid, + ) + record = result.single() + if not record: + return None + n = record["n"] + labels = [l for l in record["labels"] if l not in ("__Entity__",)] + return GraphitiNode( + uuid_=str(n.get("uuid", n.element_id)), + name=n.get("name", ""), + labels=labels, + summary=n.get("summary", ""), + attributes=dict(n) if n else {}, + created_at=str(n.get("created_at", "")) if n.get("created_at") else None, + ) + + def get_node_edges(self, node_uuid: str) -> List[GraphitiEdge]: + """Get all edges connected to a specific node.""" + with self._driver.session() as session: + result = session.run( + """ + MATCH (a)-[r:RELATES_TO]-(b) + WHERE a.uuid = $uuid + RETURN r, + CASE WHEN startNode(r) = a THEN a.uuid ELSE b.uuid END as source_uuid, + CASE WHEN startNode(r) = a THEN b.uuid ELSE a.uuid END as target_uuid + """, + uuid=node_uuid, + ) + edges = [] + for record in result: + r = record["r"] + edges.append(GraphitiEdge( + uuid_=str(r.get("uuid", r.element_id)), + name=r.get("name", ""), + fact=r.get("fact", ""), + source_node_uuid=str(record["source_uuid"] or ""), + target_node_uuid=str(record["target_uuid"] or ""), + attributes=dict(r) if r else {}, + )) + return edges + + def close(self): + """Close connections.""" + if self._driver: + self._driver.close() + if self._graphiti: + try: + _run_async(self._graphiti.close()) + except Exception: + pass + logger.info("GraphitiClient closed") diff --git a/backend/app/services/oasis_profile_generator.py b/backend/app/services/oasis_profile_generator.py index bb6a4690..dd7b3e55 100644 --- a/backend/app/services/oasis_profile_generator.py +++ b/backend/app/services/oasis_profile_generator.py @@ -16,7 +16,7 @@ from dataclasses import dataclass, field from datetime import datetime from openai import OpenAI -from zep_cloud.client import Zep +from .graphiti_client import GraphitiClient from ..config import Config from ..utils.logger import get_logger @@ -198,16 +198,13 @@ class OasisProfileGenerator: base_url=self.base_url ) - # Zep客户端用于检索丰富上下文 - self.zep_api_key = zep_api_key or Config.ZEP_API_KEY - self.zep_client = None + # Graphiti 客户端用于检索丰富上下文 self.graph_id = graph_id - - if self.zep_api_key: - try: - self.zep_client = Zep(api_key=self.zep_api_key) - except Exception as e: - logger.warning(f"Zep客户端初始化失败: {e}") + try: + self.zep_client = GraphitiClient.get_instance() + except Exception as e: + logger.warning(f"Graphiti客户端初始化失败: {e}") + self.zep_client = None def generate_profile_from_entity( self, @@ -317,53 +314,19 @@ class OasisProfileGenerator: comprehensive_query = t('progress.zepSearchQuery', name=entity_name) def search_edges(): - """搜索边（事实/关系）- 带重试机制""" - max_retries = 3 - last_exception = None - delay = 2.0 - - for attempt in range(max_retries): - try: - return self.zep_client.graph.search( - query=comprehensive_query, - graph_id=self.graph_id, - limit=30, - scope="edges", - reranker="rrf" - ) - except Exception as e: - last_exception = e - if attempt < max_retries - 1: - logger.debug(f"Zep边搜索第 {attempt + 1} 次失败: {str(e)[:80]}, 重试中...") - time.sleep(delay) - delay *= 2 - else: - logger.debug(f"Zep边搜索在 {max_retries} 次尝试后仍失败: {e}") - return None - + """搜索边（事实/关系）- 使用 Graphiti""" + try: + return self.zep_client.search( + query=comprehensive_query, + graph_id=self.graph_id, + limit=30, + ) + except Exception as e: + logger.debug(f"Graphiti 搜索失败: {str(e)[:80]}") + return None + def search_nodes(): - """搜索节点（实体摘要）- 带重试机制""" - max_retries = 3 - last_exception = None - delay = 2.0 - - for attempt in range(max_retries): - try: - return self.zep_client.graph.search( - query=comprehensive_query, - graph_id=self.graph_id, - limit=20, - scope="nodes", - reranker="rrf" - ) - except Exception as e: - last_exception = e - if attempt < max_retries - 1: - logger.debug(f"Zep节点搜索第 {attempt + 1} 次失败: {str(e)[:80]}, 重试中...") - time.sleep(delay) - delay *= 2 - else: - logger.debug(f"Zep节点搜索在 {max_retries} 次尝试后仍失败: {e}") + """搜索节点 - 返回 None，Graphiti search 已包含相关信息""" return None try: diff --git a/backend/app/services/zep_entity_reader.py b/backend/app/services/zep_entity_reader.py index 82e0d5f5..75ef31a0 100644 --- a/backend/app/services/zep_entity_reader.py +++ b/backend/app/services/zep_entity_reader.py @@ -7,11 +7,9 @@ import time from typing import Dict, Any, List, Optional, Set, Callable, TypeVar from dataclasses import dataclass, field -from zep_cloud.client import Zep - from ..config import Config from ..utils.logger import get_logger -from ..utils.zep_paging import fetch_all_nodes, fetch_all_edges +from .graphiti_client import GraphitiClient logger = get_logger('foresight.zep_entity_reader') @@ -79,11 +77,8 @@ class ZepEntityReader: """ def __init__(self, api_key: Optional[str] = None): - self.api_key = api_key or Config.ZEP_API_KEY - if not self.api_key: - raise ValueError("ZEP_API_KEY 未配置") - - self.client = Zep(api_key=self.api_key) + # api_key kept for interface compatibility + self.client = GraphitiClient.get_instance() def _call_with_retry( self, @@ -136,12 +131,12 @@ class ZepEntityReader: """ logger.info(f"获取图谱 {graph_id} 的所有节点...") - nodes = fetch_all_nodes(self.client, graph_id) + nodes = self.client.get_all_nodes(graph_id) nodes_data = [] for node in nodes: nodes_data.append({ - "uuid": getattr(node, 'uuid_', None) or getattr(node, 'uuid', ''), + "uuid": node.uuid_, "name": node.name or "", "labels": node.labels or [], "summary": node.summary or "", @@ -163,12 +158,12 @@ class ZepEntityReader: """ logger.info(f"获取图谱 {graph_id} 的所有边...") - edges = fetch_all_edges(self.client, graph_id) + edges = self.client.get_all_edges(graph_id) edges_data = [] for edge in edges: edges_data.append({ - "uuid": getattr(edge, 'uuid_', None) or getattr(edge, 'uuid', ''), + "uuid": edge.uuid_, "name": edge.name or "", "fact": edge.fact or "", "source_node_uuid": edge.source_node_uuid, @@ -190,23 +185,19 @@ class ZepEntityReader: 边列表 """ try: - # 使用重试机制调用Zep API - edges = self._call_with_retry( - func=lambda: self.client.graph.node.get_entity_edges(node_uuid=node_uuid), - operation_name=f"获取节点边(node={node_uuid[:8]}...)" - ) - + edges = self.client.get_node_edges(node_uuid) + edges_data = [] for edge in edges: edges_data.append({ - "uuid": getattr(edge, 'uuid_', None) or getattr(edge, 'uuid', ''), + "uuid": edge.uuid_, "name": edge.name or "", "fact": edge.fact or "", "source_node_uuid": edge.source_node_uuid, "target_node_uuid": edge.target_node_uuid, "attributes": edge.attributes or {}, }) - + return edges_data except Exception as e: logger.warning(f"获取节点 {node_uuid} 的边失败: {str(e)}") @@ -346,11 +337,7 @@ class ZepEntityReader: EntityNode或None """ try: - # 使用重试机制获取节点 - node = self._call_with_retry( - func=lambda: self.client.graph.node.get(uuid_=entity_uuid), - operation_name=f"获取节点详情(uuid={entity_uuid[:8]}...)" - ) + node = self.client.get_node(entity_uuid) if not node: return None @@ -397,7 +384,7 @@ class ZepEntityReader: }) return EntityNode( - uuid=getattr(node, 'uuid_', None) or getattr(node, 'uuid', ''), + uuid=node.uuid_, name=node.name or "", labels=node.labels or [], summary=node.summary or "", diff --git a/backend/app/services/zep_graph_memory_updater.py b/backend/app/services/zep_graph_memory_updater.py index 6b3a8627..3674c576 100644 --- a/backend/app/services/zep_graph_memory_updater.py +++ b/backend/app/services/zep_graph_memory_updater.py @@ -12,11 +12,10 @@ from dataclasses import dataclass from datetime import datetime from queue import Queue, Empty -from zep_cloud.client import Zep - from ..config import Config from ..utils.logger import get_logger from ..utils.locale import get_locale, set_locale +from .graphiti_client import GraphitiClient logger = get_logger('foresight.zep_graph_memory_updater') @@ -238,12 +237,7 @@ class ZepGraphMemoryUpdater: api_key: Zep API Key（可选，默认从配置读取） """ self.graph_id = graph_id - self.api_key = api_key or Config.ZEP_API_KEY - - if not self.api_key: - raise ValueError("ZEP_API_KEY未配置") - - self.client = Zep(api_key=self.api_key) + self.client = GraphitiClient.get_instance() # 活动队列 self._activity_queue: Queue = Queue() @@ -411,10 +405,10 @@ class ZepGraphMemoryUpdater: # 带重试的发送 for attempt in range(self.MAX_RETRIES): try: - self.client.graph.add( + self.client.add_episode( graph_id=self.graph_id, - type="text", - data=combined_text + text=combined_text, + source_description=f"simulation_{platform}_activity", ) self._total_sent += 1 diff --git a/backend/app/services/zep_tools.py b/backend/app/services/zep_tools.py index 398e4c74..bcbd4dd6 100644 --- a/backend/app/services/zep_tools.py +++ b/backend/app/services/zep_tools.py @@ -13,13 +13,11 @@ import json from typing import Dict, Any, List, Optional from dataclasses import dataclass, field -from zep_cloud.client import Zep - from ..config import Config from ..utils.logger import get_logger from ..utils.llm_client import LLMClient from ..utils.locale import get_locale, t -from ..utils.zep_paging import fetch_all_nodes, fetch_all_edges +from .graphiti_client import GraphitiClient logger = get_logger('foresight.zep_tools') @@ -423,12 +421,8 @@ class ZepToolsService: RETRY_DELAY = 2.0 def __init__(self, api_key: Optional[str] = None, llm_client: Optional[LLMClient] = None): - self.api_key = api_key or Config.ZEP_API_KEY - if not self.api_key: - raise ValueError("ZEP_API_KEY 未配置") - - self.client = Zep(api_key=self.api_key) - # LLM客户端用于InsightForge生成子问题 + # api_key kept for interface compatibility + self.client = GraphitiClient.get_instance() self._llm_client = llm_client logger.info(t("console.zepToolsInitialized")) @@ -485,62 +479,34 @@ class ZepToolsService: """ logger.info(t("console.graphSearch", graphId=graph_id, query=query[:50])) - # 尝试使用Zep Cloud Search API try: - search_results = self._call_with_retry( - func=lambda: self.client.graph.search( - graph_id=graph_id, - query=query, - limit=limit, - scope=scope, - reranker="cross_encoder" - ), - operation_name=t("console.graphSearchOp", graphId=graph_id) + search_results = self.client.search( + query=query, + graph_id=graph_id, + limit=limit, ) - + facts = [] edges = [] - nodes = [] - - # 解析边搜索结果 - if hasattr(search_results, 'edges') and search_results.edges: - for edge in search_results.edges: - if hasattr(edge, 'fact') and edge.fact: - facts.append(edge.fact) - edges.append({ - "uuid": getattr(edge, 'uuid_', None) or getattr(edge, 'uuid', ''), - "name": getattr(edge, 'name', ''), - "fact": getattr(edge, 'fact', ''), - "source_node_uuid": getattr(edge, 'source_node_uuid', ''), - "target_node_uuid": getattr(edge, 'target_node_uuid', ''), - }) - - # 解析节点搜索结果 - if hasattr(search_results, 'nodes') and search_results.nodes: - for node in search_results.nodes: - nodes.append({ - "uuid": getattr(node, 'uuid_', None) or getattr(node, 'uuid', ''), - "name": getattr(node, 'name', ''), - "labels": getattr(node, 'labels', []), - "summary": getattr(node, 'summary', ''), - }) - # 节点摘要也算作事实 - if hasattr(node, 'summary') and node.summary: - facts.append(f"[{node.name}]: {node.summary}") - + + for item in search_results: + fact = item.get("fact", "") + if fact: + facts.append(fact) + edges.append(item) + logger.info(t("console.searchComplete", count=len(facts))) - + return SearchResult( facts=facts, edges=edges, - nodes=nodes, + nodes=[], query=query, total_count=len(facts) ) - + except Exception as e: logger.warning(t("console.zepSearchApiFallback", error=str(e))) - # 降级：使用本地关键词匹配搜索 return self._local_search(graph_id, query, limit, scope) def _local_search( @@ -659,13 +625,12 @@ class ZepToolsService: """ logger.info(t("console.fetchingAllNodes", graphId=graph_id)) - nodes = fetch_all_nodes(self.client, graph_id) + nodes = self.client.get_all_nodes(graph_id) result = [] for node in nodes: - node_uuid = getattr(node, 'uuid_', None) or getattr(node, 'uuid', None) or "" result.append(NodeInfo( - uuid=str(node_uuid) if node_uuid else "", + uuid=node.uuid_, name=node.name or "", labels=node.labels or [], summary=node.summary or "", @@ -688,25 +653,23 @@ class ZepToolsService: """ logger.info(t("console.fetchingAllEdges", graphId=graph_id)) - edges = fetch_all_edges(self.client, graph_id) + edges = self.client.get_all_edges(graph_id) result = [] for edge in edges: - edge_uuid = getattr(edge, 'uuid_', None) or getattr(edge, 'uuid', None) or "" edge_info = EdgeInfo( - uuid=str(edge_uuid) if edge_uuid else "", + uuid=edge.uuid_, name=edge.name or "", fact=edge.fact or "", source_node_uuid=edge.source_node_uuid or "", target_node_uuid=edge.target_node_uuid or "" ) - # 添加时间信息 if include_temporal: - edge_info.created_at = getattr(edge, 'created_at', None) - edge_info.valid_at = getattr(edge, 'valid_at', None) - edge_info.invalid_at = getattr(edge, 'invalid_at', None) - edge_info.expired_at = getattr(edge, 'expired_at', None) + edge_info.created_at = edge.created_at + edge_info.valid_at = edge.valid_at + edge_info.invalid_at = edge.invalid_at + edge_info.expired_at = edge.expired_at result.append(edge_info) @@ -726,16 +689,13 @@ class ZepToolsService: logger.info(t("console.fetchingNodeDetail", uuid=node_uuid[:8])) try: - node = self._call_with_retry( - func=lambda: self.client.graph.node.get(uuid_=node_uuid), - operation_name=t("console.fetchNodeDetailOp", uuid=node_uuid[:8]) - ) - + node = self.client.get_node(node_uuid) + if not node: return None - + return NodeInfo( - uuid=getattr(node, 'uuid_', None) or getattr(node, 'uuid', ''), + uuid=node.uuid_, name=node.name or "", labels=node.labels or [], summary=node.summary or "", diff --git a/backend/requirements.txt b/backend/requirements.txt index 4f146296..365d96f8 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -13,8 +13,9 @@ flask-cors>=6.0.0 # OpenAI SDK（统一使用 OpenAI 格式调用 LLM） openai>=1.0.0 -# ============= Zep Cloud ============= -zep-cloud==3.13.0 +# ============= Knowledge Graph (Graphiti + Neo4j) ============= +graphiti-core>=0.5.0 +neo4j>=5.0.0 # ============= OASIS 社交媒体模拟 ============= # OASIS 社交模拟框架