diff --git a/docs/latest/intro/knowledge-base.md b/docs/latest/intro/knowledge-base.md
index d0621ee5..995501fa 100644
--- a/docs/latest/intro/knowledge-base.md
+++ b/docs/latest/intro/knowledge-base.md
@@ -97,16 +97,32 @@ LIMIT $num
### 1. 以三元组形式导入
+系统支持通过网页导入 `jsonl` 格式的知识图谱数据,支持**简单三元组**和**带属性三元组**两种格式。
-系统支持通过网页导入 `jsonl` 格式的知识图谱数据:
+**简单格式(兼容旧版)**:
```jsonl
{"h": "北京", "t": "中国", "r": "首都"}
{"h": "上海", "t": "中国", "r": "直辖市"}
-{"h": "深圳", "t": "广东", "r": "省会"}
```
-**格式说明**,每行一个三元组,系统自动验证数据格式,并自动导入到 Neo4j 数据库,添加 `Upload`、`Entity`、`Relation` 标签,会自动处理重复的三元组。
+**扩展格式(支持属性)**:
+
+支持 `h`(头节点)、`t`(尾节点)和 `r`(关系)为对象结构,其中:
+- 节点对象必须包含 `name` 字段。
+- 关系对象必须包含 `type` 字段。
+- 其他字段将作为**属性**存储在 Neo4j 中。
+
+```jsonl
+{"h": {"name": "孙悟空", "title": "齐天大圣", "weapon": "如意金箍棒"}, "t": {"name": "唐僧", "species": "人"}, "r": {"type": "徒弟", "order": 1}}
+{"h": "猪八戒", "t": {"name": "唐僧"}, "r": {"type": "徒弟", "order": 2}}
+```
+
+**格式说明**:
+- 每行一个数据项。
+- 系统自动验证数据格式,并自动导入到 Neo4j 数据库。
+- 自动添加 `Upload`、`Entity` 标签(节点)和 `RELATION` 类型(关系)。
+- 自动处理重复实体和关系,并合并属性。
Neo4j 访问信息可以参考 `docker-compose.yml` 中配置对应的环境变量来覆盖。
@@ -116,7 +132,9 @@ Neo4j 访问信息可以参考 `docker-compose.yml` 中配置对应的环境变
- **连接地址**: bolt://localhost:7687
::: tip 测试数据
-可以使用 `test/data/A_Dream_of_Red_Mansions_tiny.jsonl` 文件进行测试导入。
+可以使用以下文件进行测试导入:
+- 简单格式:`test/data/A_Dream_of_Red_Mansions_tiny.jsonl`
+- 扩展属性格式:`test/data/complex_graph_test.jsonl`
:::
### 2. 接入已有 Neo4j 实例
diff --git a/server/routers/graph_router.py b/server/routers/graph_router.py
index 3c024650..552e16fb 100644
--- a/server/routers/graph_router.py
+++ b/server/routers/graph_router.py
@@ -5,13 +5,213 @@ from fastapi import APIRouter, Body, Depends, HTTPException, Query
from src.storage.db.models import User
from server.utils.auth_middleware import get_admin_user
from src import graph_base, knowledge_base
+from src.knowledge.adapters.factory import GraphAdapterFactory
+from src.knowledge.adapters.base import GraphAdapter
from src.utils.logging_config import logger
graph = APIRouter(prefix="/graph", tags=["graph"])
# =============================================================================
-# === 子图查询分组 ===
+# === 统一图谱接口 (Unified Graph API) ===
+# =============================================================================
+
+
+async def _get_graph_adapter(db_id: str) -> GraphAdapter:
+ """
+ 根据数据库ID获取对应的图谱适配器
+
+ Args:
+ db_id: 数据库ID
+
+ Returns:
+ GraphAdapter: 对应的图谱适配器实例
+ """
+ # 1. 检查是否是 LightRAG 数据库
+ if knowledge_base.is_lightrag_database(db_id):
+ rag_instance = await knowledge_base._get_lightrag_instance(db_id)
+ if not rag_instance:
+ raise HTTPException(status_code=404, detail=f"LightRAG database {db_id} not found or inaccessible")
+ return GraphAdapterFactory.create_adapter("lightrag", lightrag_instance=rag_instance)
+
+ # 2. 默认为 Upload/Neo4j 数据库 (假设 db_id 为 "neo4j" 或其他 Neo4j 数据库名)
+ # 这里我们假设非 LightRAG 的 ID 都是 Neo4j 的数据库名
+ # 如果未来有更多类型,需要更完善的 ID 区分机制 (例如前缀)
+ if not graph_base.is_running():
+ raise HTTPException(status_code=503, detail="Graph database service is not running")
+
+ return GraphAdapterFactory.create_adapter("upload", graph_db_instance=graph_base, config={"kgdb_name": db_id})
+
+
+@graph.get("/list")
+async def get_graphs(current_user: User = Depends(get_admin_user)):
+ """
+ 获取所有可用的知识图谱列表
+
+ Returns:
+ 包含所有图谱信息的列表 (包括 Neo4j 和 LightRAG)
+ """
+ try:
+ graphs = []
+
+ # 1. 获取默认 Neo4j 图谱信息
+ neo4j_info = graph_base.get_graph_info()
+ if neo4j_info:
+ graphs.append(
+ {
+ "id": "neo4j",
+ "name": "默认图谱",
+ "type": "neo4j",
+ "description": "Default graph database for uploaded documents",
+ "status": neo4j_info.get("status", "unknown"),
+ "created_at": neo4j_info.get("last_updated"),
+ "node_count": neo4j_info.get("entity_count", 0),
+ "edge_count": neo4j_info.get("relationship_count", 0),
+ }
+ )
+
+ # 2. 获取 LightRAG 数据库信息
+ lightrag_dbs = knowledge_base.get_lightrag_databases()
+ for db in lightrag_dbs:
+ graphs.append(
+ {
+ "id": db.get("db_id"),
+ "name": db.get("name"),
+ "type": "lightrag",
+ "description": db.get("description"),
+ "status": "active", # LightRAG DBs are usually active if listed
+ "created_at": db.get("created_at"),
+ "metadata": db,
+ }
+ )
+
+ return {"success": True, "data": graphs}
+
+ except Exception as e:
+ logger.error(f"Failed to list graphs: {e}")
+ logger.error(f"Traceback: {traceback.format_exc()}")
+ raise HTTPException(status_code=500, detail=f"Failed to list graphs: {str(e)}")
+
+
+@graph.get("/subgraph")
+async def get_subgraph(
+ db_id: str = Query(..., description="知识图谱ID"),
+ node_label: str = Query("*", description="节点标签或查询关键词"),
+ max_depth: int = Query(2, description="最大深度", ge=1, le=5),
+ max_nodes: int = Query(100, description="最大节点数", ge=1, le=1000),
+ current_user: User = Depends(get_admin_user),
+):
+ """
+ 统一的子图查询接口
+
+ Args:
+ db_id: 图谱ID (LightRAG DB ID 或 "neo4j")
+ node_label: 查询关键词或标签
+ max_depth: 扩展深度
+ max_nodes: 返回最大节点数
+ """
+ try:
+ logger.info(f"Querying subgraph - db_id: {db_id}, label: {node_label}")
+
+ adapter = await _get_graph_adapter(db_id)
+
+ # 统一查询参数
+ # 对于 UploadGraphAdapter, kgdb_name 通常通过 kwargs 传递
+ # 对于 LightRAGGraphAdapter, max_depth/max_nodes 通过 kwargs 传递
+ result_data = await adapter.query_nodes(
+ keyword=node_label,
+ max_depth=max_depth,
+ max_nodes=max_nodes,
+ kgdb_name=db_id if not knowledge_base.is_lightrag_database(db_id) else "neo4j",
+ )
+
+ return {
+ "success": True,
+ "data": result_data,
+ }
+
+ except HTTPException:
+ raise
+ except Exception as e:
+ logger.error(f"Failed to get subgraph: {e}")
+ logger.error(f"Traceback: {traceback.format_exc()}")
+ raise HTTPException(status_code=500, detail=f"Failed to get subgraph: {str(e)}")
+
+
+@graph.get("/labels")
+async def get_graph_labels(
+ db_id: str = Query(..., description="知识图谱ID"), current_user: User = Depends(get_admin_user)
+):
+ """
+ 获取图谱的所有标签
+ """
+ try:
+ adapter = await _get_graph_adapter(db_id)
+ labels = await adapter.get_labels()
+ return {"success": True, "data": {"labels": labels}}
+
+ except Exception as e:
+ logger.error(f"Failed to get labels: {e}")
+ raise HTTPException(status_code=500, detail=f"Failed to get labels: {str(e)}")
+
+
+@graph.get("/stats")
+async def get_graph_stats(
+ db_id: str = Query(..., description="知识图谱ID"), current_user: User = Depends(get_admin_user)
+):
+ """
+ 获取图谱统计信息
+ """
+ try:
+ if knowledge_base.is_lightrag_database(db_id):
+ # 复用原有的 LightRAG 统计逻辑
+ # 这里暂时直接调用原有逻辑,理想情况下也应该封装进 Adapter
+ rag_instance = await knowledge_base._get_lightrag_instance(db_id)
+ if not rag_instance:
+ raise HTTPException(status_code=404, detail="Database not found")
+
+ knowledge_graph = await rag_instance.get_knowledge_graph(node_label="*", max_depth=1, max_nodes=10000)
+ entity_types = {}
+ for node in knowledge_graph.nodes:
+ entity_type = node.properties.get("entity_type", "unknown")
+ entity_types[entity_type] = entity_types.get(entity_type, 0) + 1
+
+ entity_types_list = [
+ {"type": k, "count": v} for k, v in sorted(entity_types.items(), key=lambda x: x[1], reverse=True)
+ ]
+
+ return {
+ "success": True,
+ "data": {
+ "total_nodes": len(knowledge_graph.nodes),
+ "total_edges": len(knowledge_graph.edges),
+ "entity_types": entity_types_list,
+ },
+ }
+ else:
+ # Neo4j stats
+ info = graph_base.get_graph_info(graph_name=db_id)
+ if not info:
+ raise HTTPException(status_code=404, detail="Graph info not found")
+
+ return {
+ "success": True,
+ "data": {
+ "total_nodes": info.get("entity_count", 0),
+ "total_edges": info.get("relationship_count", 0),
+ # Neo4j info currently returns 'labels' list, not counts per label.
+ # Improving this would require updating GraphDatabase.get_graph_info
+ "entity_types": [{"type": label, "count": "N/A"} for label in info.get("labels", [])],
+ },
+ }
+
+ except Exception as e:
+ logger.error(f"Failed to get stats: {e}")
+ raise HTTPException(status_code=500, detail=f"Failed to get stats: {str(e)}")
+
+
+# =============================================================================
+# === 兼容性接口 (Deprecated/Compatibility) ===
# =============================================================================
@@ -23,147 +223,36 @@ async def get_lightrag_subgraph(
max_nodes: int = Query(100, description="最大节点数", ge=1, le=1000),
current_user: User = Depends(get_admin_user),
):
- """
- 使用 LightRAG 原生方法获取知识图谱子图
-
- Args:
- db_id: LightRAG 数据库实例ID
- node_label: 节点标签,用于查找起始节点,使用 "*" 获取全图
- max_depth: 子图的最大深度
- max_nodes: 返回的最大节点数量
-
- Returns:
- 包含节点和边的知识图谱数据
- """
- try:
- logger.info(
- f"获取子图数据 - db_id: {db_id}, node_label: {node_label}, max_depth: {max_depth}, max_nodes: {max_nodes}"
- )
-
- # 检查是否是 LightRAG 数据库
- if not knowledge_base.is_lightrag_database(db_id):
- raise HTTPException(
- status_code=400, detail=f"数据库 {db_id} 不是 LightRAG 类型,图谱功能仅支持 LightRAG 知识库"
- )
-
- # 获取 LightRAG 实例
- rag_instance = await knowledge_base._get_lightrag_instance(db_id)
- if not rag_instance:
- raise HTTPException(status_code=404, detail=f"LightRAG 数据库 {db_id} 不存在或无法访问")
-
- # 使用 LightRAG 的原生 get_knowledge_graph 方法
- knowledge_graph = await rag_instance.get_knowledge_graph(
- node_label=node_label, max_depth=max_depth, max_nodes=max_nodes
- )
-
- # 将 LightRAG 的 KnowledgeGraph 格式转换为前端需要的格式
- nodes = []
- for node in knowledge_graph.nodes:
- nodes.append(
- {
- "id": node.id,
- "labels": node.labels,
- "entity_type": node.properties.get("entity_type", "unknown"),
- "properties": node.properties,
- }
- )
-
- edges = []
- for edge in knowledge_graph.edges:
- edges.append(
- {
- "id": edge.id,
- "source": edge.source,
- "target": edge.target,
- "type": edge.type,
- "properties": edge.properties,
- }
- )
-
- result = {
- "success": True,
- "data": {
- "nodes": nodes,
- "edges": edges,
- "is_truncated": knowledge_graph.is_truncated,
- "total_nodes": len(nodes),
- "total_edges": len(edges),
- },
- }
-
- logger.info(f"成功获取子图 - 节点数: {len(nodes)}, 边数: {len(edges)}")
- return result
-
- except HTTPException:
- # 重新抛出 HTTP 异常
- raise
- except Exception as e:
- logger.error(f"获取子图数据失败: {e}")
- logger.error(f"Traceback: {traceback.format_exc()}")
- raise HTTPException(status_code=500, detail=f"获取子图数据失败: {str(e)}")
+ """(Deprecated) Use /graph/subgraph instead"""
+ return await get_subgraph(
+ db_id=db_id, node_label=node_label, max_depth=max_depth, max_nodes=max_nodes, current_user=current_user
+ )
@graph.get("/lightrag/databases")
async def get_lightrag_databases(current_user: User = Depends(get_admin_user)):
- """
- 获取所有可用的 LightRAG 数据库
-
- Returns:
- 可用的 LightRAG 数据库列表
- """
+ """(Deprecated) Use /graph/list instead"""
try:
lightrag_databases = knowledge_base.get_lightrag_databases()
return {"success": True, "data": {"databases": lightrag_databases}}
-
except Exception as e:
- logger.error(f"获取 LightRAG 数据库列表失败: {e}")
- raise HTTPException(status_code=500, detail=f"获取 LightRAG 数据库列表失败: {str(e)}")
-
-
-# =============================================================================
-# === 节点管理分组 ===
-# =============================================================================
+ raise HTTPException(status_code=500, detail=str(e))
@graph.get("/lightrag/labels")
async def get_lightrag_labels(
db_id: str = Query(..., description="数据库ID"), current_user: User = Depends(get_admin_user)
):
- """
- 获取知识图谱中的所有标签
+ """(Deprecated) Use /graph/labels instead"""
+ return await get_graph_labels(db_id=db_id, current_user=current_user)
- Args:
- db_id: LightRAG 数据库实例ID
- Returns:
- 图谱中所有可用的标签列表
- """
- try:
- logger.info(f"获取图谱标签 - db_id: {db_id}")
-
- # 检查是否是 LightRAG 数据库
- if not knowledge_base.is_lightrag_database(db_id):
- raise HTTPException(
- status_code=400, detail=f"数据库 {db_id} 不是 LightRAG 类型,图谱功能仅支持 LightRAG 知识库"
- )
-
- # 获取 LightRAG 实例
- rag_instance = await knowledge_base._get_lightrag_instance(db_id)
- if not rag_instance:
- raise HTTPException(status_code=404, detail=f"LightRAG 数据库 {db_id} 不存在或无法访问")
-
- # 使用 LightRAG 的原生方法获取所有标签
- labels = await rag_instance.get_graph_labels()
-
- return {"success": True, "data": {"labels": labels}}
-
- except HTTPException:
- # 重新抛出 HTTP 异常
- raise
- except Exception as e:
- logger.error(f"获取图谱标签失败: {e}")
- logger.error(f"Traceback: {traceback.format_exc()}")
- raise HTTPException(status_code=500, detail=f"获取图谱标签失败: {str(e)}")
+@graph.get("/lightrag/stats")
+async def get_lightrag_stats(
+ db_id: str = Query(..., description="数据库ID"), current_user: User = Depends(get_admin_user)
+):
+ """(Deprecated) Use /graph/stats instead"""
+ return await get_graph_stats(db_id=db_id, current_user=current_user)
@graph.get("/neo4j/nodes")
@@ -172,110 +261,19 @@ async def get_neo4j_nodes(
num: int = Query(100, description="节点数量", ge=1, le=1000),
current_user: User = Depends(get_admin_user),
):
- """
- 获取图谱节点样本数据
- """
- try:
- logger.debug(f"Get graph nodes in {kgdb_name} with {num} nodes")
-
- if not graph_base.is_running():
- raise HTTPException(status_code=400, detail="图数据库未启动")
-
- result = graph_base.get_sample_nodes(kgdb_name, num)
-
- return {"success": True, "result": result, "message": "success"}
-
- except Exception as e:
- logger.error(f"获取图节点数据失败: {e}\n{traceback.format_exc()}")
- raise HTTPException(status_code=500, detail=f"获取图节点数据失败: {str(e)}")
+ """(Deprecated) Use /graph/subgraph instead"""
+ response = await get_subgraph(db_id=kgdb_name, node_label="*", max_nodes=num, current_user=current_user)
+ return {"success": True, "result": response["data"], "message": "success"}
@graph.get("/neo4j/node")
async def get_neo4j_node(
entity_name: str = Query(..., description="实体名称"), current_user: User = Depends(get_admin_user)
):
- """
- 根据实体名称查询图节点
- """
- try:
- if not graph_base.is_running():
- raise HTTPException(status_code=400, detail="图数据库未启动")
-
- result = graph_base.query_node(keyword=entity_name)
-
- return {"success": True, "result": result, "message": "success"}
-
- except Exception as e:
- logger.error(f"查询图节点失败: {e}\n{traceback.format_exc()}")
- raise HTTPException(status_code=500, detail=f"查询图节点失败: {str(e)}")
-
-
-# =============================================================================
-# === 边管理分组 ===
-# =============================================================================
-
-# 可以在这里添加边相关的管理功能
-
-# =============================================================================
-# === 图谱分析分组 ===
-# =============================================================================
-
-
-@graph.get("/lightrag/stats")
-async def get_lightrag_stats(
- db_id: str = Query(..., description="数据库ID"), current_user: User = Depends(get_admin_user)
-):
- """
- 获取知识图谱统计信息
- """
- try:
- logger.info(f"获取图谱统计信息 - db_id: {db_id}")
-
- # 检查是否是 LightRAG 数据库
- if not knowledge_base.is_lightrag_database(db_id):
- raise HTTPException(
- status_code=400, detail=f"数据库 {db_id} 不是 LightRAG 类型,图谱功能仅支持 LightRAG 知识库"
- )
-
- # 获取 LightRAG 实例
- rag_instance = await knowledge_base._get_lightrag_instance(db_id)
- if not rag_instance:
- raise HTTPException(status_code=404, detail=f"LightRAG 数据库 {db_id} 不存在或无法访问")
-
- # 通过获取全图来统计节点和边的数量
- knowledge_graph = await rag_instance.get_knowledge_graph(
- node_label="*",
- max_depth=1,
- max_nodes=10000, # 设置较大值以获取完整统计
- )
-
- # 统计实体类型分布
- entity_types = {}
- for node in knowledge_graph.nodes:
- entity_type = node.properties.get("entity_type", "unknown")
- entity_types[entity_type] = entity_types.get(entity_type, 0) + 1
-
- entity_types_list = [
- {"type": k, "count": v} for k, v in sorted(entity_types.items(), key=lambda x: x[1], reverse=True)
- ]
-
- return {
- "success": True,
- "data": {
- "total_nodes": len(knowledge_graph.nodes),
- "total_edges": len(knowledge_graph.edges),
- "entity_types": entity_types_list,
- "is_truncated": knowledge_graph.is_truncated,
- },
- }
-
- except HTTPException:
- # 重新抛出 HTTP 异常
- raise
- except Exception as e:
- logger.error(f"获取图谱统计信息失败: {e}")
- logger.error(f"Traceback: {traceback.format_exc()}")
- raise HTTPException(status_code=500, detail=f"获取图谱统计信息失败: {str(e)}")
+ """(Deprecated) Use /graph/subgraph instead"""
+ # neo4j/node uses query_nodes(keyword=entity_name)
+ response = await get_subgraph(db_id="neo4j", node_label=entity_name, current_user=current_user)
+ return {"success": True, "result": response["data"], "message": "success"}
@graph.get("/neo4j/info")
@@ -298,10 +296,7 @@ async def index_neo4j_entities(data: dict = Body(default={}), current_user: User
if not graph_base.is_running():
raise HTTPException(status_code=400, detail="图数据库未启动")
- # 获取参数或使用默认值
kgdb_name = data.get("kgdb_name", "neo4j")
-
- # 调用GraphDatabase的add_embedding_to_nodes方法
count = graph_base.add_embedding_to_nodes(kgdb_name=kgdb_name)
return {
diff --git a/src/agents/common/middlewares/dynamic_tool_middleware.py b/src/agents/common/middlewares/dynamic_tool_middleware.py
index acce5f4d..a8dc196b 100644
--- a/src/agents/common/middlewares/dynamic_tool_middleware.py
+++ b/src/agents/common/middlewares/dynamic_tool_middleware.py
@@ -60,7 +60,8 @@ class DynamicToolMiddleware(AgentMiddleware):
logger.warning(f"MCP server '{mcp}' not pre-loaded. Please add it to mcp_servers list.")
logger.info(
- f"Dynamic tool selection: {len(enabled_tools)} tools enabled: {[tool.name for tool in enabled_tools]}"
+ f"Dynamic tool selection: {len(enabled_tools)} tools enabled: {[tool.name for tool in enabled_tools]}, "
+ f"selected_tools: {selected_tools}, selected_mcps: {selected_mcps}"
)
# 更新 request 中的工具列表
diff --git a/src/agents/common/toolkits/mysql/tools.py b/src/agents/common/toolkits/mysql/tools.py
index 3a255806..69af8238 100644
--- a/src/agents/common/toolkits/mysql/tools.py
+++ b/src/agents/common/toolkits/mysql/tools.py
@@ -38,7 +38,9 @@ def get_connection_manager() -> MySQLConnectionManager:
required_keys = ["host", "user", "password", "database"]
for key in required_keys:
if not mysql_config[key]:
- raise MySQLConnectionError(f"MySQL configuration missing required key: {key}")
+ raise MySQLConnectionError(
+ f"MySQL configuration missing required key: {key}, please check your environment variables."
+ )
_connection_manager = MySQLConnectionManager(mysql_config)
return _connection_manager
diff --git a/src/knowledge/adapters/__init__.py b/src/knowledge/adapters/__init__.py
new file mode 100644
index 00000000..027747fc
--- /dev/null
+++ b/src/knowledge/adapters/__init__.py
@@ -0,0 +1,6 @@
+from .base import GraphAdapter
+from .factory import GraphAdapterFactory
+from .lightrag import LightRAGGraphAdapter
+from .upload import UploadGraphAdapter
+
+__all__ = ["GraphAdapter", "UploadGraphAdapter", "LightRAGGraphAdapter", "GraphAdapterFactory"]
diff --git a/src/knowledge/adapters/base.py b/src/knowledge/adapters/base.py
new file mode 100644
index 00000000..a757c278
--- /dev/null
+++ b/src/knowledge/adapters/base.py
@@ -0,0 +1,87 @@
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class GraphAdapter(ABC):
+ """图谱适配器基类 (Base Graph Adapter)"""
+
+ @abstractmethod
+ async def query_nodes(self, keyword: str, **kwargs) -> dict[str, Any]:
+ """查询节点 (Query nodes)"""
+ pass
+
+ @abstractmethod
+ async def add_entity(self, triples: list[dict], **kwargs) -> bool:
+ """添加实体三元组 (Add entity triples)"""
+ pass
+
+ @abstractmethod
+ async def get_sample_nodes(self, num: int = 50, **kwargs) -> dict[str, list]:
+ """获取样本节点 (Get sample nodes)"""
+ pass
+
+ @abstractmethod
+ def normalize_node(self, raw_node: Any) -> dict[str, Any]:
+ """标准化节点格式 (Normalize node format)"""
+ pass
+
+ @abstractmethod
+ def normalize_edge(self, raw_edge: Any) -> dict[str, Any]:
+ """标准化边格式 (Normalize edge format)"""
+ pass
+
+ @abstractmethod
+ async def get_labels(self) -> list[str]:
+ """获取所有标签 (Get all labels)"""
+ pass
+
+ def _create_standard_node(
+ self,
+ node_id: str,
+ name: str,
+ entity_type: str,
+ labels: list[str],
+ properties: dict[str, Any],
+ source: str,
+ ) -> dict[str, Any]:
+ """
+ Helper to create a standardized node dictionary.
+ """
+ return {
+ "id": node_id,
+ "name": name,
+ "original_id": node_id,
+ "type": entity_type,
+ "labels": labels,
+ "properties": properties,
+ "normalized": {
+ "name": name,
+ "type": entity_type,
+ "source": source,
+ },
+ "graph_type": source,
+ }
+
+ def _create_standard_edge(
+ self,
+ edge_id: str,
+ source_id: str,
+ target_id: str,
+ edge_type: str,
+ properties: dict[str, Any],
+ direction: str = "directed",
+ ) -> dict[str, Any]:
+ """
+ Helper to create a standardized edge dictionary.
+ """
+ return {
+ "id": edge_id,
+ "source_id": source_id,
+ "target_id": target_id,
+ "type": edge_type,
+ "properties": properties,
+ "normalized": {
+ "type": edge_type,
+ "direction": direction,
+ },
+ }
diff --git a/src/knowledge/adapters/factory.py b/src/knowledge/adapters/factory.py
new file mode 100644
index 00000000..66a23680
--- /dev/null
+++ b/src/knowledge/adapters/factory.py
@@ -0,0 +1,23 @@
+from .base import GraphAdapter
+from .lightrag import LightRAGGraphAdapter
+from .upload import UploadGraphAdapter
+
+
+class GraphAdapterFactory:
+ """图谱适配器工厂 (Graph Adapter Factory)"""
+
+ _registry: dict[str, type[GraphAdapter]] = {"upload": UploadGraphAdapter, "lightrag": LightRAGGraphAdapter}
+
+ @classmethod
+ def register(cls, graph_type: str, adapter_class: type[GraphAdapter]):
+ """注册适配器类 (Register adapter class)"""
+ cls._registry[graph_type] = adapter_class
+
+ @classmethod
+ def create_adapter(cls, graph_type: str, **kwargs) -> GraphAdapter:
+ """创建适配器实例 (Create adapter instance)"""
+ adapter_class = cls._registry.get(graph_type)
+ if not adapter_class:
+ raise ValueError(f"Unknown graph type: {graph_type}")
+
+ return adapter_class(**kwargs)
diff --git a/src/knowledge/adapters/lightrag.py b/src/knowledge/adapters/lightrag.py
new file mode 100644
index 00000000..1b130209
--- /dev/null
+++ b/src/knowledge/adapters/lightrag.py
@@ -0,0 +1,145 @@
+import logging
+from typing import Any
+
+from .base import GraphAdapter
+
+logger = logging.getLogger(__name__)
+
+
+class LightRAGGraphAdapter(GraphAdapter):
+ """LightRAG图谱适配器 (LightRAG Graph Adapter)"""
+
+ def __init__(self, lightrag_instance: Any, config: dict[str, Any] = None):
+ self.config = config or {
+ "node_label_field": "labels",
+ "id_field": "id",
+ "type_field": "entity_type",
+ "relation_prefix": "HAS_",
+ }
+ self.lightrag = lightrag_instance
+
+ async def query_nodes(self, keyword: str, **kwargs) -> dict[str, Any]:
+ # Map keyword to node_label
+ # If keyword is empty or "*", query all (or sample)
+ node_label = keyword if keyword and keyword != "*" else "*"
+
+ max_depth = kwargs.get("max_depth", 2)
+ max_nodes = kwargs.get("max_nodes", 100)
+
+ # lightrag.get_knowledge_graph is async
+ # Note: if node_label is "*", LightRAG might return a large graph or sample depending on implementation
+ raw_graph = await self.lightrag.get_knowledge_graph(
+ node_label=node_label, max_depth=max_depth, max_nodes=max_nodes
+ )
+
+ return self._convert_lightrag_graph(raw_graph)
+
+ async def add_entity(self, triples: list[dict], **kwargs) -> bool:
+ """
+ LightRAG typically builds graph from text.
+ Direct triple injection might not be supported or requires different API.
+ """
+ logger.warning("add_entity is not fully supported for LightRAG adapter yet.")
+ return False
+
+ async def get_sample_nodes(self, num: int = 50, **kwargs) -> dict[str, list]:
+ # Use query_nodes with wildcard to get a subgraph
+ return await self.query_nodes("*", max_nodes=num, **kwargs)
+
+ def normalize_node(self, raw_node: Any) -> dict[str, Any]:
+ # Handle LightRAG Node object
+ node_id = getattr(raw_node, "id", None)
+ if node_id is None:
+ node_id = raw_node.get("id")
+
+ labels = getattr(raw_node, "labels", [])
+ if not labels and hasattr(raw_node, "get"):
+ labels = raw_node.get("labels", [])
+
+ properties = getattr(raw_node, "properties", {})
+ if not properties and hasattr(raw_node, "get"):
+ properties = raw_node.get("properties", {})
+
+ # 优先使用 entity_id 作为显示名称,因为 Neo4j 中 LightRAG 存储的实体名称在 entity_id 字段
+ # 如果不存在,则回退到 id
+ name = properties.get("entity_id", node_id)
+
+ # 尝试从 properties 获取 entity_type,或者从 labels 中推断(排除 kb_ 前缀的 label)
+ entity_type = properties.get("entity_type", "unknown")
+ if entity_type == "unknown" and labels:
+ for label in labels:
+ if not label.startswith("kb_"):
+ entity_type = label
+ break
+
+ return self._create_standard_node(
+ node_id=node_id, name=name, entity_type=entity_type, labels=labels, properties=properties, source="lightrag"
+ )
+
+ def normalize_edge(self, raw_edge: Any) -> dict[str, Any]:
+ # Handle LightRAG Edge object
+ edge_id = getattr(raw_edge, "id", None)
+ if edge_id is None:
+ edge_id = raw_edge.get("id")
+
+ source = getattr(raw_edge, "source", None)
+ if source is None:
+ source = raw_edge.get("source")
+
+ target = getattr(raw_edge, "target", None)
+ if target is None:
+ target = raw_edge.get("target")
+
+ edge_type = getattr(raw_edge, "type", None)
+ if edge_type is None:
+ edge_type = raw_edge.get("type")
+
+ properties = getattr(raw_edge, "properties", {})
+ if not properties and hasattr(raw_edge, "get"):
+ properties = raw_edge.get("properties", {})
+
+ # 优化边的显示类型
+ # LightRAG 的边类型通常是 "DIRECTED",具体含义在 keywords 或 description 中
+ display_type = edge_type
+ if edge_type == "DIRECTED":
+ keywords = properties.get("keywords", [])
+ if keywords and isinstance(keywords, list) and len(keywords) > 0:
+ display_type = keywords[0]
+ elif properties.get("description"):
+ # 如果没有 keywords,尝试从 description 截取(太长就算了)
+ desc = properties.get("description", "")
+ if len(desc) < 20:
+ display_type = desc
+ else:
+ display_type = "related" # fallback
+
+ return self._create_standard_edge(
+ edge_id=edge_id, source_id=source, target_id=target, edge_type=display_type, properties=properties
+ )
+
+ async def get_labels(self) -> list[str]:
+ return await self.lightrag.get_graph_labels()
+
+ def _convert_lightrag_graph(self, raw_graph) -> dict[str, Any]:
+ nodes = []
+ edges = []
+
+ # raw_graph has .nodes and .edges lists
+ if hasattr(raw_graph, "nodes"):
+ for node in raw_graph.nodes:
+ nodes.append(self.normalize_node(node))
+
+ if hasattr(raw_graph, "edges"):
+ for edge in raw_graph.edges:
+ edges.append(self.normalize_edge(edge))
+
+ result = {"nodes": nodes, "edges": edges}
+
+ # Add metadata if available
+ if hasattr(raw_graph, "is_truncated"):
+ result["is_truncated"] = raw_graph.is_truncated
+
+ result["total_nodes"] = len(nodes)
+ result["total_edges"] = len(edges)
+
+ return result
diff --git a/src/knowledge/adapters/upload.py b/src/knowledge/adapters/upload.py
new file mode 100644
index 00000000..3d4d3810
--- /dev/null
+++ b/src/knowledge/adapters/upload.py
@@ -0,0 +1,109 @@
+from typing import Any
+
+from src.knowledge.graph import GraphDatabase
+
+from .base import GraphAdapter
+
+
+class UploadGraphAdapter(GraphAdapter):
+ """Upload图谱适配器 (Upload Graph Adapter)"""
+
+ def __init__(self, graph_db_instance: GraphDatabase, config: dict[str, Any] = None):
+ self.config = config or {
+ "node_label": "Entity:Upload",
+ "id_field": "name",
+ "relation_label": "RELATION",
+ "default_tags": ["upload", "user_generated"],
+ }
+ self.graph_db = graph_db_instance
+
+ async def query_nodes(self, keyword: str, **kwargs) -> dict[str, Any]:
+ params = self._normalize_query_params(keyword, kwargs)
+
+ # 如果关键词是 "*" 或者为空,则执行采样查询
+ if not params["keyword"] or params["keyword"] == "*":
+ # 映射 max_nodes 到 num
+ num = kwargs.get("max_nodes", 100)
+ raw_results = self.graph_db.get_sample_nodes(kgdb_name=params.get("kgdb_name", "neo4j"), num=num)
+ else:
+ # 否则执行关键词搜索
+ # graph_db.query_node is sync
+ raw_results = self.graph_db.query_node(
+ keyword=params["keyword"],
+ threshold=params.get("threshold", 0.9),
+ kgdb_name=params.get("kgdb_name", "neo4j"),
+ hops=params.get("hops", 2),
+ return_format="graph",
+ )
+
+ return self._format_results(raw_results)
+
+ async def add_entity(self, triples: list[dict], **kwargs) -> bool:
+ kgdb_name = kwargs.get("kgdb_name", "neo4j")
+ # txt_add_vector_entity is async
+ await self.graph_db.txt_add_vector_entity(triples, kgdb_name=kgdb_name)
+ return True
+
+ async def get_sample_nodes(self, num: int = 50, **kwargs) -> dict[str, list]:
+ kgdb_name = kwargs.get("kgdb_name", "neo4j")
+ # get_sample_nodes is sync
+ raw_results = self.graph_db.get_sample_nodes(kgdb_name=kgdb_name, num=num)
+ return self._format_results(raw_results)
+
+ def normalize_node(self, raw_node: Any) -> dict[str, Any]:
+ """
+ raw_node expected format: {id: str, name: str, ...}
+ """
+ node_id = raw_node.get("id")
+ name = raw_node.get("name")
+
+ return self._create_standard_node(
+ node_id=node_id,
+ name=name,
+ entity_type="entity",
+ labels=["Entity", "Upload"],
+ properties=raw_node,
+ source="upload",
+ )
+
+ def normalize_edge(self, raw_edge: Any) -> dict[str, Any]:
+ """
+ raw_edge expected format: {type: str, source_id: str, target_id: str, ...}
+ """
+ # Generate an ID if not present (Upload graph edges might not have explicit ID in simple dict return)
+ edge_id = raw_edge.get("id")
+ if not edge_id:
+ edge_id = f"{raw_edge.get('source_id')}_{raw_edge.get('type')}_{raw_edge.get('target_id')}"
+
+ return self._create_standard_edge(
+ edge_id=edge_id,
+ source_id=raw_edge.get("source_id"),
+ target_id=raw_edge.get("target_id"),
+ edge_type=raw_edge.get("type"),
+ properties=raw_edge,
+ )
+
+ async def get_labels(self) -> list[str]:
+ kgdb_name = self.config.get("kgdb_name", "neo4j")
+ info = self.graph_db.get_graph_info(graph_name=kgdb_name)
+ return info.get("labels", []) if info else []
+
+ def _normalize_query_params(self, keyword: str, kwargs: dict) -> dict[str, Any]:
+ # Map max_depth to hops if present
+ hops = kwargs.get("hops")
+ if hops is None:
+ hops = kwargs.get("max_depth", 2)
+
+ return {
+ "keyword": keyword,
+ "threshold": kwargs.get("threshold", 0.9),
+ "kgdb_name": kwargs.get("kgdb_name", "neo4j"),
+ "hops": hops,
+ "filters": kwargs.get("filters", {}),
+ "context": kwargs.get("context", {}),
+ }
+
+ def _format_results(self, raw_results: dict[str, list]) -> dict[str, list]:
+ nodes = [self.normalize_node(n) for n in raw_results.get("nodes", [])]
+ edges = [self.normalize_edge(e) for e in raw_results.get("edges", [])]
+ return {"nodes": nodes, "edges": edges}
diff --git a/src/knowledge/graph.py b/src/knowledge/graph.py
index 1651f215..bfd325c1 100644
--- a/src/knowledge/graph.py
+++ b/src/knowledge/graph.py
@@ -64,6 +64,22 @@ class GraphDatabase:
assert self.driver is not None, "Database is not connected"
self.use_database(kgdb_name)
+ def _process_record_props(record):
+ """处理记录中的属性:扁平化 properties 并移除 embedding"""
+ if record is None:
+ return None
+
+ # 复制一份以避免修改原字典
+ data = dict(record)
+ props = data.pop("properties", {}) or {}
+
+ # 移除 embedding
+ if "embedding" in props:
+ del props["embedding"]
+
+ # 合并属性(优先保留原字典中的 id, name, type 等核心字段)
+ return {**props, **data}
+
def query(tx, num):
"""Note: 使用连通性查询获取集中的节点子图"""
# 首先尝试获取一个连通的子图
@@ -104,12 +120,18 @@ class GraphDatabase:
OPTIONAL MATCH (n)-[rel]-(m)
WHERE m IN final_nodes AND elementId(n) < elementId(m)
RETURN
- {id: elementId(n), name: n.name} AS h,
+ {id: elementId(n), name: n.name, properties: properties(n)} AS h,
CASE WHEN rel IS NOT NULL THEN
- {type: rel.type, source_id: elementId(n), target_id: elementId(m)}
+ {
+ id: elementId(rel),
+ type: rel.type,
+ source_id: elementId(startNode(rel)),
+ target_id: elementId(endNode(rel)),
+ properties: properties(rel)
+ }
ELSE null END AS r,
CASE WHEN m IS NOT NULL THEN
- {id: elementId(m), name: m.name}
+ {id: elementId(m), name: m.name, properties: properties(m)}
ELSE null END AS t
"""
@@ -119,7 +141,7 @@ class GraphDatabase:
node_ids = set()
for item in results:
- h_node = item["h"]
+ h_node = _process_record_props(item["h"])
# 始终添加头节点
if h_node["id"] not in node_ids:
@@ -128,14 +150,15 @@ class GraphDatabase:
# 只有当边和尾节点都存在时才处理
if item["r"] is not None and item["t"] is not None:
- t_node = item["t"]
+ t_node = _process_record_props(item["t"])
+ r_edge = _process_record_props(item["r"])
# 避免重复添加尾节点
if t_node["id"] not in node_ids:
formatted_results["nodes"].append(t_node)
node_ids.add(t_node["id"])
- formatted_results["edges"].append(item["r"])
+ formatted_results["edges"].append(r_edge)
# 如果连通查询返回的节点数不足,补充更多节点
if len(formatted_results["nodes"]) < num:
@@ -145,14 +168,14 @@ class GraphDatabase:
supplement_query = """
MATCH (n:Entity)
WHERE NOT elementId(n) IN $existing_ids
- RETURN {id: elementId(n), name: n.name} AS node
+ RETURN {id: elementId(n), name: n.name, properties: properties(n)} AS node
LIMIT $count
"""
supplement_results = tx.run(supplement_query, existing_ids=list(node_ids), count=remaining_count)
for item in supplement_results:
- node = item["node"]
+ node = _process_record_props(item["node"])
formatted_results["nodes"].append(node)
node_ids.add(node["id"])
@@ -165,9 +188,15 @@ class GraphDatabase:
MATCH (n:Entity)-[r]-(m:Entity)
WHERE elementId(n) < elementId(m)
RETURN
- {id: elementId(n), name: n.name} AS h,
- {type: r.type, source_id: elementId(n), target_id: elementId(m)} AS r,
- {id: elementId(m), name: m.name} AS t
+ {id: elementId(n), name: n.name, properties: properties(n)} AS h,
+ {
+ id: elementId(r),
+ type: r.type,
+ source_id: elementId(startNode(r)),
+ target_id: elementId(endNode(r)),
+ properties: properties(r)
+ } AS r,
+ {id: elementId(m), name: m.name, properties: properties(m)} AS t
LIMIT $num
"""
results = tx.run(fallback_query, num=int(num))
@@ -175,8 +204,9 @@ class GraphDatabase:
node_ids = set()
for item in results:
- h_node = item["h"]
- t_node = item["t"]
+ h_node = _process_record_props(item["h"])
+ t_node = _process_record_props(item["t"])
+ r_edge = _process_record_props(item["r"])
# 避免重复添加节点
if h_node["id"] not in node_ids:
@@ -186,7 +216,7 @@ class GraphDatabase:
formatted_results["nodes"].append(t_node)
node_ids.add(t_node["id"])
- formatted_results["edges"].append(item["r"])
+ formatted_results["edges"].append(r_edge)
return formatted_results
@@ -230,18 +260,47 @@ class GraphDatabase:
return True
return False
+ def _parse_node(node_data):
+ """解析节点数据,返回 (name, props)"""
+ if isinstance(node_data, dict):
+ props = node_data.copy()
+ name = props.pop("name", "")
+ return name, props
+ return str(node_data), {}
+
+ def _parse_relation(rel_data):
+ """解析关系数据,返回 (type, props)"""
+ if isinstance(rel_data, dict):
+ props = rel_data.copy()
+ rel_type = props.pop("type", "")
+ return rel_type, props
+ return str(rel_data), {}
+
def _create_graph(tx, data):
"""添加一个三元组"""
for entry in data:
+ h_name, h_props = _parse_node(entry.get("h"))
+ t_name, t_props = _parse_node(entry.get("t"))
+ r_type, r_props = _parse_relation(entry.get("r"))
+
+ if not h_name or not t_name or not r_type:
+ continue
+
tx.run(
"""
- MERGE (h:Entity:Upload {name: $h})
- MERGE (t:Entity:Upload {name: $t})
- MERGE (h)-[r:RELATION {type: $r}]->(t)
+ MERGE (h:Entity:Upload {name: $h_name})
+ SET h += $h_props
+ MERGE (t:Entity:Upload {name: $t_name})
+ SET t += $t_props
+ MERGE (h)-[r:RELATION {type: $r_type}]->(t)
+ SET r += $r_props
""",
- h=entry["h"],
- t=entry["t"],
- r=entry["r"],
+ h_name=h_name,
+ h_props=h_props,
+ t_name=t_name,
+ t_props=t_props,
+ r_type=r_type,
+ r_props=r_props,
)
def _create_vector_index(tx, dim):
@@ -263,6 +322,9 @@ class GraphDatabase:
# 构建参数字典,将列表转换为"param0"、"param1"等键值对形式
params = {f"param{i}": name for i, name in enumerate(entity_names)}
+ if not params:
+ return []
+
# 构建查询参数列表
param_placeholders = ", ".join([f"${key}" for key in params.keys()])
@@ -305,20 +367,24 @@ class GraphDatabase:
session.execute_write(_create_vector_index, getattr(cur_embed_info, "dimension", 1024))
# 收集所有需要处理的实体名称,去重
- all_entities = []
+ all_entities = set()
for entry in triples:
- if entry["h"] not in all_entities:
- all_entities.append(entry["h"])
- if entry["t"] not in all_entities:
- all_entities.append(entry["t"])
+ h_name, _ = _parse_node(entry.get("h"))
+ t_name, _ = _parse_node(entry.get("t"))
+ if h_name:
+ all_entities.add(h_name)
+ if t_name:
+ all_entities.add(t_name)
+
+ all_entities_list = list(all_entities)
# 筛选出没有embedding的节点
- nodes_without_embedding = session.execute_read(_get_nodes_without_embedding, all_entities)
+ nodes_without_embedding = session.execute_read(_get_nodes_without_embedding, all_entities_list)
if not nodes_without_embedding:
logger.info("所有实体已有embedding,无需重新计算")
return
- logger.info(f"需要为{len(nodes_without_embedding)}/{len(all_entities)}个实体计算embedding")
+ logger.info(f"需要为{len(nodes_without_embedding)}/{len(all_entities_list)}个实体计算embedding")
# 批量处理实体
max_batch_size = 1024 # 限制此部分的主要是内存大小 1024 * 1024 * 4 / 1024 / 1024 = 4GB
@@ -587,30 +653,70 @@ class GraphDatabase:
self.use_database(kgdb_name)
+ def _process_record_props(record):
+ """处理记录中的属性:扁平化 properties 并移除 embedding"""
+ if record is None:
+ return None
+
+ # 复制一份以避免修改原字典
+ data = dict(record)
+ props = data.pop("properties", {}) or {}
+
+ # 移除 embedding
+ if "embedding" in props:
+ del props["embedding"]
+
+ # 合并属性(优先保留原字典中的 id, name, type 等核心字段)
+ return {**props, **data}
+
def query(tx, entity_name, hops, limit):
try:
query_str = """
WITH [
// 1跳出边
[(n {name: $entity_name})-[r1]->(m1) |
- {h: {id: elementId(n), name: n.name},
- r: {type: r1.type, source_id: elementId(n), target_id: elementId(m1)},
- t: {id: elementId(m1), name: m1.name}}],
+ {h: {id: elementId(n), name: n.name, properties: properties(n)},
+ r: {
+ id: elementId(r1),
+ type: r1.type,
+ source_id: elementId(n),
+ target_id: elementId(m1),
+ properties: properties(r1)
+ },
+ t: {id: elementId(m1), name: m1.name, properties: properties(m1)}}],
// 2跳出边
[(n {name: $entity_name})-[r1]->(m1)-[r2]->(m2) |
- {h: {id: elementId(m1), name: m1.name},
- r: {type: r2.type, source_id: elementId(m1), target_id: elementId(m2)},
- t: {id: elementId(m2), name: m2.name}}],
+ {h: {id: elementId(m1), name: m1.name, properties: properties(m1)},
+ r: {
+ id: elementId(r2),
+ type: r2.type,
+ source_id: elementId(m1),
+ target_id: elementId(m2),
+ properties: properties(r2)
+ },
+ t: {id: elementId(m2), name: m2.name, properties: properties(m2)}}],
// 1跳入边
[(m1)-[r1]->(n {name: $entity_name}) |
- {h: {id: elementId(m1), name: m1.name},
- r: {type: r1.type, source_id: elementId(m1), target_id: elementId(n)},
- t: {id: elementId(n), name: n.name}}],
+ {h: {id: elementId(m1), name: m1.name, properties: properties(m1)},
+ r: {
+ id: elementId(r1),
+ type: r1.type,
+ source_id: elementId(m1),
+ target_id: elementId(n),
+ properties: properties(r1)
+ },
+ t: {id: elementId(n), name: n.name, properties: properties(n)}}],
// 2跳入边
[(m2)-[r2]->(m1)-[r1]->(n {name: $entity_name}) |
- {h: {id: elementId(m2), name: m2.name},
- r: {type: r2.type, source_id: elementId(m2), target_id: elementId(m1)},
- t: {id: elementId(m1), name: m1.name}}]
+ {h: {id: elementId(m2), name: m2.name, properties: properties(m2)},
+ r: {
+ id: elementId(r2),
+ type: r2.type,
+ source_id: elementId(m2),
+ target_id: elementId(m1),
+ properties: properties(r2)
+ },
+ t: {id: elementId(m1), name: m1.name, properties: properties(m1)}}]
] AS all_results
UNWIND all_results AS result_list
UNWIND result_list AS item
@@ -626,9 +732,13 @@ class GraphDatabase:
formatted_results = {"nodes": [], "edges": [], "triples": []}
for item in results:
- formatted_results["nodes"].extend([item["h"], item["t"]])
- formatted_results["edges"].append(item["r"])
- formatted_results["triples"].append((item["h"]["name"], item["r"]["type"], item["t"]["name"]))
+ h = _process_record_props(item["h"])
+ r = _process_record_props(item["r"])
+ t = _process_record_props(item["t"])
+
+ formatted_results["nodes"].extend([h, t])
+ formatted_results["edges"].append(r)
+ formatted_results["triples"].append((h["name"], r["type"], t["name"]))
logger.debug(f"Query Results: {results}")
return formatted_results
diff --git a/src/knowledge/implementations/lightrag.py b/src/knowledge/implementations/lightrag.py
index 08feed01..d62d1f44 100644
--- a/src/knowledge/implementations/lightrag.py
+++ b/src/knowledge/implementations/lightrag.py
@@ -200,6 +200,7 @@ class LightRagKB(KnowledgeBase):
def _get_embedding_func(self, embed_info: dict):
"""获取 embedding 函数"""
config_dict = get_embedding_config(embed_info)
+ logger.debug(f"Embedding config dict: {config_dict}")
if config_dict.get("model_id") and config_dict["model_id"].startswith("ollama"):
from lightrag.llm.ollama import ollama_embed
@@ -219,12 +220,19 @@ class LightRagKB(KnowledgeBase):
),
)
+ # 尝试获取模型名称,支持多种键名以保持兼容性
+ if "name" in config_dict and config_dict["name"]:
+ model_name = config_dict["name"]
+ elif "model" in config_dict and config_dict["model"]:
+ model_name = config_dict["model"]
+ else:
+ raise ValueError(f"Neither 'name' nor 'model' found in config_dict or both are empty: {config_dict}")
return EmbeddingFunc(
embedding_dim=config_dict["dimension"],
max_token_size=8192,
func=lambda texts: openai_embed(
texts=texts,
- model=config_dict["model"],
+ model=model_name,
api_key=config_dict["api_key"],
base_url=config_dict["base_url"].replace("/embeddings", ""),
),
diff --git a/src/knowledge/utils/kb_utils.py b/src/knowledge/utils/kb_utils.py
index ac9411ed..75dbec58 100644
--- a/src/knowledge/utils/kb_utils.py
+++ b/src/knowledge/utils/kb_utils.py
@@ -289,30 +289,52 @@ def get_embedding_config(embed_info: dict) -> dict:
Returns:
dict: 标准化的嵌入配置
"""
- config_dict = {}
-
try:
- if embed_info:
- # 优先检查是否有 model_id 字段
- if "model_id" in embed_info:
- return config.embed_model_names[embed_info["model_id"]].model_dump()
- elif hasattr(embed_info, "name") and isinstance(embed_info, EmbedModelInfo):
- return embed_info.model_dump()
- else:
- # 字典形式(保持向后兼容)
- config_dict["model"] = embed_info["name"]
- config_dict["api_key"] = os.getenv(embed_info["api_key"]) or embed_info["api_key"]
- config_dict["base_url"] = embed_info["base_url"]
- config_dict["dimension"] = embed_info.get("dimension", 1024)
- else:
- return config.embed_model_names[config.embed_model].model_dump()
+ # 检查 embed_info 是否有效
+ if not embed_info or ("model" not in embed_info and "name" not in embed_info):
+ logger.error(f"Invalid embed_info: {embed_info}, using default embedding model config")
+ raise ValueError("Invalid embed_info: must be a non-empty dictionary")
+
+ # 优先检查是否有 model_id 字段
+ if "model_id" in embed_info and embed_info["model_id"]:
+ logger.warning(f"Using model_id: {embed_info['model_id']}")
+ config_dict = config.embed_model_names[embed_info["model_id"]].model_dump()
+ config_dict["api_key"] = os.getenv(config_dict["api_key"]) or config_dict["api_key"]
+ return config_dict
+
+ # 检查是否是 EmbedModelInfo 对象(在某些情况下可能直接传入对象)
+ if hasattr(embed_info, "name") and isinstance(embed_info, EmbedModelInfo):
+ logger.debug(f"Using EmbedModelInfo object: {embed_info.name}")
+ config_dict = embed_info.model_dump()
+ config_dict["api_key"] = os.getenv(config_dict["api_key"]) or config_dict["api_key"]
+ return config_dict
+
+ # 字典形式(保持向后兼容)
+ # 检查必需字段是否存在
+ if not embed_info.get("name") or not embed_info.get("base_url"):
+ logger.warning(f"embed_info missing required 'name' or 'base_url' field: {embed_info}, using default")
+ raise ValueError("embed_info missing required 'name' or 'base_url' field")
+
+ config_dict = {
+ "model": embed_info["name"],
+ "api_key": os.getenv(embed_info["api_key"]) or embed_info["api_key"],
+ "base_url": embed_info["base_url"],
+ "dimension": embed_info.get("dimension", 1024),
+ }
+ logger.debug(f"Embedding config from dict: {config_dict}")
+ return config_dict
except Exception as e:
- logger.error(f"Error in get_embedding_config: {e}, {embed_info}")
- raise ValueError(f"Error in get_embedding_config: {e}")
-
- logger.debug(f"Embedding config: {config_dict}")
- return config_dict
+ logger.error(f"Error in get_embedding_config: {e}, embed_info={embed_info}")
+ # 返回默认配置作为fallback
+ logger.warning("Falling back to default embedding model config")
+ try:
+ config_dict = config.embed_model_names[config.embed_model].model_dump()
+ config_dict["api_key"] = os.getenv(config_dict["api_key"]) or config_dict["api_key"]
+ return config_dict
+ except Exception as fallback_error:
+ logger.error(f"Failed to get default embedding config: {fallback_error}")
+ raise ValueError(f"Failed to get embedding config and fallback failed: {e}")
def is_minio_url(file_path: str) -> bool:
diff --git a/test/api/test_unified_graph_router.py b/test/api/test_unified_graph_router.py
new file mode 100644
index 00000000..632b1b7f
--- /dev/null
+++ b/test/api/test_unified_graph_router.py
@@ -0,0 +1,118 @@
+"""
+Integration tests for the unified graph router endpoints.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+
+async def test_get_graphs_list(test_client, admin_headers):
+ """Test retrieving the list of all graphs."""
+ response = await test_client.get("/api/graph/list", headers=admin_headers)
+ assert response.status_code == 200
+ payload = response.json()
+ assert payload["success"] is True
+ graphs = payload["data"]
+ assert isinstance(graphs, list)
+
+ # Check for Neo4j default graph
+ neo4j_graph = next((g for g in graphs if g["id"] == "neo4j"), None)
+ assert neo4j_graph is not None
+ assert neo4j_graph["type"] == "neo4j"
+
+ # Note: LightRAG graphs might be empty if none created, but we check structure
+
+
+async def test_get_subgraph_neo4j(test_client, admin_headers):
+ """Test unified subgraph query for Neo4j."""
+ # Query with a wildcard or a known node. Using "*" to get a sample.
+ response = await test_client.get(
+ "/api/graph/subgraph", params={"db_id": "neo4j", "node_label": "*", "max_nodes": 10}, headers=admin_headers
+ )
+ assert response.status_code == 200
+ payload = response.json()
+ assert payload["success"] is True
+ data = payload["data"]
+ assert "nodes" in data
+ assert "edges" in data
+ assert isinstance(data["nodes"], list)
+
+
+async def test_get_subgraph_lightrag(test_client, admin_headers, knowledge_database):
+ """Test unified subgraph query for LightRAG."""
+ db_id = knowledge_database["db_id"]
+ response = await test_client.get(
+ "/api/graph/subgraph", params={"db_id": db_id, "node_label": "*", "max_nodes": 10}, headers=admin_headers
+ )
+ assert response.status_code == 200
+ payload = response.json()
+ assert payload["success"] is True
+ data = payload["data"]
+ assert "nodes" in data
+ assert "edges" in data
+
+
+async def test_get_stats_neo4j(test_client, admin_headers):
+ """Test stats endpoint for Neo4j."""
+ response = await test_client.get("/api/graph/stats", params={"db_id": "neo4j"}, headers=admin_headers)
+ assert response.status_code == 200
+ payload = response.json()
+ assert payload["success"] is True
+ data = payload["data"]
+ assert "total_nodes" in data
+ assert "total_edges" in data
+ assert "entity_types" in data
+
+
+async def test_get_stats_lightrag(test_client, admin_headers, knowledge_database):
+ """Test stats endpoint for LightRAG."""
+ db_id = knowledge_database["db_id"]
+ response = await test_client.get("/api/graph/stats", params={"db_id": db_id}, headers=admin_headers)
+ assert response.status_code == 200
+ payload = response.json()
+ assert payload["success"] is True
+ data = payload["data"]
+ assert "total_nodes" in data
+ assert "total_edges" in data
+ assert "entity_types" in data
+
+
+async def test_get_labels_neo4j(test_client, admin_headers):
+ """Test labels endpoint for Neo4j."""
+ response = await test_client.get("/api/graph/labels", params={"db_id": "neo4j"}, headers=admin_headers)
+ assert response.status_code == 200
+ payload = response.json()
+ assert payload["success"] is True
+ data = payload["data"]
+ assert "labels" in data
+ assert isinstance(data["labels"], list)
+
+
+async def test_deprecated_neo4j_endpoints(test_client, admin_headers):
+ """Verify deprecated endpoints still work and return correct structure."""
+
+ # /neo4j/nodes
+ response = await test_client.get(
+ "/api/graph/neo4j/nodes", params={"kgdb_name": "neo4j", "num": 5}, headers=admin_headers
+ )
+ assert response.status_code == 200
+ payload = response.json()
+ # Check compatibility structure
+ assert payload["success"] is True
+ assert "result" in payload
+ assert payload["message"] == "success"
+ assert "nodes" in payload["result"]
+
+ # /neo4j/node
+ # This might return empty if "NonExistentEntity" doesn't exist, but structure should be valid
+ response = await test_client.get(
+ "/api/graph/neo4j/node", params={"entity_name": "NonExistentEntity"}, headers=admin_headers
+ )
+ assert response.status_code == 200
+ payload = response.json()
+ assert payload["success"] is True
+ assert "result" in payload
+ assert payload["message"] == "success"
diff --git a/test/data/complex_graph_test.jsonl b/test/data/complex_graph_test.jsonl
new file mode 100644
index 00000000..d36b0de9
--- /dev/null
+++ b/test/data/complex_graph_test.jsonl
@@ -0,0 +1,6 @@
+{"h": {"name": "孙悟空", "title": "齐天大圣", "weapon": "如意金箍棒", "species": "猴"}, "t": {"name": "唐僧", "title": "旃檀功德佛", "species": "人"}, "r": {"type": "徒弟", "order": 1}}
+{"h": {"name": "猪八戒", "title": "天蓬元帅", "weapon": "九齿钉耙", "species": "猪"}, "t": {"name": "唐僧"}, "r": {"type": "徒弟", "order": 2}}
+{"h": {"name": "沙悟净", "title": "卷帘大将", "weapon": "降妖宝杖"}, "t": {"name": "唐僧"}, "r": {"type": "徒弟", "order": 3}}
+{"h": {"name": "白龙马", "origin": "西海龙宫"}, "t": {"name": "唐僧"}, "r": {"type": "坐骑", "original_form": "龙"}}
+{"h": "孙悟空", "t": {"name": "猪八戒"}, "r": {"type": "师兄弟", "relationship": "conflict/cooperate"}}
+{"h": {"name": "如来佛祖", "location": "西天雷音寺"}, "t": {"name": "孙悟空"}, "r": {"type": "压制", "tool": "五指山", "duration": "500年"}}
\ No newline at end of file
diff --git a/test/test_graph_unit.py b/test/test_graph_unit.py
new file mode 100644
index 00000000..4c1b9d89
--- /dev/null
+++ b/test/test_graph_unit.py
@@ -0,0 +1,96 @@
+import pytest
+from unittest.mock import MagicMock, patch
+import os
+import sys
+
+# Add project root to path
+sys.path.append(os.getcwd())
+
+from src.knowledge.graph import GraphDatabase
+
+
+@pytest.mark.asyncio
+async def test_txt_add_vector_entity_parsing():
+ # Mock driver and session
+ mock_driver = MagicMock()
+ mock_session = MagicMock()
+ mock_driver.session.return_value.__enter__.return_value = mock_session
+
+ # Setup mock transaction
+ mock_tx = MagicMock()
+
+ def side_effect_execute_write(func, *args, **kwargs):
+ return func(mock_tx, *args, **kwargs)
+
+ # Mock execute_read to return empty list (no missing embeddings for this test)
+ # The code calls _get_nodes_without_embedding which returns [record['name']]
+ # If we return [], it means all nodes have embeddings or none found.
+ # Actually, the code checks:
+ # nodes_without_embedding = session.execute_read(_get_nodes_without_embedding, all_entities)
+ # Let's mock it to return empty list so we skip embedding generation loop which simplifies test
+ mock_session.execute_read.return_value = []
+
+ mock_session.execute_write.side_effect = side_effect_execute_write
+
+ # Mock embedding model
+ with patch("src.knowledge.graph.select_embedding_model") as mock_select_model:
+ mock_embed_model = MagicMock()
+ mock_select_model.return_value = mock_embed_model
+
+ # Instantiate GraphDatabase with mocked driver
+ # We also need to patch GD.driver in the init
+ with patch("src.knowledge.graph.GD.driver", return_value=mock_driver):
+ gd = GraphDatabase()
+ # Manually set driver and status just in case init didn't work as expected due to other mocks
+ gd.driver = mock_driver
+ gd.status = "open"
+ gd.embed_model_name = "test_model" # avoid config check issues if possible
+
+ # Mock config to match
+ with patch("src.knowledge.graph.config.embed_model", "test_model"):
+ with patch("src.knowledge.graph.config.embed_model_names", {"test_model": MagicMock(dimension=1024)}):
+ # Test data: Mixed format
+ triples = [
+ # Legacy format
+ {"h": "A", "r": "KNOWS", "t": "B"},
+ # Extended format
+ {
+ "h": {"name": "C", "age": 30},
+ "r": {"type": "LIKES", "weight": 0.8},
+ "t": {"name": "D", "role": "User"},
+ },
+ ]
+
+ # Run the method
+ await gd.txt_add_vector_entity(triples)
+
+ # Verify calls to mock_tx.run
+ merge_calls = []
+ for call in mock_tx.run.call_args_list:
+ args, kwargs = call
+ query = args[0] if args else kwargs.get("query", "")
+ if "MERGE (h:Entity:Upload" in query:
+ # The args are passed as kwargs to run: h_name=..., etc.
+ merge_calls.append(kwargs)
+
+ assert len(merge_calls) == 2, f"Expected 2 merge calls, got {len(merge_calls)}"
+
+ # Call 1 (Legacy)
+ call1 = merge_calls[0]
+ assert call1["h_name"] == "A"
+ assert call1["h_props"] == {}
+ assert call1["t_name"] == "B"
+ assert call1["t_props"] == {}
+ assert call1["r_type"] == "KNOWS"
+ assert call1["r_props"] == {}
+
+ # Call 2 (Extended)
+ call2 = merge_calls[1]
+ assert call2["h_name"] == "C"
+ assert call2["h_props"] == {"age": 30}
+ assert call2["t_name"] == "D"
+ assert call2["t_props"] == {"role": "User"}
+ assert call2["r_type"] == "LIKES"
+ assert call2["r_props"] == {"weight": 0.8}
+
+ print("Verification passed!")
diff --git a/web/src/apis/graph_api.js b/web/src/apis/graph_api.js
index aed00c9a..be5b67d1 100644
--- a/web/src/apis/graph_api.js
+++ b/web/src/apis/graph_api.js
@@ -7,15 +7,23 @@ import { apiGet, apiPost } from './base'
*/
// =============================================================================
-// === LightRAG图知识库接口分组 ===
+// === 统一图谱接口 (Unified Graph API) ===
// =============================================================================
-export const lightragApi = {
+export const unifiedApi = {
/**
- * 获取LightRAG知识图谱子图数据
+ * 获取所有可用的知识图谱列表
+ * @returns {Promise} - 图谱列表
+ */
+ getGraphs: async () => {
+ return await apiGet('/api/graph/list', {}, true)
+ },
+
+ /**
+ * 获取子图数据 (统一接口)
* @param {Object} params - 查询参数
- * @param {string} params.db_id - LightRAG数据库ID
- * @param {string} params.node_label - 节点标签("*"获取全图)
+ * @param {string} params.db_id - 图谱ID
+ * @param {string} params.node_label - 节点标签/关键词
* @param {number} params.max_depth - 最大深度
* @param {number} params.max_nodes - 最大节点数
* @returns {Promise} - 子图数据
@@ -34,37 +42,12 @@ export const lightragApi = {
max_nodes: max_nodes.toString()
})
- return await apiGet(`/api/graph/lightrag/subgraph?${queryParams.toString()}`, {}, true)
+ return await apiGet(`/api/graph/subgraph?${queryParams.toString()}`, {}, true)
},
/**
- * 获取所有可用的LightRAG数据库
- * @returns {Promise} - LightRAG数据库列表
- */
- getDatabases: async () => {
- return await apiGet('/api/graph/lightrag/databases', {}, true)
- },
-
- /**
- * 获取LightRAG图谱标签列表
- * @param {string} db_id - LightRAG数据库ID
- * @returns {Promise} - 标签列表
- */
- getLabels: async (db_id) => {
- if (!db_id) {
- throw new Error('db_id is required')
- }
-
- const queryParams = new URLSearchParams({
- db_id: db_id
- })
-
- return await apiGet(`/api/graph/lightrag/labels?${queryParams.toString()}`, {}, true)
- },
-
- /**
- * 获取LightRAG图谱统计信息
- * @param {string} db_id - LightRAG数据库ID
+ * 获取图谱统计信息 (统一接口)
+ * @param {string} db_id - 图谱ID
* @returns {Promise} - 统计信息
*/
getStats: async (db_id) => {
@@ -76,10 +59,28 @@ export const lightragApi = {
db_id: db_id
})
- return await apiGet(`/api/graph/lightrag/stats?${queryParams.toString()}`, {}, true)
+ return await apiGet(`/api/graph/stats?${queryParams.toString()}`, {}, true)
+ },
+
+ /**
+ * 获取图谱标签列表 (统一接口)
+ * @param {string} db_id - 图谱ID
+ * @returns {Promise} - 标签列表
+ */
+ getLabels: async (db_id) => {
+ if (!db_id) {
+ throw new Error('db_id is required')
+ }
+
+ const queryParams = new URLSearchParams({
+ db_id: db_id
+ })
+
+ return await apiGet(`/api/graph/labels?${queryParams.toString()}`, {}, true)
}
}
+
// =============================================================================
// === Neo4j图数据库接口分组 ===
// =============================================================================
@@ -222,11 +223,21 @@ export const getGraphStats = async () => {
return neo4jApi.getInfo()
}
-// 保持旧的分组导出,便于批量替换
+// 兼容性导出 - 使用统一接口替代旧有的 graphApi
export const graphApi = {
- getSubgraph: lightragApi.getSubgraph,
- getDatabases: lightragApi.getDatabases,
- getLabels: lightragApi.getLabels,
- getStats: lightragApi.getStats,
- ...neo4jApi // 临时兼容
+ // 使用统一接口替代 LightRAG 接口
+ getSubgraph: unifiedApi.getSubgraph,
+ getDatabases: async () => {
+ // 使用统一接口获取所有图谱,然后过滤出 LightRAG 类型的
+ const response = await unifiedApi.getGraphs()
+ if (response.success) {
+ const lightragDbs = response.data.filter(graph => graph.type === 'lightrag')
+ return { success: true, data: { databases: lightragDbs } }
+ }
+ return response
+ },
+ getLabels: unifiedApi.getLabels,
+ getStats: unifiedApi.getStats,
+ // 保留 Neo4j 接口
+ ...neo4jApi
}
\ No newline at end of file
diff --git a/web/src/components/GraphCanvas.vue b/web/src/components/GraphCanvas.vue
index 0997c478..9458be8e 100644
--- a/web/src/components/GraphCanvas.vue
+++ b/web/src/components/GraphCanvas.vue
@@ -37,7 +37,7 @@ const props = defineProps({
highlightKeywords: { type: Array, default: () => [] }
})
-const emit = defineEmits(['ready', 'data-rendered'])
+const emit = defineEmits(['ready', 'data-rendered', 'node-click', 'edge-click', 'canvas-click'])
const container = ref(null)
const rootEl = ref(null)
@@ -88,6 +88,7 @@ function formatData() {
data: {
label: n[props.labelField] ?? n.name ?? String(n.id),
degree: degrees.get(String(n.id)) || 0,
+ original: n // 保存原始数据
},
}))
@@ -95,7 +96,10 @@ function formatData() {
id: e.id ? String(e.id) : `edge-${idx}`,
source: String(e.source_id),
target: String(e.target_id),
- data: { label: e.type ?? '' },
+ data: {
+ label: e.type ?? '',
+ original: e // 保存原始数据
+ },
}))
return { nodes, edges }
@@ -185,9 +189,38 @@ function initGraph() {
unselectedState: 'inactive', // 未选中节点状态
multiple: true,
trigger: ['shift'],
+ // 禁用默认的选中效果,避免与自定义事件冲突
+ disableDefault: false,
}
],
})
+
+ // 绑定事件
+ graphInstance.on('node:click', (evt) => {
+ const { target } = evt
+ // 获取节点ID
+ const nodeId = target.id
+ // 从 graph data 中找到对应的节点数据
+ // 注意:G6 v5 的事件对象结构可能有所不同,这里假设 target.id 是节点ID
+ // 也可以通过 graphInstance.getNodeData(nodeId) 获取
+ const nodeData = graphInstance.getNodeData(nodeId)
+ emit('node-click', nodeData)
+ })
+
+ graphInstance.on('edge:click', (evt) => {
+ const { target } = evt
+ const edgeId = target.id
+ const edgeData = graphInstance.getEdgeData(edgeId)
+ emit('edge-click', edgeData)
+ })
+
+ graphInstance.on('canvas:click', (evt) => {
+ // 只有点击画布空白处才触发
+ if (!evt.target) {
+ emit('canvas-click')
+ }
+ })
+
emit('ready', graphInstance)
}
@@ -364,7 +397,7 @@ defineExpose({
position: relative;
width: 100%;
height: 100%;
- background-color: var(--gray-0);
+ // background-color: var(--gray-0);
.graph-canvas {
width: 100%;
diff --git a/web/src/components/GraphDetailPanel.vue b/web/src/components/GraphDetailPanel.vue
new file mode 100644
index 00000000..86fb9ff7
--- /dev/null
+++ b/web/src/components/GraphDetailPanel.vue
@@ -0,0 +1,167 @@
+
+
如需上传文档到当前选中的知识库,请前往对应的知识库详情页面进行操作:
+ +本页面展示的是 Neo4j 图数据库中的知识图谱信息。
具体展示内容包括:
注意:
graphbase.py 中的 get_sample_nodes 方法实现:graphbase.py 中的 get_sample_nodes 方法实现。MATCH (n:Entity)-[r]->(m:Entity)
-RETURN
- {id: elementId(n), name: n.name} AS h,
- {type: r.type, source_id: elementId(n), target_id: elementId(m)} AS r,
- {id: elementId(m), name: m.name} AS t
-LIMIT $num
- 如需查看完整的 Neo4j 数据库内容,请使用 "Neo4j 浏览器" 按钮访问原生界面。
+本页面展示的是 LightRAG 知识库生成的图谱信息。
+数据来源于选定的知识库实例。
+支持通过实体名称进行模糊搜索,输入 "*" 可查看采样全图。