> ## Documentation Index > Fetch the complete documentation index at: https://docs.cognee.ai/llms.txt > Use this file to discover all available pages before exploring further. # Typed Claim Extraction > Build a three-step custom pipeline that turns a paragraph of research text into typed Person and ScientificClaim nodes, each claim attributed to the person who made it 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`](https://github.com/topoteretes/cognee/blob/main/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](/guides/custom-tasks-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](/core-concepts/main-operations/legacy-operations/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](/core-concepts/building-blocks/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](/guides/custom-data-models) — `add_data_points` persists the typed objects and their nesting as nodes and edges, without a cognify pass * [Low-Level LLM](/guides/low-level-llm) — `LLMGateway.acreate_structured_output` runs both extraction passes against plain Pydantic response models * [Recall](/core-concepts/main-operations/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. ```text theme={null} 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. ```text theme={null} --- 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. ```text theme={null} --- Forget everything --- {'datasets_removed': 1, 'status': 'success'} ``` ## Before You Start * Complete [Quickstart](/getting-started/quickstart) to understand basic operations * Ensure you have [LLM Providers](/setup-configuration/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](/core-concepts/main-operations/forget) ## How It Works ### Stage 1: Model the Graph as Typed DataPoints ```python theme={null} 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 ```python theme={null} 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 ```python theme={null} 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 ```python theme={null} 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 ```python theme={null} # 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 ```python theme={null} # 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](/core-concepts/building-blocks/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 ```python theme={null} # 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 ```bash theme={null} 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](#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. Writing your own tasks and running them as a pipeline, step by step. How `Embeddable`, `Dedup`, and nested models shape the graph a DataPoint becomes. More on `add_data_points` and modeling your own node types. Calling `acreate_structured_output` directly, including Pydantic response models.