Review Conflicting Facts

A minimal guide to reviewing conflicting facts with improve(). Use it when two sources in the same dataset assert different values for one attribute — here, two documents naming different CEOs for the same company — and you want Cognee to record the disagreement, ordered by each source’s date, instead of silently keeping both.

Before You Start

Complete Quickstart to understand basic operations
Ensure you have LLM Providers configured — the review stage is skipped when no LLM is available
Read Improve for how improve() runs its stages, including review_conflicts
Keep EDGE_EVIDENCE_ENABLED at its default true — the review stage is skipped when edge evidence is off

Code in Action

import asyncio
from uuid import uuid4

import cognee
from cognee.context_global_variables import set_database_global_context_variables
from cognee.infrastructure.databases.graph import get_graph_engine
from cognee.modules.data.methods import get_authorized_existing_datasets
from cognee.modules.engine.operations.setup import setup
from cognee.modules.graph.utils.fact_conflicts import read_conflict_marks
from cognee.modules.improve.config import get_improve_config
from cognee.modules.users.methods import get_default_user
from cognee.tasks.ingestion.data_item import DataItem


async def main():
    await setup()
    dataset_name = f"review_conflicts_{uuid4().hex[:8]}"
    user = await get_default_user()
    date_key = get_improve_config().effective_date_key
    for text, date, label in [
        ("Acme is led by CEO Alice.", "2020-05-01", "2020 directory"),
        ("Acme is led by CEO Bob.", "2026-06-01", "2026 board notice"),
    ]:
        await cognee.remember(
            DataItem(text, label=label, external_metadata={date_key: date}, literal_text=True),
            dataset_name=dataset_name,
            user=user,
            self_improvement=False,
            custom_prompt="Extract the company Acme and the named person. Connect Acme to its CEO with has_ceo.",
        )

    result = await cognee.improve(dataset=dataset_name, user=user, review_conflicts=True)
    print(dataset_name, result.stage("review_conflicts"))
    dataset = (
        await get_authorized_existing_datasets(
            user=user, datasets=[dataset_name], permission_type="read"
        )
    )[0]
    async with set_database_global_context_variables(dataset.id, dataset.owner_id):
        graph = await get_graph_engine()
        nodes, edges = await graph.get_graph_data()
        conflict_ids = set()
        for node_id, properties in nodes:
            if properties.get("type") == "FactConflict" and str(
                properties.get("dataset_id")
            ) == str(dataset.id):
                conflict_ids.add(str(node_id))
                print("Conflict:", properties)
        for source, target, relationship, properties in edges:
            if str(source) in conflict_ids:
                print("Link:", relationship, target, properties)
            marks = [
                mark
                for mark in read_conflict_marks(properties)
                if mark.get("conflict_id") in conflict_ids
            ]
            if marks:
                print("Fact:", properties.get("edge_text"), marks)

    repeat = await cognee.improve(dataset=dataset_name, user=user, review_conflicts=True)
    print("Unchanged repeat:", repeat.stage("review_conflicts"))


if __name__ == "__main__":
    asyncio.run(main())

The dataset is kept after the run so you can inspect it, and every run uses a fresh dataset name.

What Just Happened

Step 1: Remember Dated Facts

date_key = get_improve_config().effective_date_key
for text, date, label in [
    ("Acme is led by CEO Alice.", "2020-05-01", "2020 directory"),
    ("Acme is led by CEO Bob.", "2026-06-01", "2026 board notice"),
]:
    await cognee.remember(
        DataItem(text, label=label, external_metadata={date_key: date}, literal_text=True),
        dataset_name=dataset_name,
        user=user,
        self_improvement=False,
        custom_prompt="Extract the company Acme and the named person. Connect Acme to its CEO with has_ceo.",
    )

Each source is wrapped in a DataItem whose external_metadata carries the date the source’s claims hold as of, under the key the review reads (effective_date by default, configurable with IMPROVE_EFFECTIVE_DATE_KEY). self_improvement=False keeps remember() from improving on its own, so the review runs only when the script asks for it.

Step 2: Run the Conflict Review

result = await cognee.improve(dataset=dataset_name, user=user, review_conflicts=True)
print(dataset_name, result.stage("review_conflicts"))

review_conflicts=True opts in to the review_conflicts stage of improve(), which is off by default (IMPROVE_REVIEW_CONFLICTS=true enables it too). An LLM reviews the dataset’s source facts and their dates, stores each disagreement it accepts as a FactConflict node, and updates the affected entity descriptions. result.stage("review_conflicts") returns that stage’s result, including its status.

Step 3: Inspect the Stored Conflicts

for node_id, properties in nodes:
    if properties.get("type") == "FactConflict" and str(
        properties.get("dataset_id")
    ) == str(dataset.id):
        conflict_ids.add(str(node_id))
        print("Conflict:", properties)
for source, target, relationship, properties in edges:
    if str(source) in conflict_ids:
        print("Link:", relationship, target, properties)
    marks = [
        mark
        for mark in read_conflict_marks(properties)
        if mark.get("conflict_id") in conflict_ids
    ]
    if marks:
        print("Fact:", properties.get("edge_text"), marks)

Each FactConflict node describes one attribute of one entity with competing values, and its outgoing edges link it to the entity it is about, the competing values, and the source chunks it cites. The facts themselves carry conflict marks — read_conflict_marks() reads them on any graph backend — that record each fact’s status within the conflict: current, superseded, or conflicting.

Step 4: Repeat on Unchanged Data

repeat = await cognee.improve(dataset=dataset_name, user=user, review_conflicts=True)
print("Unchanged repeat:", repeat.stage("review_conflicts"))

The review keeps a watermark per dataset and re-reviews only entities whose facts changed since the last run. With no new facts and no unfinished conflicts to repair, the second run reports the stage as already_completed instead of calling the LLM again.

Self-Improving Memory Quickstart

How improve() runs its stages and what each one does.

Temporal Knowledge Graph

The other time axis: when a fact happened, not when its source dated it.

Getting Started

Core Concepts

Setup Configuration

Guides

Cookbooks

Examples

CLI

OSS

Before You Start

Code in Action

What Just Happened

Step 1: Remember Dated Facts

Step 2: Run the Conflict Review

Step 3: Inspect the Stored Conflicts

Step 4: Repeat on Unchanged Data

Self-Improving Memory Quickstart

Temporal Knowledge Graph

​Before You Start

​Code in Action

​What Just Happened

​Step 1: Remember Dated Facts

​Step 2: Run the Conflict Review

​Step 3: Inspect the Stored Conflicts

​Step 4: Repeat on Unchanged Data

Self-Improving Memory Quickstart

Temporal Knowledge Graph

Before You Start

Code in Action

What Just Happened

Step 1: Remember Dated Facts

Step 2: Run the Conflict Review

Step 3: Inspect the Stored Conflicts

Step 4: Repeat on Unchanged Data