5.3 · Edge 抽取与消解（Edge Extraction and Resolution）

时序知识图谱与动态事实记忆 · 本章是 Graphiti DeepWiki 中文译文的独立章节页，保留原始链接、源码锚点、模块标签和章节层级。

项目Graphiti 章节5.3 状态全文译文模块图谱与关系、测试、发布与运维、界面与交互、模型调用与提供方适配

项目要点页2.5 参考项目项目章节目录Graphiti DeepWiki 原始章节Edge Extraction and Resolution 上一章5.2 下一章5.4

源码线索

graphiti_core/prompts/dedupe_edges.py
graphiti_core/prompts/dedupe_nodes.py
graphiti_core/prompts/extract_edges.py
graphiti_core/prompts/extract_nodes.py
graphiti_core/prompts/extract_nodes_and_edges.py
graphiti_core/prompts/lib.py
graphiti_core/prompts/summarize_nodes.py
graphiti_core/utils/maintenance/combined_extraction.py
graphiti_core/utils/maintenance/edge_operations.py
graphiti_core/utils/maintenance/node_operations.py

模块标签

图谱与关系
测试、发布与运维
界面与交互
模型调用与提供方适配
评测、反馈与人工复核

中文译文

Edge 抽取与消解（中文译文）

原始 DeepWiki 页面：https://deepwiki.com/getzep/graphiti/5.3-edge-extraction-and-resolution

翻译时间：2026-05-27T08:44:53.416Z

翻译模型：deepseek-chat

原文字符数：11730

项目：Graphiti (graphiti)

---

边提取与解析

边提取与解析管线

边提取与解析管线通过多阶段流程将片段内容转换为结构化的 EntityEdge 关系，该流程涉及基于大语言模型（LLM）的提取和用于解析的双时间逻辑。

整体管线流程

graph TD
    EpisodicNode["EpisodicNode"] --> extract_edges["extract_edges()"]
    EntityNodes["list[EntityNode]"] --> extract_edges
    PrevEpisodes["previous_episodes"] --> extract_edges
    edge_type_map["edge_type_map"] --> extract_edges
    edge_types["edge_types"] --> extract_edges

    extract_edges --> ExtractedEdges["list[EntityEdge]<br/>(新边)"]

    ExtractedEdges --> resolve_extracted_edges["resolve_extracted_edges()"]

    resolve_extracted_edges --> SearchRelated["search() 查找<br/>相关边"]
    resolve_extracted_edges --> SearchInvalidation["search() 查找<br/>失效候选边"]

    SearchRelated --> resolve_extracted_edge["resolve_extracted_edge()<br/>(通过 semaphore_gather 并行)"]
    SearchInvalidation --> resolve_extracted_edge

    resolve_extracted_edge --> ResolvedEdges["list[EntityEdge]<br/>(已解析)"]
    resolve_extracted_edge --> InvalidatedEdges["list[EntityEdge]<br/>(已失效)"]

    ResolvedEdges --> create_embeddings["create_entity_edge_embeddings()"]
    InvalidatedEdges --> create_embeddings

    create_embeddings --> GraphDriver["GraphDriver.upsert_edges()"]

来源：graphiti_core/utils/maintenance/edge_operations.py:116-288, graphiti_core/utils/maintenance/edge_operations.py:290-467

边提取过程

extract_edges 函数 graphiti_core/utils/maintenance/edge_operations.py:116-288 使用大语言模型（LLM）识别实体间的事实关系。该过程会验证提取的源名称和目标名称是否与提供的 EntityNode 列表匹配，并过滤掉自引用边 graphiti_core/utils/maintenance/edge_operations.py:214-250。

提取组件

