CAMEL 图存储（Graph Storages）模块全解析：从抽象基类到 Neo4j 知识图谱落地-平芜编程栈

CAMEL 图存储（Graph Storages）模块全解析：从抽象基类到 Neo4j 知识图谱落地

【免费下载链接】camel🐫 CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel

导读

本文以 CAMEL 官方 API 文档入口 docs/camel.storages.graph_storages.rst 为骨架，深入剖析camel.storages.graph_storages包的设计与实现。该模块为 CAMEL 多智能体框架提供统一的知识图谱存储抽象：一套面向实体—关系三元组的抽象基类、一套与文档解析无缝衔接的图数据模型（Node/Relationship/GraphElement），以及 Neo4j、NebulaGraph 两个生产级后端实现。读完本文，你将掌握如何在 CAMEL 中初始化图数据库连接、管理图 Schema、执行三元组的增删查、批量导入文档解析出的图元素，并理解KnowledgeGraphAgent等上层组件与图存储的协作方式。

一、模块全景：`camel.storages.graph_storages`的组成

根据 RST 文档的结构，该包包含三个核心子模块与一个包级导出入口：

RST 文档小节	对应源码文件	职责
`camel.storages.graph_storages.base`	camel/storages/graph_storages/base.py	定义`BaseGraphStorage`抽象基类
`camel.storages.graph_storages.graph_element`	camel/storages/graph_storages/graph_element.py	定义`Node`、`Relationship`、`GraphElement`图数据模型
`camel.storages.graph_storages.neo4j_graph`	camel/storages/graph_storages/neo4j_graph.py	Neo4j 后端实现
Module contents（包内容）	camel/storages/graph_storages/init.py	统一导出对外 API

从 camel/storages/graph_storages/init.py 可以看到，包对外共导出 4 个符号，其中NebulaGraph同样通过__init__.py对外暴露（源码见 camel/storages/graph_storages/nebula_graph.py）：

from .base import BaseGraphStorage from .graph_element import GraphElement from .nebula_graph import NebulaGraph from .neo4j_graph import Neo4jGraph __all__ = ['BaseGraphStorage', 'GraphElement', 'Neo4jGraph', 'NebulaGraph']

设计要点：上层组件（如知识图谱 Agent）只面向BaseGraphStorage抽象编程，具体后端通过子类注入，因此切换 Neo4j 与 NebulaGraph 时业务代码无需改动。

二、统一抽象：`BaseGraphStorage`接口定义

camel/storages/graph_storages/base.py 中，BaseGraphStorage继承ABC，定义了所有图存储后端必须实现的 7 个抽象成员：

抽象成员	签名	说明
`get_client`（property）	`-> Any`	获取底层图存储客户端（如 Neo4j Driver / Nebula Session）
`get_schema`（property）	`-> str`	获取图的 Schema 字符串，用于向 LLM 描述图谱结构
`get_structured_schema`（property）	`-> Dict[str, Any]`	获取结构化 Schema（节点属性、关系属性、关系列表、元数据）
`refresh_schema()`	`-> None`	刷新图 Schema 信息
`add_triplet(subj, obj, rel)`	`-> None`	添加主体—客体—关系三元组
`delete_triplet(subj, obj, rel)`	`-> None`	删除指定三元组
`query(query, params=None)`	`-> List[Dict[str, Any]]`	执行查询语句，返回字典列表（每行一条结果）

这套接口的价值在于：三元组是知识图谱最小语义单元，add_triplet/delete_triplet让 Agent 可以原子地增删知识；而get_schema输出的人类可读字符串（形如Node properties are the following: ...）通常会被直接注入提示词，帮助大模型生成合法的图查询语句——这是 CAMEL 中图谱类任务的关键链路。

三、图数据模型：`Node`/`Relationship`/`GraphElement`

图元素模型定义在 camel/storages/graph_storages/graph_element.py，全部基于pydantic.BaseModel构建。

3.1`Node`：图中的节点

