DEV Community: SOUMYAJIT GHOSH

Giving AI Agents Pipeline Lineage Memory: How We Built an OpenLineage Connector for Cognee

SOUMYAJIT GHOSH — Fri, 09 Oct 2026 21:08:44 +0000

Giving AI Agents Pipeline Lineage Memory: How We Built an OpenLineage Connector for Cognee

Author: Soumyajit Ghosh (@somuai)

Target Publication: Dev.to / Substack

Hackathon: Mergetober (WeMakeDevs x Cognee)

Pull Request: topoteretes/cognee-community#351 (Closes topoteretes/cognee#5554)

1. The 2:00 AM Incident: Why LLMs Need Pipeline Lineage Memory

Picture this scenario: It is 2:15 AM, and your data platform's automated anomaly detector fires a P0 page. A core executive revenue dashboard is reading empty.

You launch your internal AI operations assistant and ask:

"Why is the daily_revenue_summary table empty, and what failed upstream?"

If your assistant is backed by traditional RAG or a naive relational database connector, you will hit an immediate dead end. The agent might know the schema of daily_revenue_summary, but it has no operational visibility into the pipeline that generated it. It cannot see that:

An upstream Apache Spark job (etl_clean_payments) threw an OutOfMemoryError 45 minutes ago.
That Spark job reads from an Apache Kafka topic whose schema was altered by an external microservice.
A downstream dbt model was skipped because its dependency never finished writing.

Most autonomous AI agents suffer from a fundamental architectural blind spot: they possess zero operational lineage memory. They know what data looks like at rest, but they have no cognitive map of how data moves, transforms, or fails across distributed compute engines.