组件	类型	描述
`episode`	`EpisodicNode \	list[EpisodicNode]`	要提取的片段。多个片段会被拼接处理 `graphiti_core/utils/maintenance/edge_operations.py:135-186`。
`nodes`	`list[EntityNode]`	已提取的实体，用于建立关系。用于名称验证 `graphiti_core/utils/maintenance/edge_operations.py:167-167`。
`edge_type_map`	`dict[tuple[str, str], list[str]]`	将 `(source_label, target_label)` 映射到允许的边类型 `graphiti_core/utils/maintenance/edge_operations.py:144-150`。
`edge_types`	`dict[str, type[BaseModel]]`	自定义边类型模式，包含属性和描述 `graphiti_core/utils/maintenance/edge_operations.py:151-164`。

来源：graphiti_core/utils/maintenance/edge_operations.py:116-164

大语言模型（LLM）提取与响应模型

提取过程使用 prompt_library.extract_edges.edge() graphiti_core/prompts/extract_edges.py:94-164，该函数指示大语言模型（LLM）使用 REFERENCE_TIME 来解析相对时间表达式 graphiti_core/prompts/extract_edges.py:108-110。对于多片段处理，大语言模型（LLM）会被指示使用 episode_indices 将事实归因到特定的片段索引 graphiti_core/prompts/extract_edges.py:48-52。

ExtractedEdges 响应模型

graph TD
    ExtractedEdges["ExtractedEdges"] --> edges["edges: list[Edge]"]

    Edge["Edge"] --> relation_type["relation_type: str<br/>(SCREAMING_SNAKE_CASE 格式)"]
    Edge --> source_entity_name["source_entity_name: str"]
    Edge --> target_entity_name["target_entity_name: str"]
    Edge --> fact["fact: str"]
    Edge --> valid_at["valid_at: str | None<br/>(ISO 8601 格式)"]
    Edge --> invalid_at["invalid_at: str | None<br/>(ISO 8601 格式)"]
    Edge --> episode_indices["episode_indices: list[int]"]

来源：graphiti_core/prompts/extract_edges.py:25-56, graphiti_core/utils/maintenance/edge_operations.py:214-215

边解析过程

resolve_extracted_edges 函数 graphiti_core/utils/maintenance/edge_operations.py:290-467 负责去重边、检测矛盾并管理时间有效性。

解析管线

graph TD
    ExtractedEdges["list[EntityEdge]<br/>(已提取)"] --> FastPath["快速路径：<br/>在提取的边内<br/>去重精确匹配"]

    FastPath --> Embeddings["create_entity_edge_embeddings()"]

    Embeddings --> GetValid["EntityEdge.get_between_nodes()<br/>(每条边并行)"]

    GetValid --> SearchRelated["search() 使用<br/>EDGE_HYBRID_SEARCH_RRF"]

    Embeddings --> SearchInvalidation["search() 查找<br/>失效候选边"]

    SearchRelated --> ParallelResolve["semaphore_gather(<br/>resolve_extracted_edge())"]
    SearchInvalidation --> ParallelResolve

    ParallelResolve --> ResolvedEdges["list[EntityEdge]<br/>(已解析)"]
    ParallelResolve --> InvalidatedEdges["list[EntityEdge]<br/>(已失效)"]

来源：graphiti_core/utils/maintenance/edge_operations.py:290-467

快速路径去重

在并行处理之前，resolve_extracted_edges 会在新提取的边内执行精确去重，以最小化冗余的大语言模型（LLM）调用 graphiti_core/utils/maintenance/edge_operations.py:298-312。

# 快速路径：去重精确匹配
seen: dict[tuple[str, str, str], EntityEdge] = {}
deduplicated_edges: list[EntityEdge] = []

for edge in extracted_edges:
    key = (
        edge.source_node_uuid,
        edge.target_node_uuid,
        _normalize_string_exact(edge.fact),
    )
    if key not in seen:
        seen[key] = edge
        deduplicated_edges.append(edge)

来源：graphiti_core/utils/maintenance/edge_operations.py:298-312

单条边解析

resolve_extracted_edge 函数 graphiti_core/utils/maintenance/edge_operations.py:507-711 会对照已有的图知识处理每条边。

解析阶段

