​

cognee.cognify()

async def cognify(
    datasets: Union[str, list[str], list[UUID]] = None,
    user: User = None,
    graph_model: BaseModel = KnowledgeGraph,
    chunker = TextChunker,
    chunk_size: int = None,
    chunks_per_batch: int = None,
    config: Config = None,
    vector_db_config: dict = None,
    graph_db_config: dict = None,
    run_in_background: bool = False,
    incremental_loading: bool = True,
    custom_prompt: Optional[str] = None,
    temporal_cognify: bool = False,
    data_per_batch: int = 20,
    llm_config: Optional[LLMConfig] = None,
    embedding_config: Optional[EmbeddingConfig] = None,
    dry_run: bool = False,
)

​

Description

• LLM_API_KEY: Must be configured (required for entity extraction and graph generation)
• Data Added: Must have data previously added via cognee.add()
• Vector Database: Must be accessible for embeddings storage
• Graph Database: Must be accessible for relationship storage

• Datasets: Must contain data previously added via cognee.add()
• Content Types: Works with any text-extractable content including:
  • Natural language documents
  • Structured data (CSV, JSON)
  • Code repositories
  • Academic papers and technical documentation
  • Mixed multimedia content (with text extraction)

1. Document Classification: Identifies document types and structures
2. Text Chunking: Breaks content into semantically meaningful segments
3. Entity Extraction: Identifies key concepts, people, places, organizations
4. Relationship Detection: Discovers connections between entities
5. Graph Construction: Builds semantic knowledge graph with embeddings
6. Content Summarization: Creates hierarchical summaries for navigation

• Default: General-purpose KnowledgeGraph for any domain
• Custom Models: Domain-specific schemas (e.g., scientific papers, code analysis)
• Ontology Integration: Pass an ontology resolver via config (or set the ONTOLOGY_FILE_PATH environment variable) for predefined vocabularies

