Graphiti 如何用 Pydantic 模型定义自定义 entity_types 与 edge_types 约束图谱结构【免费下载链接】graphitiBuild Real-Time Knowledge Graphs for AI Agents项目地址: https://gitcode.com/GitHub_Trending/grap/graphiti默认情况下Graphiti 从 episode 中自由抽取实体与关系类型由数据自然产生learned ontology。当你的业务需要图谱结构服从一套预定义词表——例如实体只允许是Person、City关系只允许是IS_PRESIDENT_OF——时可以在add_episode/add_episode_bulk调用中传入entity_types、edge_types、edge_type_map三个参数把抽取约束到你定义的 Pydantic 模型上。本文基于仓库内的 podcast 示例 和 实体排除集成测试 给出完整操作路径。准备条件按 README.md 的要求Python 3.10安装graphiti-core使用 FalkorDB 时安装graphiti-core[falkordb]一个可用的图数据库Neo4j 5.26 / FalkorDB 1.1.2 / Amazon NeptuneKuzu 已弃用OPENAI_API_KEYGraphiti 默认用 OpenAI 做 LLM 推理与 embedding。另外注意 README 中的提示Graphiti 依赖结构化JSON输出完成实体/边抽取与去重最好使用支持 Structured Output 的模型服务OpenAI、Anthropic、Gemini否则可能出现输出 schema 不正确导致的摄取失败。第一步用 Pydantic 模型定义实体类型与边类型entity_types是一个dict[str, type[BaseModel]]键是实体类型名值是 Pydantic 模型。模型的字段会成为实体节点可抽取的属性字段上的description会进入抽取 prompt 指导 LLM。下面这两个模型直接取自 examples/podcast/podcast_runner.pyfrom pydantic import BaseModel, Field class Person(BaseModel): A human person, fictional or nonfictional. first_name: str | None Field(..., descriptionFirst name) last_name: str | None Field(..., descriptionLast name) occupation: str | None Field(..., descriptionThe persons work occupation) class City(BaseModel): A city country: str | None Field(..., descriptionThe country the city is in)边类型的模型同样继承BaseModel。只写 docstring不加字段的模型在示例中被用作“文档型”边类型class IsPresidentOf(BaseModel): Relationship between a person and the entity they are a president of class InterpersonalRelationship(BaseModel): A relationship between two people (e.g., knows, works with, interviewed) class LocatedIn(BaseModel): A relationship indicating something is located in or associated with a placeadd_episode入口处会先执行validate_entity_types见 entity_types_utils.py如果你的模型字段名与EntityNode的保留字段名冲突会抛出EntityTypeValidationError。所以定义字段时避免使用EntityNode已有字段名。第二步把类型传给 add_episodeedge_type_map声明哪种实体类型对之间允许出现哪些边类型。示例中同一个边类型可以出现在多组节点类型对里from datetime import datetime, timezone from graphiti_core import Graphiti from graphiti_core.driver.neo4j_driver import Neo4jDriver from graphiti_core.nodes import EpisodeType edge_types { IS_PRESIDENT_OF: IsPresidentOf, INTERPERSONAL_RELATIONSHIP: InterpersonalRelationship, LOCATED_IN: LocatedIn, } edge_type_map { (Person, Entity): [IS_PRESIDENT_OF, INTERPERSONAL_RELATIONSHIP], (Person, Person): [INTERPERSONAL_RELATIONSHIP], (Person, City): [LOCATED_IN], (Entity, City): [LOCATED_IN], }其中Entity是兜底的默认实体类型可作为通配符出现在端点上mcp_server/config/config.yaml 中注释说明Entity acts as a wildcard for either endpoint。初始化 Graphiti 并写入一条 episode。下面这条 episode 内容取自 tests/test_entity_exclusion_int.py便于核对结果driver Neo4jDriver( uribolt://localhost:7687, userneo4j, passwordpassword, ) client Graphiti(graph_driverdriver) await client.build_indices_and_constraints() result await client.add_episode( nameBusiness Meeting, episode_bodyJohn Smith works at Acme Corporation in New York. The weather is nice today., source_descriptionMeeting notes, reference_timedatetime.now(timezone.utc), entity_types{Person: Person, City: City}, edge_typesedge_types, edge_type_mapedge_type_map, group_iddemo, )group_id是图分区标识不传时使用驱动对应的默认 group id。add_episode返回AddEpisodeResults定义于 graphiti_core/graphiti.py包含本次写入的episode、nodes、edges、episodic_edges等可以直接检查节点 labels 与边名。几个行为细节均来自源码若传了edge_types但没传edge_type_map默认映射为{(Entity, Entity): list(edge_types.keys())}即所有自定义边类型在任意实体间可用。excluded_entity_types参数可整体排除某些类型不入库包括默认的Entitygraphiti_core/graphiti.py 中add_episode的参数说明。批量写入用add_episode_bulk(raw_episodes, entity_types..., edge_types..., edge_type_map...)参数含义相同episode 用RawEpisode对象构造见 podcast_runner.py 的use_bulk分支。还可以传custom_extraction_instructions追加到实体/边抽取 prompt 里做进一步引导。验证检查结果节点的 labelspodcast_runner.py 的做法是摄取后用混合搜索回查边结果打印[edge.name] edge.fact节点结果打印node.name (labels)。最小验证代码from graphiti_core.search.search_config_recipes import NODE_HYBRID_SEARCH_RRF search_results await client.search_( queryJohn Smith Acme Corporation, group_ids[demo], configNODE_HYBRID_SEARCH_RRF.model_copy(update{limit: 5}), ) for node in search_results.nodes: print(node.name, node.labels)实体排除集成测试 给出了可对照的断言方式每个结果节点仍带有基础Entitylabel同时应携带它被分类到的自定义类型 label如Person、Organization。你可以用同样的断言判断类型约束是否生效——分类不中的实体不会带自定义 label。可选分支在 MCP Server 中用 YAML 声明类型如果你的使用形态是 mcp_server类型不走 Python 代码而是在config.yaml的graphiti段声明。mcp_server/config/config.yaml 中已给出三类键的结构graphiti: entity_types: - name: Person description: An individual human referenced in the content # 边类型名称若与 src/models/edge_types.py 中注册的模型匹配则使用富 Pydantic 模型 # 否则由下方 description 构建文档型模型 # edge_types: # - name: WorksFor # description: Employment or membership of a person in an organization # 约束哪些边类型可连接哪些实体类型Entity 作为通配符 # edge_type_map: # - source: Person # target: Organization # edge_types: [WorksFor]YAML 形式只能给namedescription不能像核心库那样定义带字段的 Pydantic 模型需要属性级抽取时仍应使用graphiti-core的代码路径。限制与注意点模型必须可靠支持 Structured Output。类型定义会转换成抽取/去重的响应 schema小模型或本地模型经常产出不符合 schema 的 JSON表现为摄取失败README 与 mcp_server README 均有说明。字段命名冲突直接报错自定义模型字段名不能与EntityNode字段重名否则EntityTypeValidationError在add_episode一开始就抛出。edge_type_map只约束允许的组合文档没有承诺它会阻止模型输出映射之外的边它的作用是把候选边类型按节点类型对注入 prompt 与抽取流程。示例代码中的clear_data(client.driver)会清空图中的数据podcast_runner.py 用它重置演示环境在自己的图数据上运行前不要照抄这一步。完整可运行的参照实现是 examples/podcast/podcast_runner.py它定义了Person/City实体类型与三组边类型逐条写入 podcast 消息最后用search与search_验证抽取结果可直接作为你的模板替换 episode 内容后运行。【免费下载链接】graphitiBuild Real-Time Knowledge Graphs for AI Agents项目地址: https://gitcode.com/GitHub_Trending/grap/graphiti创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
