feat: migrate from Zep Cloud to self-hosted Graphiti + Neo4j
Replace Zep Cloud ($25/mo) with open-source Graphiti + Neo4j: - New graphiti_client.py: unified wrapper with sync bridge for async Graphiti - Modified graph_builder.py: use Graphiti add_episode instead of Zep batch API - Modified zep_entity_reader.py: Neo4j Cypher queries replace Zep pagination - Modified zep_tools.py: Graphiti search replaces Zep Cloud search - Modified zep_graph_memory_updater.py: Graphiti add_episode replaces Zep add - Modified oasis_profile_generator.py: GraphitiClient replaces Zep client - Updated config.py: NEO4J_URI/USER/PASSWORD replace ZEP_API_KEY - Updated requirements.txt: graphiti-core + neo4j replace zep-cloud - Added PRD.md: product requirements document with token analysis Generated with [Claude Code](https://claude.ai/code) via [Happy](https://happy.engineering) Co-Authored-By: Claude <noreply@anthropic.com> Co-Authored-By: Happy <yesreply@happy.engineering>
This commit is contained in:
liyizhouAI
parent
4b682f24f4
commit
fff7edce2a
|
|
@ -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/<task_id>` — 构建进度查询
|
||||
|
||||
**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/<sim_id>` — 查询状态
|
||||
- `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=<MiniMax API Key>
|
||||
LLM_BASE_URL=https://api.minimax.chat/v1
|
||||
LLM_MODEL_NAME=MiniMax-M2.5
|
||||
|
||||
# Zep 配置
|
||||
ZEP_API_KEY=<Zep Cloud 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 和配置 |
|
||||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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 "",
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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 "",
|
||||
|
|
|
|||
|
|
@ -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 社交模拟框架
|
||||
|
|
|
|||
Loading…
Reference in New Issue