阶段 1：快速路径逐字检查 如果相同源节点和目标节点之间已存在一条边，且其事实在语义上相同（通过 _normalize_string_exact 规范化后），则会复用该边，并将新片段的 UUID 追加到其 episodes 列表中 graphiti_core/utils/maintenance/edge_operations.py:545-557。

阶段 2：大语言模型（LLM）去重与矛盾检测 大语言模型（LLM）使用 prompt_library.dedupe_edges.resolve_edge() graphiti_core/prompts/dedupe_edges.py:43-101 判断新事实是否是已有事实的重复，或者是否与之前的信息矛盾。

graph TD
    PrepareContext["准备上下文：<br/>existing_edges（相关边）<br/>new_edge（提取的边）<br/>edge_invalidation_candidates"] --> LLMCall["llm_client.generate_response(<br/>prompt_library.dedupe_edges.resolve_edge(),<br/>EdgeDuplicate)"]

    LLMCall --> EdgeDuplicate["EdgeDuplicate 响应"]

    EdgeDuplicate --> duplicate_facts["duplicate_facts: list[int]"]
    EdgeDuplicate --> contradicted_facts["contradicted_facts: list[int]"]

来源：graphiti_core/utils/maintenance/edge_operations.py:560-637, graphiti_core/prompts/dedupe_edges.py:24-33

阶段 3：属性提取 如果识别到自定义边类型，并且其 Pydantic 模型中定义了属性，则会使用大语言模型（LLM）从事实文本中提取这些特定属性 graphiti_core/utils/maintenance/edge_operations.py:657-677。

时间矛盾解析

resolve_edge_contradictions 函数 graphiti_core/utils/maintenance/edge_operations.py:469-505 实现了双时间逻辑来管理冲突事实。

时间矛盾逻辑

当新事实与已有事实矛盾时，已有事实的 invalid_at 时间戳会被更新为新事实的 valid_at 时间戳，其 expired_at 会被设置为当前系统时间 graphiti_core/utils/maintenance/edge_operations.py:494-497。

graph TD
    ResolvedEdge["resolved_edge<br/>(新边)"] --> GetCandidates["invalidation_candidates<br/>(被矛盾的边)"]

    GetCandidates --> Iterate["遍历每条候选边"]

    Iterate --> Condition1{"edge.invalid_at <=<br/>resolved.valid_at?"}

    Condition1 -->|是| NoInvalidation["跳过：已失效"]
    Condition1 -->|否| Condition3{"edge.valid_at <<br/>resolved.valid_at?"}

    Condition3 -->|是| Invalidate["edge.invalid_at =<br/>resolved.valid_at<br/>edge.expired_at =<br/>utc_now()"]
    Condition3 -->|否| ExpireNew["新边已被<br/>历史信息取代"]

来源：graphiti_core/utils/maintenance/edge_operations.py:469-505

边过期

系统还会检查*新*解析的边是否应立即过期。如果某个矛盾候选边的 valid_at 时间戳比新边的 valid_at 更近，则新边在入库时会被标记为已过期 graphiti_core/utils/maintenance/edge_operations.py:688-703。

边维护操作

Graphiti 维护结构边以组织图谱和片段上下文。

片段边与社区边

build_episodic_edges：在 EntityNode 和 EpisodicNode 之间创建 MENTIONED_IN 边（由 EpisodicEdge 类表示）graphiti_core/utils/maintenance/edge_operations.py:51-95。
build_community_edges：在 CommunityNode 和 EntityNode 之间创建 HAS_MEMBER 边（由 CommunityEdge 类表示）graphiti_core/utils/maintenance/edge_operations.py:98-113。

重复边过滤

filter_existing_duplicate_of_edges 函数 graphiti_core/utils/maintenance/edge_operations.py:713-775 通过检查数据库中节点对之间是否已存在关系，来防止插入冗余的 IS_DUPLICATE_OF 边。

来源：graphiti_core/utils/maintenance/edge_operations.py:51-113, graphiti_core/utils/maintenance/edge_operations.py:713-775