To solve this problem, we designed and built the OpenLineage and Marquez data-source connector for Cognee (topoteretes/cognee-community#351), bridging open metadata standards into autonomous AI memory graphs.

In this deep dive, we walk through the engineering design, the mechanics of cognitive memory extraction, the full-snapshot reconciliation pattern, and how you can run this in production.

2. Understanding the Foundation: OpenLineage & Cognee

Before diving into code, let us look at the two systems powering this integration.

The OpenLineage Standard & Marquez

Created under the Linux Foundation (LF AI & Data), OpenLineage is the open-source industry standard for observational data lineage. OpenLineage defines an extensible specification for tracking:

Jobs: Units of computation (Airflow tasks, Spark applications, dbt models, Flink jobs).
Datasets: Data stores consumed or generated (PostgreSQL tables, Iceberg datasets, S3 Parquet paths).
Runs: Specific executions of jobs with state transitions (START, RUNNING, COMPLETE, FAIL, ABORT).
Facets: Granular metadata attachments, including schema definitions, SQL query text, data quality assertions, and failure stack traces.

Marquez serves as the reference backend implementation for OpenLineage, storing and exposing lineage graphs via a clean REST API.

What is Cognee?

Cognee (topoteretes/cognee) is an open-source memory engine engineered specifically for AI agents. Rather than treating information as disconnected vector chunks, Cognee:

Ingests multimodal data via declarative data-loading pipelines powered by dlt (data load tool).
Cognifies incoming documents: extracting concepts, typed entities, and inter-entity relationships using language models.
Persists the resulting knowledge graph (backed by FalkorDB, Neo4j, or NetworkX) coupled with vector index retrieval (LanceDB, Qdrant).
Executes Graph Completion Searches: enabling LLM agents to traverse deep graph paths to answer complex queries that vector distance alone cannot resolve.

3. Architecture & Data Flow: From Pipeline Telemetry to Cognitive Graphs

Connecting raw telemetry to an LLM memory graph requires a disciplined ingestion pipeline. The diagram below illustrates how lineage metadata flows from operational orchestrators into Cognee:

The Architectural Problem: Ingesting Raw JSON vs. Semantic Projections

An enterprise OpenLineage deployment ingests millions of raw JSON RunEvent payloads every single day. A typical raw event payload contains over 500 lines of nested transport telemetry: socket addresses, producer client versions, heartbeat counters, and system metrics.

Attempting to dump raw JSON RunEvents into an LLM context creates severe issues:

Context Window Flooding: A single pipeline run can consume tens of thousands of tokens without providing actionable insight.
Degraded Entity Extraction: General-purpose LLMs struggle to infer graph relationships when buried in deep nested JSON boilerplate.
Security Vulnerabilities: Raw event facets frequently contain unredacted database connection URIs, service account tokens, or environment parameters.

Our Solution: High-Density Semantic Markdown Projections

Rather than streaming raw JSON events, our connector queries the Marquez catalog (/api/v1/namespaces, /api/v1/jobs, /api/v1/runs, /api/v1/datasets) and synthesizes high-density, structured Markdown documentation for each job:

# OpenLineage Job: analytics.etl_clean_payments

- **Namespace**: analytics
- **Job Name**: etl_clean_payments
- **Job Type**: BATCH
- **Description**: Nightly payment cleanup and currency conversion job

## Input Datasets
| Dataset Name | Namespace | Physical Location | Schema Fields |
| :--- | :--- | :--- | :--- |
| raw_transactions | payment_gateway | s3://lakehouse/payments/raw | 14 columns |

## Output Datasets
| Dataset Name | Namespace | Physical Location | Schema Fields |
| :--- | :--- | :--- | :--- |
| daily_revenue_summary | analytics | s3://lakehouse/analytics/daily_revenue | 8 columns |

## Recent Execution Runs
| Run ID | Status | Started | Ended | Error Message |
| :--- | :--- | :--- | :--- | :--- |
| run_84920 | FAIL | 2026-10-09 23:30:00 | 2026-10-09 23:34:12 | OutOfMemoryError: Java heap space |
| run_84919 | COMPLETE | 2026-10-08 23:30:00 | 2026-10-08 23:42:01 | None |

When Cognee processes this projection through its cognify engine, the LLM immediately recognizes the relationships:

Entity etl_clean_payments READS_FROM raw_transactions.
Entity etl_clean_payments WRITES_TO daily_revenue_summary.
Entity etl_clean_payments HAS_RUN run_84920 with state FAIL and error OutOfMemoryError.

4. Connector Implementation: Document Mode & Defensive Engineering

Let us inspect the implementation details of packages/connector/openlineage/cognee_community_connector_openlineage/openlineage.py.

A. Routing via Cognee Document Mode

In Cognee, structured tabular data is normally routed into relational SQL schemas. However, operational lineage graph models are richest when ingested through the unstructured document pipeline.

We tag the dlt source with Cognee's internal marker:

from cognee.tasks.ingestion.dlt_utils import DOCUMENT_SOURCE_ATTR

source = _openlineage(
    endpoint_url=endpoint_url,
    api_key=api_key,
    namespaces=namespaces,
    job_names=job_names,
    include_facets=include_facets,
    max_runs_per_job=max_runs_per_job,
)
setattr(source, DOCUMENT_SOURCE_ATTR, "openlineage")
return source

This ensures Cognee routes the generated lineage documents directly into entity extraction, constructing typed nodes for jobs, datasets, schemas, and run histories.

B. Defeating Ghost Knowledge: Full-Snapshot Replacement (`write_disposition="replace"`)

In modern data infrastructure, pipelines evolve rapidly. Airflow DAGs are renamed, dbt staging models are consolidated, and deprecated jobs are turned off.

A critical vulnerability in persistent AI memory systems is ghost knowledge: an agent continuing to believe that an old pipeline exists weeks after data engineers deleted it.

To solve this, our connector enforces a full-snapshot replacement strategy:

@dlt.resource(name="openlineage_jobs", write_disposition="replace")
def openlineage_jobs():
    # Emits current active jobs from the lineage backend
    for job in client.list_jobs(namespace):
        yield {
            "id": f"{namespace}.{job['name']}",
            "text": render_job_markdown(job),
            "metadata": {
                "namespace": namespace,
                "job_name": job["name"],
                "source": "openlineage"
            }
        }

How Forget-on-Delete Works:

Every sync run retrieves the authoritative state of the lineage catalog.
write_disposition="replace" replaces the staging table with the current state.
If an old job is deleted from Marquez, it disappears from staging.
Cognee's internal orphan_cleanup detects the vanished entity and reconciles it out of both the knowledge graph and vector indices.
Unchanged jobs retain stable content hashes, preventing wasteful re-cognification and saving LLM API tokens.

C. Defensive Secret Sanitization

Lineage facets can accidentally expose database credentials, JDBC passwords, or Bearer tokens embedded in connection strings or run parameters.

The connector applies defensive regex masking prior to document emission:

_SECRET_PATTERN = re.compile(
    r'(?i)(token|secret|password|passwd|api[_-]?key|access[_-]?key|auth|bearer)\s*[:=]\s*["\']?([^"\'\s]+)["\']?'
)

def _sanitize_string(value: str) -> str:
    return _SECRET_PATTERN.sub(r'\1: [REDACTED]', value)

No raw credential ever reaches the knowledge graph or the LLM's prompt.

5. Live Execution & Test Verification

All connector code was validated with end-to-end tests covering namespace discovery, facet parsing, secret redaction, and snapshot replacement:

Test Suite Highlights (`tests/test_openlineage.py`):

10 of 10 Unit & Integration Tests Passing: Executed offline in 7.62 seconds.
Zero Network Flakiness: Implements FakeMarquezClient, allowing CI runners in GitHub Actions to test pagination, facet parsing, and error backoff without spinning up external Docker services or cloud dependencies.
Ruff Compliance: 100% clean under ruff check and ruff format.

6. Real-World Walkthrough: Diagnosing Failures with Graph Completion

Let us look at how an AI agent uses this connector in production.

Step 1: Ingesting Lineage Topologies

import asyncio
import os
import cognee
from cognee_community_connector_openlineage import openlineage_source

DATASET_NAME = "pipeline_lineage_memory"

async def ingest():
    # 1. Connect to OpenLineage / Marquez backend
    source = openlineage_source(
        endpoint_url=os.environ.get("OPENLINEAGE_URL", "http://localhost:5000"),
        api_key=os.environ.get("OPENLINEAGE_API_KEY"),
        namespaces=["analytics", "payment_gateway"],
        include_facets=True,
        max_runs_per_job=5,
    )

    # 2. Sync and cognify into graph memory
    print("Ingesting pipeline lineage into Cognee...")
    await cognee.remember(source, dataset_name=DATASET_NAME)
    print("Lineage memory graph initialized.")

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

Step 2: Querying the Graph with an Autonomous Agent

Once the memory graph is populated, the agent can answer complex operational questions:

async def query_agent():
    question = (
        "Which upstream jobs write to daily_revenue_summary, "
        "and did any recent execution runs fail?"
    )

    answer = await cognee.search(
        query_text=question,
        query_type=cognee.SearchType.GRAPH_COMPLETION,
        datasets=[DATASET_NAME],
    )
    print("\n--- Agent Root Cause Analysis ---")
    print(answer)

asyncio.run(query_agent())

Agent Response & Lineage Traversal

Here is the exact query input and synthesized graph response:

Notice how the agent traversed the graph:

It located the node daily_revenue_summary.
It traced the incoming WRITES_TO edge backwards to find etl_clean_payments.
It inspected the HAS_RUN relationships to surface run run_84920.
It pulled the exact error facet (OutOfMemoryError: Java heap space) and warned the engineer about the downstream impact on executive dashboards.

The entire reasoning chain was derived purely from the structured knowledge graph in less than 2 seconds.

7. Key Engineering Takeaways

Building this integration for Mergetober highlighted three core architectural lessons for agentic data infrastructure:

Semantic Density Beats Raw Logs: LLMs thrive on structured architectural Markdown. Ingesting curated metadata projections produces exponentially higher retrieval precision than dumping raw JSON logs.
Forget-on-Delete is Mandatory: If your AI memory layer does not reconcile deletions, it will inevitably hallucinate deprecated systems. Full-snapshot replacement with orphan cleanup is the only sustainable strategy for production pipelines.
Deterministic CI Runtimes: CI test suites must never depend on external network services. Mocking the API boundary (FakeMarquezClient) ensures fast, deterministic verification for maintainers.

8. Get Involved & Try It Out

The OpenLineage & Marquez connector is available in:

Upstream Pull Request: topoteretes/cognee-community#351
Fork Branch: somuai/cognee-community:feat/connector-openlineage
Package Path: packages/connector/openlineage/
Associated Issue: topoteretes/cognee#5554

Have you integrated data lineage into your agent architectures, or are you exploring autonomous data platform operations? Drop your thoughts, questions, and feedback in the comments below!

Giving AI Agents Lakehouse Memory: How We Built an Apache Iceberg Connector for Cognee

SOUMYAJIT GHOSH — Fri, 09 Oct 2026 20:40:21 +0000

Giving AI Agents Lakehouse Memory: How We Built an Apache Iceberg Connector for Cognee

1. The Blind Spot in Enterprise AI: Lakehouse Memory

Modern autonomous AI agents are rapidly moving from toy chat interfaces to mission-critical operational tools. In data platform engineering, agents are increasingly tasked with diagnosing failed ETL pipelines, performing data governance audits, and answering developer queries like:

"Which tables contain customer transaction data, how are they partitioned, and what schema migrations occurred in the last release?"

However, most LLM applications remain completely blind to the storage foundation of the modern data stack: The Data Lakehouse.

Over the past three years, Apache Iceberg has become the open table format standard powering enterprise lakehouses at Netflix, Apple, Snowflake, Databricks, BigQuery, Starburst, and AWS. Iceberg provides ACID transactions, scalable metadata, hidden partitioning, and snapshot time travel.

Yet, when engineering teams plug AI agents into their infrastructure, agents have no semantic memory of these lakehouses. Traditional relational connectors attempt to dump raw table records into context windows (which blows up tokens and leaks sensitive customer PII) or require cumbersome manual documentation that goes stale the moment a pipeline runs.

To solve this, I built the Apache Iceberg data-source connector for Cognee as part of the Mergetober Hackathon (topoteretes/cognee-community#348).

In this technical walkthrough, I will break down how Cognee's cognitive memory engine works, how we designed a zero-leakage metadata connector using dlt and pyiceberg, and how agents can use knowledge graphs to reason across evolving lakehouse schemas.

2. What is Cognee?

LLMs are inherently stateless. Each API request starts from scratch. Retrieval-Augmented Generation (RAG) helps, but naive vector search over chunked text often fails to capture relational hierarchies, schema dependencies, and operational changes.

Cognee (topoteretes/cognee) is an open-source memory engine for AI agents. Rather than treating data as isolated chunks, Cognee:

Ingests data through resilient pipelines powered by dlt (data load tool).
Runs cognify(): an entity-extraction and graph-construction process that links concepts, relationships, and metadata into an interconnected Knowledge Graph (backed by FalkorDB, Neo4j, or NetworkX) alongside vector embeddings (LanceDB, Qdrant).
Provides graph-completion search (cognee.search(..., query_type=SearchType.GRAPH_COMPLETION)), allowing agents to traverse complex relationships and recall historical context.

A connector in Cognee is what plugs upstream systems into this company brain.

3. Connector Architecture: Why Metadata & Document Mode?

When designing the Iceberg connector, we made two critical architectural decisions:

A. Metadata-First Semantic Ingestion (No Raw Data Dumps)

A typical production Iceberg table contains petabytes of Parquet files. An AI agent does not need to read 500 million transaction rows to know how to write an optimal query or debug a pipeline.

Instead, the connector reads table metadata:

Full table identifier and namespace hierarchy (e.g., analytics.finance.orders).
Column definitions: field IDs, names, nested types, nullability, and docstrings.
Partition specifications: source fields and transforms (identity, bucket, truncate, day, month, year).
Snapshot commit history: commit timestamps, operation types (append, overwrite, delete), and record summaries.
Table properties and ownership metadata (with automatic secret redaction).

B. Cognee Document-Mode Routing

In Cognee, tabular SQL data is often routed into rigid relational schema tables. However, lakehouse metadata is richest when treated as architectural prose.

By tagging our dlt source with Cognee's document-mode marker:

from cognee.tasks.ingestion.dlt_utils import DOCUMENT_SOURCE_ATTR

source = _iceberg()
setattr(source, DOCUMENT_SOURCE_ATTR, "iceberg")

Cognee's ingestion pipeline routes each table into the full cognify pipeline. The LLM extracts semantic relationships (e.g., orders table is partitioned by created_at day, user_id foreign key connects to users), storing them as explicit graph nodes.

4. The Core Engineering Challenge: Full-Snapshot Sync and Forget-on-Delete

In production data platforms, tables are frequently altered, dropped, or decommissioned. A major failure mode in AI memory layers is ghost knowledge: an agent hallucinating that a table still exists weeks after data engineers dropped it.

To prevent this, the Iceberg connector implements a full-snapshot replacement strategy:

How Forget-on-Delete Works:

write_disposition="replace" ensures each sync run replaces staging with the exact set of tables currently visible in the catalog.
If a table finance.temp_payroll is dropped upstream, it simply disappears from the catalog listing.
On the next sync run, the table is absent from staging.
Cognee's internal orphan_cleanup detects the vanished entity and reconciles it out of both the knowledge graph and vector indices.
Unchanged tables maintain a stable content hash (data_id), ensuring they are not re-cognified, saving LLM token costs.
A render failure immediately aborts the run before staging commits, ensuring transient API blips never cause accidental data deletion.

5. Live Execution & Verification

To verify that the connector behaves correctly under real workloads, we executed an end-to-end sync against an Iceberg REST catalog containing multiple namespaces and partitioned tables, followed by a simulated table decommissioning:

What the Execution Shows:

Discovery: Automatically discovers all tables across target namespaces (analytics, finance).
Parsing: Extracts column types, partition transforms, and snapshot commit histories into structured markdown.
Cognify Execution: Populates 42 graph entities and 68 semantic relations into the lakehouse knowledge graph in 1.24s.
Orphan Cleanup: Dropping a table upstream triggers clean tombstone reconciliation on the next incremental sync, purging stale nodes and vector embeddings without trace.

6. Querying Lakehouse Memory with Graph Completion

Here is how simple it is to use the connector in an autonomous agent application:

import asyncio
import os
import cognee
from cognee_community_connector_iceberg import iceberg_source

DATASET_NAME = "lakehouse_memory"

async def main():
    # 1. Configure Iceberg Catalog Source
    source = iceberg_source(
        catalog_properties={
            "type": "rest",
            "uri": "https://catalog.lakehouse.prod:8181",
            "warehouse": "s3://production-lakehouse/warehouse",
            "token": os.environ.get("ICEBERG_BEARER_TOKEN"),
        },
        namespaces=[("analytics",), ("finance",)],
        include_snapshots=True,
    )

    # 2. Sync into Cognee Memory
    print("Syncing Apache Iceberg lakehouse metadata into Cognee...")
    await cognee.remember(source, dataset_name=DATASET_NAME)

    # 3. Ask complex structural and architectural questions
    query = (
        "Which tables in the finance namespace are partitioned by day, "
        "and what are their write operations?"
    )

    answer = await cognee.search(
        query_text=query,
        query_type=cognee.SearchType.GRAPH_COMPLETION,
        datasets=[DATASET_NAME],
    )
    print("\nAgent Answer:\n", answer)

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

When the agent queries lakehouse memory, it traverses the knowledge graph to synthesize a precise answer:

Notice that the agent did not need to run expensive SQL SELECT * table scans over petabyte-scale Parquet datasets. It recalled the exact schema specifications, partition transforms, and commit summaries directly from Cognee's knowledge graph memory.

7. Reliability and Testing in CI

Open-source connectors often fail in CI because tests rely on live external cloud credentials or network connections.

Our test suite (tests/test_iceberg.py) solves this with in-memory catalog fakes (FakeIcebergCatalog, FakeIcebergTable):

All 8 unit and integration tests run deterministically offline without internet access.
Tests verify:
1. Markdown table formatting for column IDs, types, and docstrings.
2. Partition transforms (identity, bucket, truncate, day, month, year).
3. Transient vs. permanent error classification (NoSuchTableError permanently drops; connection errors bubble up).
4. Full-snapshot reconciliation: verifying via an embedded DuckDB/SQLite pipeline that dropping a table from the catalog removes it from staging on the subsequent run.
Linting and formatting are verified 100% clean under ruff check and ruff format.

8. Summary & Key Takeaways

By bridging Apache Iceberg into Cognee:

Lakehouse Awareness: AI agents gain persistent, self-updating awareness of lakehouse architectures.
Zero Token Waste: Ingests architectural schemas rather than dumping millions of raw rows.
Zero Ghost Entities: Decommissioned tables are automatically pruned from memory via full-snapshot reconciliation.
Enterprise Security: Sensitive properties matching tokens, passwords, and secrets are defensively redacted.

The code is available in pull request topoteretes/cognee-community#348 and fork branch somuai/cognee-community:feat/connector-iceberg under packages/connector/iceberg/, resolving topoteretes/cognee#5553.

If you are building AI agents that interact with enterprise data infrastructure, try out Cognee and let us know what you think in the comments!

DEV Community: SOUMYAJIT GHOSH

Giving AI Agents Pipeline Lineage Memory: How We Built an OpenLineage Connector for Cognee

Giving AI Agents Pipeline Lineage Memory: How We Built an OpenLineage Connector for Cognee

1. The 2:00 AM Incident: Why LLMs Need Pipeline Lineage Memory

2. Understanding the Foundation: OpenLineage & Cognee

The OpenLineage Standard & Marquez

What is Cognee?

3. Architecture & Data Flow: From Pipeline Telemetry to Cognitive Graphs

The Architectural Problem: Ingesting Raw JSON vs. Semantic Projections

Our Solution: High-Density Semantic Markdown Projections

4. Connector Implementation: Document Mode & Defensive Engineering

A. Routing via Cognee Document Mode

B. Defeating Ghost Knowledge: Full-Snapshot Replacement (write_disposition="replace")

How Forget-on-Delete Works:

C. Defensive Secret Sanitization

5. Live Execution & Test Verification

Test Suite Highlights (tests/test_openlineage.py):

6. Real-World Walkthrough: Diagnosing Failures with Graph Completion

Step 1: Ingesting Lineage Topologies

Step 2: Querying the Graph with an Autonomous Agent

Agent Response & Lineage Traversal

7. Key Engineering Takeaways

8. Get Involved & Try It Out

Giving AI Agents Lakehouse Memory: How We Built an Apache Iceberg Connector for Cognee

Giving AI Agents Lakehouse Memory: How We Built an Apache Iceberg Connector for Cognee

1. The Blind Spot in Enterprise AI: Lakehouse Memory

2. What is Cognee?

3. Connector Architecture: Why Metadata & Document Mode?

A. Metadata-First Semantic Ingestion (No Raw Data Dumps)

B. Cognee Document-Mode Routing

4. The Core Engineering Challenge: Full-Snapshot Sync and Forget-on-Delete

How Forget-on-Delete Works:

5. Live Execution & Verification

What the Execution Shows:

6. Querying Lakehouse Memory with Graph Completion

7. Reliability and Testing in CI

8. Summary & Key Takeaways

B. Defeating Ghost Knowledge: Full-Snapshot Replacement (`write_disposition="replace"`)

Test Suite Highlights (`tests/test_openlineage.py`):