class Node(BaseModel): id: Union[str, int] # 唯一标识 type: str = "Node" # 节点类型（对应 Neo4j Label / Nebula Tag） properties: dict = Field(default_factory=dict) # 附加属性与元数据

3.2`Relationship`：有向关系

class Relationship(BaseModel): subj: Node # 关系的起点/主体 obj: Node # 关系的终点/客体 type: str = "Relationship" # 关系类型 timestamp: Optional[str] = None # 关系时间戳（可选） properties: dict = Field(default_factory=dict) # 关系附加属性

3.3`GraphElement`：文档驱动的图片段

class GraphElement(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True) nodes: List[Node] relationships: List[Relationship] source: Element # 图信息的来源文档元素（unstructured）

source字段的类型是unstructured.documents.elements.Element，表明图元素通常从非结构化文档解析而来。源码中通过__post_init__检查：若环境中未安装unstructured包，使用source属性时会抛出ImportError。这一点在 test/storages/graph_storages/test_graph_element.py 中有直接验证：测试用UnstructuredIO().create_element_from_text("sample text")构造source，并断言GraphElement能正确承载节点与关系列表。

四、Neo4j 后端：`Neo4jGraph`深度解析

camel/storages/graph_storages/neo4j_graph.py 中的Neo4jGraph是 RST 文档着墨最多的模块，也是默认推荐的生产后端。其类注释明确说明设计参考了 LangChain 与 LlamaIndex 的相关工作。

4.1 构造函数与参数

