Typed Claim Extraction

You have text where the facts belong to people — who claimed what, and how sure they were — and you want the graph to record exactly that, in your own node types with your own fields, rather than whatever cognee’s default extraction happens to produce.

What You’ll Build

A paragraph of science history is ingested into a dataset, then a custom pipeline of three chained tasks turns it into a graph: the first reads the ingested document and asks the LLM for the people and claims it can find, the second asks which person each claim belongs to and attaches them, and the third writes the resulting objects to the graph and vector stores and prints a summary of what it stored. What comes out is a small graph of Person nodes, each linked to the ScientificClaim nodes it owns, with every node stamped with the content hash of the document it came from and the claim text and person names indexed for semantic search — so the script can close by asking “Who worked on gravity?” and getting an answer back from the graph it just built. The complete runnable script is examples/demos/custom_pipelines/custom_pipeline_single_object_example.py — this page walks through its key moments rather than reproducing it.

Features in Play

Custom Tasks and Pipelines — Task wraps each step and run_custom_pipeline runs them in order against a dataset, feeding each task’s return value to the next
Add — add() ingests the text into a dataset first, which is what gives the pipeline a document with a content hash to work from and a place for recall to look
DataPoints — Person and ScientificClaim are the graph’s node types; the Embeddable and Dedup annotations decide which fields get embedded and which give a node its identity
Custom Data Models — add_data_points persists the typed objects and their nesting as nodes and edges, without a cognify pass
Low-Level LLM — LLMGateway.acreate_structured_output runs both extraction passes against plain Pydantic response models
Recall — one GRAPH_COMPLETION query scoped to the dataset proves the hand-built nodes are searchable like any other memory

What to Expect

The excerpts below come from one real run, trimmed of log lines. Two of the three tasks and the recall are live LLM calls, so the roles, claim wording, confidences, and answer vary from run to run; the shape of the output does not. The pipeline’s summary is the first thing the script prints, one block per person. Every line ends in the same twelve-character source_hash: all three people were extracted from the one document that was ingested, and the pipeline stamped its content hash onto each node it stored. A person with no attributed claims would show (no claims linked) instead of an indented list.

Albert Einstein (Physicist) [source_hash: f9bf9de04a56]
  - Albert Einstein published the theory of general relativity in 1915. [confidence: 1.0]
  - General relativity describes gravity as spacetime curvature. [confidence: 1.0]
Marie Curie (Physicist and chemist) [source_hash: f9bf9de04a56]
  - Marie Curie discovered polonium and radium. [confidence: 1.0]
  - Marie Curie won Nobel Prizes in both physics and chemistry. [confidence: 1.0]
Niels Bohr (Physicist) [source_hash: f9bf9de04a56]
  - Niels Bohr proposed the atomic model with quantized electron orbits in 1913. [confidence: 1.0]

The recall answers from the hand-built graph. The entry is a ResponseGraphEntry carrying dataset_name='science_claims', which is the proof that the custom pipeline’s nodes landed in the dataset and are reachable through the ordinary retrieval path. If this section instead shows a ResponseMarkerEntry saying memory is warming up, the nodes were stored outside a dataset and recall cannot see them.