• Single dataset: “my_dataset”
• Multiple datasets: [“docs”, “research”, “reports”]
• None: Process all datasets for the user user: User context for authentication and data access. Uses default if None. graph_model: Pydantic model defining the knowledge graph structure. Defaults to KnowledgeGraph for general-purpose processing. chunker: Text chunking strategy (TextChunker, LangchainChunker).
  • TextChunker: Paragraph-based chunking (default, most reliable)
  • LangchainChunker: Recursive character splitting with overlap Determines how documents are segmented for processing. chunk_size: Maximum tokens per chunk. Auto-calculated based on LLM if None. Formula: min(embedding_max_completion_tokens, llm_max_completion_tokens // 2) Default limits: ~512-8192 tokens depending on models. Smaller chunks = more granular but potentially fragmented knowledge. chunks_per_batch: Number of chunks to be processed in a single batch in Cognify tasks. vector_db_config: Custom vector database configuration for embeddings storage. graph_db_config: Custom graph database configuration for relationship storage. run_in_background: If True, starts processing asynchronously and returns immediately. If False, waits for completion before returning. Background mode recommended for large datasets (>100MB). Use pipeline_run_id from return value to monitor progress. custom_prompt: Optional custom prompt string to use for entity extraction and graph generation. If provided, this prompt will be used instead of the default prompts for knowledge graph extraction. The prompt should guide the LLM on how to extract entities and relationships from the text content. dry_run: If True, return a stage-level estimate of LLM token usage and rough cost without making LLM calls or writing graph results. The estimate covers all data in the selected dataset(s); an incremental run may process fewer items.

• Blocking mode: Dictionary mapping dataset_id -> PipelineRunInfo with:
  • Processing status (completed/failed/in_progress)
  • Extracted entity and relationship counts
  • Processing duration and resource usage
  • Error details if any failures occurred
• Background mode: List of PipelineRunInfo objects for tracking progress
  • Use pipeline_run_id to monitor status
  • Check completion via pipeline monitoring APIs

import cognee
from cognee import SearchType

# Process your data into knowledge graph
await cognee.cognify()

# Query for insights using different search types:

# 1. Natural language completion with graph context
insights = await cognee.search(
    "What are the main themes?",
    query_type=SearchType.GRAPH_COMPLETION
)

# 2. Get entity relationships and connections
relationships = await cognee.search(
    "connections between concepts",
    query_type=SearchType.GRAPH_COMPLETION
)

# 3. Find relevant document chunks
chunks = await cognee.search(
    "specific topic",
    query_type=SearchType.CHUNKS
)

# Custom domain model for scientific papers
class ScientificPaper(DataPoint):
    title: str
    authors: List[str]
    methodology: str
    findings: List[str]

await cognee.cognify(
    datasets=["research_papers"],
    graph_model=ScientificPaper,
)

# Ground extraction in an ontology (there is no `ontology_file_path` argument;
# pass a resolver through `config` instead).
from cognee.modules.ontology.rdf_xml.RDFLibOntologyResolver import RDFLibOntologyResolver
from cognee.modules.ontology.ontology_config import Config

config: Config = {
    "ontology_config": {
        "ontology_resolver": RDFLibOntologyResolver(ontology_file="scientific_ontology.owl")
    }
}
await cognee.cognify(datasets=["research_papers"], config=config)

# Background processing for large datasets
run_info = await cognee.cognify(
    datasets=["large_corpus"],
    run_in_background=True
)
# Check status later with run_info.pipeline_run_id

• LLM_API_KEY: API key for your LLM provider

• LLM_PROVIDER, LLM_MODEL, VECTOR_DB_PROVIDER, GRAPH_DATABASE_PROVIDER
• AUTO_RATE_LIMIT: Turn the rate limiter on automatically when the provider shows overload evidence (default: True)
• LLM_RATE_LIMIT_ENABLED: Enable rate limiting from the first request (default: False)
• LLM_RATE_LIMIT_REQUESTS: Max requests per interval (default: 60; 10 for local inference servers)

• CONTRADICTION_DETECTION: Append the opt-in contradiction check to the pipeline (default: False)
• CONTRADICTION_CONFIDENCE_THRESHOLD: Minimum LLM confidence for a pair to be flagged (default: 0.5)
• CONTRADICTION_MAX_FACTS: Cap on the facts sent to the LLM in a single check (default: 500)

• PROVENANCE_TRACKING: Append the opt-in provenance-ledger task to the pipeline (default: False)

​

Parameters

​

Dry-run cost estimation

import cognee

estimate = await cognee.cognify(datasets=["my_dataset"], dry_run=True)
print(estimate)                    # human-readable summary table
print(estimate.estimated_cost_usd) # e.g. 0.012345
print(estimate.total_tokens)       # input + output tokens across stages
print(estimate.to_dict())          # JSON-serializable dict

• Datasets are resolved read-only. Unlike a normal run, a dry run never creates a missing dataset, so estimating a typo’d dataset name fails loudly instead of silently creating one.
• Estimates are upper bounds for re-runs. With incremental_loading=True, a real run skips already-processed documents, so a dry run may over-estimate a re-run.
• Not supported with temporal_cognify=True (only the default pipeline is estimated) or while connected to a remote Cognee instance via serve() — both raise a ValueError.
• Unknown models emit a warning rather than reporting a $0 cost when no pricing entry exists for the configured model.
• Code files are counted as skipped, at zero cost. Items that cognify() would route down the code graph pipeline are separated out before any document is read, so they are never chunked and contribute no tokens or cost. They are folded into skipped_items and reported in warnings as Skipped N code file(s) because they run the deterministic code graph pipeline — no LLM calls. See Loaders for which extensions take that route.

​

Processing Pipeline

1. Document classification — identify content type
2. Text chunking — split into manageable segments
3. Entity extraction — identify entities using the LLM
4. Relationship detection — find connections between entities
5. Graph construction — build the knowledge graph
6. Summarization — generate summaries of content
7. Provenance recording (opt-in, off by default) — append an audit-ledger entry for every document, chunk, entity, and relationship this run produced. Enabled with PROVENANCE_TRACKING=true; when off, the task list is identical to the pipeline above. See Provenance ledger.
8. Contradiction detection (opt-in, off by default) — compare the facts this run touched against the facts already stored around them and record each conflict as a contradicts edge. Enabled with CONTRADICTION_DETECTION=true; when off, the task list is identical to the pipeline above. See Contradiction detection.

​

Provenance ledger

PROVENANCE_TRACKING=true   # default: false

• Failure is never fatal. The task returns its input unchanged and swallows all of its own errors, logging a warning (Provenance recording failed; ingestion unaffected: ...) instead of failing the run. Missing pipeline context or ids degrade to entries with a source_ref_key of None rather than raising. A swallowed failure rolls the whole batch back, so those entries are simply absent — they never claimed sequence numbers, the hash chain stays intact, and verify_chain() still reports valid. Verification proves the stored entries were not tampered with, not that everything an ingestion produced was recorded, so watch that warning if you depend on the ledger being complete.
• Ledger keys are dataset-scoped. Cognee entity ids are deterministic and the ledger lives in the shared relational database, so every node-derived ledger id is prefixed with the dataset id ("{dataset_id}:{raw_id}"), and relationship ids are built from the prefixed endpoints. Two datasets mentioning the same entity name keep separate version chains.
• There is one chain, and one writer at a time. Dataset scoping applies to ledger keys and version chains, not to the hash chain itself: every entry in the table shares a single sequence_id sequence, and each commit takes a ledger-wide write lock (a Postgres advisory lock, so it serializes across processes too). Concurrent cognify() runs — several datasets, several workers — therefore queue behind each other for their ledger commits, which is why entries are batched one transaction per task invocation.
• Custom graph_model schemas are covered. DataPoints that do not follow the default made_from / is_part_of / contains shape are walked with the same traversal add_data_points uses, so every node and edge is still recorded.
• The table must exist. It ships as Alembic revision b8c1d3e5f7a9; run migrations (cognee.run_migrations(), or alembic upgrade head) before enabling the flag on an existing deployment. The migration is idempotent — it is a no-op if the table is already present.

from cognee.modules.provenance import get_provenance_manager

manager = get_provenance_manager()

report = await manager.verify_chain()
print(report["valid"], report["total_entries"], report["broken_links"])

​

Contradiction detection

CONTRADICTION_DETECTION=true             # default: false
CONTRADICTION_CONFIDENCE_THRESHOLD=0.5   # minimum LLM confidence to flag a pair
CONTRADICTION_MAX_FACTS=500              # cap on facts sent to the LLM in one check

import cognee

await cognee.add("Alice was born in 1985.")
await cognee.cognify()

# Later ingestion, with CONTRADICTION_DETECTION=true
await cognee.add("Alice was born in 1990.")
await cognee.cognify()

Contradiction detected (confidence 0.95): 'alice born in 1985' contradicts 'alice born in 1990' — <reason>

​

The contradicts edge

from cognee.infrastructure.databases.graph import get_graph_engine

graph_engine = await get_graph_engine()
_, edges = await graph_engine.get_graph_data()

for source, target, relationship_name, properties in edges:
    if relationship_name == "contradicts":
        print(properties["first_fact"], "<>", properties["second_fact"])
        print(properties["reason"], properties["confidence"])

​

Scope, cost, and limits

​

Examples

import cognee

# Process all datasets
await cognee.cognify()

# Process a specific dataset
await cognee.cognify(datasets=["my_dataset"])

# Process in background
await cognee.cognify(datasets=["large_dataset"], run_in_background=True)

# Use a custom graph model
from pydantic import BaseModel

class MyGraph(BaseModel):
    nodes: list
    edges: list

await cognee.cognify(graph_model=MyGraph)

# Custom extraction prompt
await cognee.cognify(
    custom_prompt="Extract all technical concepts and their relationships."
)

# Pin specific entity and relationship types
custom_prompt = """
Extract only people and cities as entities.
Connect people to cities with the relationship "lives_in".
Ignore all other entities.
"""

await cognee.cognify(custom_prompt=custom_prompt)

​

Further details

import cognee

# Start processing without waiting for completion
run_info = await cognee.cognify(
    datasets=["large_corpus"],
    run_in_background=True
)

# run_info is a dict of {dataset_id: PipelineRunInfo}
for dataset_id, info in run_info.items():
    print(info.pipeline_run_id)  # UUID to track this run
    print(info.dataset_id)       # Dataset being processed
    print(info.status)           # Initial status (e.g. "PipelineRunStarted")

WebSocket: /cognify/subscribe/{pipeline_run_id}

const pipelineRunId = "your-pipeline-run-id-uuid";
const ws = new WebSocket(`ws://your-server/cognify/subscribe/${pipelineRunId}`);

ws.onmessage = (event) => {
    const data = JSON.parse(event.data);
    console.log("Status:", data.status);
    console.log("Run ID:", data.pipeline_run_id);
    // data.payload contains the current graph data for the dataset
};

ws.onclose = () => {
    // Server closes the connection when processing completes (status: PipelineRunCompleted)
    console.log("Pipeline run finished");
};

{
    "pipeline_run_id": "uuid-string",
    "status": "PipelineRunYield",
    "payload": { /* current graph data for the dataset */ }
}