Neo4jGraph( url: str, # Neo4j 服务地址 username: str, # 认证用户名 password: str, # 认证密码 database: str = "neo4j", # 数据库名，默认 neo4j timeout: Optional[float] = None, # 事务超时秒数，用于终止长查询 truncate: bool = False, # 是否裁剪超过 LIST_LIMIT(128) 元素的结果列表 )

构造函数上标注了@dependencies_required('neo4j')，未安装neo4j驱动时会给出明确提示。三个连接参数均支持环境变量兜底（neo4j_graph.py）：

url = os.environ.get("NEO4J_URI") or url username = os.environ.get("NEO4J_USERNAME") or username password = os.environ.get("NEO4J_PASSWORD") or password

4.2 连接校验与 APOC 依赖

初始化时自动执行两类校验（neo4j_graph.py）：

连通性校验：调用driver.verify_connectivity()，地址错误抛ValueError("Could not connect to Neo4j database...")，用户名密码错误抛对应AuthError提示；
APOC 插件校验：Schema 刷新依赖apoc.meta.data()过程，若未安装 APOC 插件或该过程被禁用，会抛出ValueError明确要求安装 APOC 并允许该调用。

使用前提：Neo4j 后端依赖 APOC 插件（Schema 自动发现、批量导入均使用apoc.*过程）；图采样功能还额外依赖 Graph Data Science（GDS）库。

4.3 Schema 管理与`refresh_schema`

refresh_schema()（neo4j_graph.py）通过三条基于apoc.meta.data()的 Cypher 查询（模块顶部的NODE_PROPERTY_QUERY、REL_PROPERTY_QUERY、REL_QUERY）收集节点属性、关系属性与关系集合，同时用SHOW CONSTRAINTS/SHOW INDEXES获取约束与索引；只读用户无权限时这两个查询会被安全降级为空列表。之后同时生成两种形态：

结构化 Schema（get_structured_schema）：{"node_props": {...}, "rel_props": {...}, "relationships": [...], "metadata": {"constraint": [...], "index": [...]}}；
文本 Schema（get_schema）：拼装为三段人类可读文本，形如：

Node properties are the following: LabelA {property_a: STRING} Relationship properties are the following: REL_TYPE {rel_prop: STRING} The relationships are the following: (:LabelA)-[:REL_TYPE]->(:LabelB)

这段文本正是喂给大模型、辅助其生成合法 Cypher 的关键上下文。get_schema默认读取缓存，只有传入refresh=True或首次调用时才强制刷新。

模块顶部的EXCLUDED_LABELS = ["Excluded_Label_A", "Excluded_Label_B"]与EXCLUDED_RELS = ["Excluded_Rel_A"]允许在 Schema 发现时过滤特定标签/关系；test/storages/graph_storages/test_neo4j_graph.py 中的test_neo4j_filtering_labels验证了这些标签会被正确排除在node_props与relationships之外。

4.4 三元组操作：增、删、查

新增add_triplet(subj, obj, rel, timestamp=None)（neo4j_graph.py）使用MERGE语义实现幂等写入：节点统一挂到基础标签BASE_ENTITY_LABEL = "__Entity__"（下划线会被剥离为Entity），关系类型则经过rel.replace(" ", "_").upper()规范化（空格转下划线并大写），并可选写入timestamp属性。

删除delete_triplet（neo4j_graph.py）是一个"级联清理"流程：先删除关系，再分别检查主体与客体是否还残留边（_check_edges），若无边则连同孤立实体一并删除（_delete_entity），避免图谱中残留孤儿节点。

查询get_triplet(subj=None, obj=None, rel=None)（neo4j_graph.py）支持按主体、客体、关系类型的任意组合过滤，任一参数为None时匹配任意值，返回[{subj, obj, rel, timestamp}, ...]。

4.5 批量导入：`add_graph_elements`

这是将"文档 → 图"链路打通的核心方法（neo4j_graph.py），入参为List[GraphElement]，两个关键开关：

include_source=True：将来源文档元素以:Element节点入库，并用MENTIONS关系把每个实体节点链接回其出处（INCLUDE_DOCS_QUERY），实现知识溯源。来源元素的合并键优先取元数据中的element_id，缺失时退化为对图元素整体求 MD5；
base_entity_label=True：为每个新建节点追加__Entity__基础标签并自动创建id唯一约束（CREATE CONSTRAINT IF NOT EXISTS），借助基础标签上的索引提升大批量导入性能。

批量导入的内部实现（_get_node_import_query/_get_rel_import_query）展示了两种导入策略：base_entity_label=True时先以基础标签MERGE节点，再用apoc.create.addLabels追加真实标签；False时直接以apoc.merge.node([row.type], {id: row.id}, ...)按类型合并。关系导入统一使用apoc.merge.relationship保证幂等。test/storages/graph_storages/test_neo4j_graph.py 中的test_neo4j_add_data、test_neo4j_add_data_source、test_neo4j_add_data_base、test_neo4j_add_data_base_source四个用例分别覆盖了该方法的四种组合，验证节点标签、Element来源节点以及约束创建均符合预期。

4.6 查询执行：`query`与超时、截断控制

query(query, params=None)（neo4j_graph.py）在指定数据库的 Session 中执行 Cypher 并返回List[Dict]；Cypher 语法错误会被包装为ValueError("Generated Cypher Statement is not valid...")。构造Neo4jGraph时传入的timeout会通过neo4j.Query(text=query, timeout=self.timeout)生效，用于终止长事务；truncate=True时结果会经过_value_truncate递归裁剪——长度达到LIST_LIMIT = 128的列表（典型如嵌入向量）会被移除，从而降低结果噪声与下游计算开销。test/storages/graph_storages/test_neo4j_graph.py 的test_neo4j_timeout与test_neo4j_truncate_values分别对这两个行为做了回归验证。

4.7 GDS 图采样：RWR 与 CNARW

面向大规模图的可扩展性需求，Neo4jGraph还封装了两个基于 Neo4j Graph Data Science 库的采样算法：

random_walk_with_restarts(graph_name, sampling_ratio, start_node_ids, restart_probability=0.1, node_label_stratification=False, relationship_weight_property=None)：带重启的随机游走采样，采样结果图命名为{graph_name}_sampled；
common_neighbour_aware_random_walk(...)：公共邻居感知随机游走，结果图命名为{graph_name}_sampled_cnarw，可在采样时保持原图的节点标签分布（node_label_stratification=True）或按关系属性加权（relationship_weight_property）。

两者在执行前都会先调用gds.version()探测 GDS 库是否可用，未安装时抛明确提示；测试中对应用例标注了Skipping since need GDS installation（见 test/storages/graph_storages/test_neo4j_graph.py），说明该能力属于可选扩展。

五、NebulaGraph 后端：另一套实现范式

虽然 RST 正文只列出neo4j_graph子模块，但包级导出（Module contents）同时包含NebulaGraph（camel/storages/graph_storages/nebula_graph.py）。其构造参数为：

NebulaGraph(host, username, password, space, port=9669, timeout=10000)

与 Neo4j 后端形成鲜明对照的实现细节：

自动建空间：初始化时自动执行CREATE SPACE IF NOT EXISTS {space} (vid_type=FIXED_STRING(30))，随后USE {space}采用最多MAX_RETRIES = 5次、每次间隔 3 秒的重试策略（应对图空间创建后的生效延迟）；
自愈式 Schema：add_node/add_triplet内部通过ensure_tag_exists、ensure_edge_type_exists自动创建 Tag 与 Edge Type，节点 ID 与标签会经正则清洗（仅保留字母、数字与中日韩字符）；
时间标签：Tag/Edge 可带time_label DATETIME属性，格式须为YYYY-MM-DDThh:mm:ss（由_validate_time_label校验）；
Schema 发现：基于 nGQL 的SHOW TAGS/DESCRIBE TAG/SHOW EDGES/SHOW TAG INDEXES实现get_schema与get_structured_schema，输出文本格式与 Neo4j 后端保持一致（Node properties are the following: ...）；
依赖：构造函数标注@dependencies_required('nebula3')，需安装nebula3驱动。

六、与上层组件的联动：知识图谱 Agent

GraphElement并非孤立的数据结构，它是 CAMEL 知识图谱能力的中间产物。camel/agents/knowledge_graph_agent.py 中：

图谱抽取方法（约 L153-L166）返回类型为Union[str, GraphElement]——当parse_graph_elements=True时，LLM 抽取出的实体与关系文本会被_parse_graph_elements解析成GraphElement（L219-L274），随后即可直接交给Neo4jGraph.add_graph_elements()批量入库。

由此形成的完整链路为：非结构化文档 →UnstructuredIO加载 → LLM 抽取实体关系 →GraphElement承载 → 图存储批量导入 → Schema 反馈给 LLM 支撑后续图查询。test/agents/test_knowledge_agent.py与 test/storages/graph_storages/test_neo4j_graph.py 分别从 Agent 侧与存储侧对这条链路做了测试覆盖。

七、测试与可验证性

图存储模块拥有完整的三套测试，可直接作为使用参考与回归保障：

test/storages/graph_storages/test_graph_element.py：覆盖Node、Relationship、GraphElement的初始化与空列表场景；
test/storages/graph_storages/test_neo4j_graph.py：覆盖 Schema 发现、超时、结果截断、add_graph_elements四种模式、标签过滤与 GDS 采样；测试通过NEO4J_URI/NEO4J_USERNAME/NEO4J_PASSWORD三个环境变量连接真实数据库，连接失败时自动pytest.skip，保证无环境也能安全跑通其他用例；
test/storages/graph_storages/test_nebula_graph.py：覆盖 NebulaGraph 的建空间、Tag/Edge 自动创建与三元组写入。

八、小结

camel.storages.graph_storages包以"统一抽象 + 多后端实现"的方式，为 CAMEL 的多智能体系统提供了可靠的知识图谱持久化能力：BaseGraphStorage定义了与 LLM 协作友好的最小接口（Schema 描述 + 三元组操作 + 通用查询），GraphElement完成了文档解析与图数据库之间的数据桥接，Neo4jGraph与NebulaGraph则分别在 Cypher/APOC/GDS 生态与 nGQL 生态中落地了完整实现。无论你是想为 Agent 搭建带溯源的知识库，还是需要在大图上做高效采样，都可以从 camel/storages/graph_storages/init.py 暴露的四个 API 入手，快速接入这套图存储体系。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

CAMEL 图存储（Graph Storages）模块全解析：从抽象基类到 Neo4j 知识图谱落地