--- Recall: 'Who worked on gravity?' ---
  [ResponseGraphEntry(kind='graph_completion', search_type='GRAPH_COMPLETION', text='Albert Einstein — he developed general relativity, which describes gravity as spacetime curvature.', score=None, dataset_id='3f474d11-b9ca-500b-a7ec-2f84dd88a49c', dataset_name='science_claims', ...

The cleanup removes exactly one dataset. forget(everything=True) reports the science_claims dataset that add() created, confirming the graph, its vectors, and the ingested document all lived there.

--- Forget everything ---
  {'datasets_removed': 1, 'status': 'success'}

Before You Start

Complete Quickstart to understand basic operations
Ensure you have LLM Providers configured — two of the three tasks are live LLM calls, and the script requires LLM_API_KEY in your .env or environment
Run it from a checkout of the cognee repo, so the uv run command below resolves the repo’s dependencies
The script calls cognee.forget(everything=True) both before and after the pipeline, so point it at a scratch instance rather than memory you want to keep — see Forget

How It Works

Stage 1: Model the Graph as Typed DataPoints

class ScientificClaim(DataPoint):
    """A factual claim extracted from text."""

    text: Annotated[str, Embeddable("Claim text for semantic search"), Dedup()]
    subject: str = ""
    confidence: float = 1.0


class Person(DataPoint):
    """A person mentioned in the text."""

    name: Annotated[str, Embeddable("Person name"), Dedup()]
    role: str = ""
    claims: list[ScientificClaim] | None = None

These two classes are the whole schema of the resulting graph. Embeddable marks the field that gets embedded and indexed for semantic search, Dedup marks the field that derives the node’s identity — so re-running the pipeline over text that mentions the same person again lands on the same node instead of a second one. The claims field is what makes this a graph rather than a table: a list of ScientificClaim objects on a Person becomes edges from that person to those claims.

Stage 2: Keep the LLM Schema Separate from the Graph Schema

class ExtractedPerson(BaseModel):
    name: str
    role: str = ""


class ExtractedClaim(BaseModel):
    text: str
    subject: str = ""
    confidence: float = 1.0


class ExtractionResult(BaseModel):
    people: list[ExtractedPerson] = Field(default_factory=list)
    claims: list[ExtractedClaim] = Field(default_factory=list)

The LLM is asked for these plain Pydantic models, not for the DataPoint classes directly. A DataPoint carries an id, metadata, versioning and provenance fields, and a structured-output call would hand every one of them to the model to fill in — with invented ids and metadata as the result. Extracting into a plain schema and building the DataPoints from it afterwards keeps those fields under cognee’s control.

Stage 3: Extract People and Claims from the Ingested Document

    text_parts = []
    for data_item in data_items:
        async with open_data_file(data_item.raw_data_location, mode="r", encoding="utf-8") as file:
            text_parts.append(file.read())

    extraction = await LLMGateway.acreate_structured_output(
        text_input="\n".join(text_parts),
        system_prompt=(
            "Extract all people and scientific claims from the text. "
            "For each person, provide their name and role. "
            "For each claim, provide the claim text, subject, and confidence (0-1)."
        ),
        response_model=ExtractionResult,
    )

    people = [Person(name=p.name, role=p.role) for p in extraction.people]
    claims = [
        ScientificClaim(text=c.text, subject=c.subject, confidence=c.confidence)
        for c in extraction.claims
    ]

    # Returned as one list of DataPoints so the pipeline stamps provenance —
    # including the source document's content hash — on every node before the
    # next task wires them together.
    return [*people, *claims]

This is the body of the extract_entities task. Because the pipeline runs against a dataset, the task receives the dataset’s Data records rather than a raw string, and reads the text back from where add() stored it. The LLM’s ExtractionResult is then turned into real Person and ScientificClaim objects, whose ids now come from their Dedup fields. Returning them as one list matters: the pipeline stamps provenance onto every DataPoint a task returns, and the content hash of the Data record the task started from is part of that stamp.

Stage 4: Attribute Each Claim to Its Author

    people = [node for node in nodes if isinstance(node, Person)]
    claims = [node for node in nodes if isinstance(node, ScientificClaim)]

    assignments = await LLMGateway.acreate_structured_output(
        text_input=(f"People: {[p.name for p in people]}\nClaims: {[c.text for c in claims]}"),
        system_prompt=(
            "Assign each claim to the person who made it or is most associated with it. "
            "Return a list of assignments, each with a person_name and their claim_texts."
        ),
        response_model=Assignments,
    )

    # Build lookup and attach claims to people
    claim_lookup = {c.text: c for c in claims}
    for assignment in assignments.assignments:
        for person in people:
            if person.name.lower() == assignment.person_name.lower():
                person.claims = [
                    claim_lookup[t] for t in assignment.claim_texts if t in claim_lookup
                ]

    return people

The body of the link_claims_to_people task sorts the incoming nodes back into people and claims, then splits attribution into its own LLM call over just the names and claim texts rather than the source document. Assignments is another plain response model. The returned assignments are resolved back to the actual objects through claim_lookup and hung off each matching Person, so the task returns people whose claims lists are populated — the edges the graph will get.

Stage 5: Store the Objects and Read Their Provenance

    # add_data_points persists nodes and edges to graph DB,
    # and indexes embeddable fields in vector DB
    await add_data_points(people)

    lines = []
    for person in people:
        # source_content_hash is stamped by the pipeline provenance system;
        # it carries the content hash of the source document this node came from
        hash_display = person.source_content_hash or "N/A"
        lines.append(f"{person.name} ({person.role}) [source_hash: {hash_display[:12]}]")

One call to add_data_points writes the people, the claims nested inside them, and the edges between them — plus the embeddings for every Embeddable field. Because the task runs inside a dataset pipeline, the nodes land in that dataset’s graph and vector stores and are attributed to it, which is what lets recall find them later. The summary then reads source_content_hash back off each stored person: the same twelve characters on every line, because every node traces back to the one document that was ingested.

Stage 6: Ingest, Then Run the Pipeline Against the Dataset

    # Ingest the text into a dataset first. This creates the dataset, stores the
    # text as a Data record with a content hash, and is what makes the graph the
    # custom pipeline builds below both attributable and searchable.
    await cognee.add(sample_text, dataset_name=DATASET_NAME)

    # Run the custom pipeline over the dataset's ingested documents. With no
    # `data` argument the first task receives the dataset's Data records.
    await cognee.run_custom_pipeline(
        tasks=[
            Task(extract_entities),
            Task(link_claims_to_people),
            Task(store_and_summarize),
        ],
        dataset=DATASET_NAME,
        pipeline_name="entity_extraction",
    )

add() does the ingestion work cognify would otherwise build on: it creates the science_claims dataset and stores the text as a Data record with a content hash. run_custom_pipeline then runs the three Task objects in order inside that dataset’s context — each task’s return value becomes the next task’s first argument, and every document in the dataset goes through the chain. Running the pipeline against a dataset is also what gives the tasks a pipeline context with a dataset in it, so the stored nodes belong somewhere. The pipeline_name labels the run for logging and status tracking.

Stage 7: Recall from the Hand-Built Graph

    # Recall from the graph
    print("\n--- Recall: 'Who worked on gravity?' ---")
    answer = await cognee.recall(
        "Who worked on gravity?",
        query_type=cognee.SearchType.GRAPH_COMPLETION,
        datasets=[DATASET_NAME],
    )
    print(f"  {answer}")

Nothing about the query knows the graph was assembled by hand. GRAPH_COMPLETION traverses the Person and ScientificClaim nodes in the science_claims dataset exactly as it would traverse nodes produced by cognify(), which is the point of writing typed DataPoint models: custom extraction buys you a custom schema without giving up the ordinary retrieval path.

Run It

uv run python examples/demos/custom_pipelines/custom_pipeline_single_object_example.py

The run takes about a minute, most of it in the two extraction calls. Its output is walked through in What to Expect above.

Adapting It to Your Data

The three-task shape generalizes to any typed extraction: define the DataPoint classes your domain actually has, annotate the field that should be searchable with Embeddable and the field that identifies a record with Dedup, then nest one model inside another wherever you want an edge. Give each DataPoint a matching plain Pydantic extraction model and build the DataPoints from what the LLM returns, as Stage 3 does. What changes per domain is the two system prompts and the response models; the pipeline around them stays as it is. To run it over a real corpus, add() the documents into the dataset instead of one string — every Data record in the dataset goes through the three tasks, and each stored node carries the hash of the document it came from.

Custom Tasks and Pipelines

Writing your own tasks and running them as a pipeline, step by step.

DataPoints

How Embeddable, Dedup, and nested models shape the graph a DataPoint becomes.

Custom Data Models

More on add_data_points and modeling your own node types.

Low-Level LLM

Calling acreate_structured_output directly, including Pydantic response models.

Getting Started

Core Concepts

Setup Configuration

Guides

Examples

CLI

Rust SDK

TypeScript SDK

OSS

What You’ll Build

Features in Play

What to Expect

Before You Start

How It Works

Stage 1: Model the Graph as Typed DataPoints

Stage 2: Keep the LLM Schema Separate from the Graph Schema

Stage 3: Extract People and Claims from the Ingested Document

Stage 4: Attribute Each Claim to Its Author

Stage 5: Store the Objects and Read Their Provenance

Stage 6: Ingest, Then Run the Pipeline Against the Dataset

Stage 7: Recall from the Hand-Built Graph

Run It

Adapting It to Your Data

Custom Tasks and Pipelines

DataPoints

Custom Data Models

Low-Level LLM

​What You’ll Build

​Features in Play

​What to Expect

​Before You Start

​How It Works

​Stage 1: Model the Graph as Typed DataPoints

​Stage 2: Keep the LLM Schema Separate from the Graph Schema

​Stage 3: Extract People and Claims from the Ingested Document

​Stage 4: Attribute Each Claim to Its Author

​Stage 5: Store the Objects and Read Their Provenance

​Stage 6: Ingest, Then Run the Pipeline Against the Dataset

​Stage 7: Recall from the Hand-Built Graph

​Run It

​Adapting It to Your Data

Custom Tasks and Pipelines

DataPoints

Custom Data Models

Low-Level LLM

What You’ll Build

Features in Play

What to Expect

Before You Start

How It Works

Stage 1: Model the Graph as Typed DataPoints

Stage 2: Keep the LLM Schema Separate from the Graph Schema

Stage 3: Extract People and Claims from the Ingested Document

Stage 4: Attribute Each Claim to Its Author

Stage 5: Store the Objects and Read Their Provenance

Stage 6: Ingest, Then Run the Pipeline Against the Dataset

Stage 7: Recall from the Hand-Built Graph

Run It

Adapting It to Your Data