Building AI Agent Memory for Prefect Workflows with Cognee and dlt

Building an Intelligent Workflow Brain: Syncing Prefect Orchestration into Cognee Memory with DLT

Modern data engineering and orchestration platforms manage hundreds of pipelines, scheduled deployments, and thousands of execution runs. When a midnight ETL job crashes, an engineer is often left piecing together what happened across disconnected logs, API records, and dashboards.

What if your AI agents had instant, contextual memory of your entire orchestration topology?

In this guide, we dive into the newly created Prefect data-source connector for Cognee (cognee-community-connector-prefect), built as part of the Mergetober 2026 hackathon. We'll explore how it seamlessly bridges Prefect 2/3 orchestration metadata into Cognee's knowledge graph and vector stores using dlt, while upholding production-grade security, incremental state transitions, and forget-on-delete capabilities.

LLM-driven assistants and autonomous DevOps agents are only as capable as the memory backing them. While general documents and database tables are easy to ingest, workflow orchestration data introduces three unique engineering hurdles:

State Dynamics: Unlike static files, a flow run is mutable. It starts as PENDING, transitions to RUNNING, and terminates as COMPLETED, FAILED, or CRASHED. A naive ingestion pipeline that only queries creation timestamps misses terminal state transitions. Secret Parameter Leakage: Prefect deployments and flow runs routinely accept runtime parameters—such as database passwords, connection strings, and webhook tokens. Indexing raw parameters into vector databases or knowledge graphs poses an unacceptable security risk. Graph Deletion & Drift: When obsolete flows or ephemeral testing deployments are deleted in Prefect, they must be cleanly removed from Cognee's graph and vector memory rather than lingering as orphaned hallucinations.

The cognee-community-connector-prefect package lives in the topoteretes/cognee-community monorepo under packages/connector/prefect.

Key Engineering Decisions: Strict Zero-Secret Parameter Policy: The connector parses parameters on deployments and runs, strictly rendering parameter keys only (e.g. Parameters (keys only): batch_size, target_table), ensuring zero passwords or API keys are ever committed to memory. Merge Disposition with Tombstones: Configured with write_disposition="merge" and columns={"_deleted": {"data_type": "bool", "hard_delete": True}}. When an entity is deleted upstream, a tombstone row is emitted, enabling Cognee's orphan_cleanup to purge it from relational, graph, and vector stores without wiping the dataset.

Deterministic SHA-256 Fingerprints: Each record is fingerprinted across its core operational metadata. When a flow run updates from RUNNING to COMPLETED, its fingerprint shifts, prompting Cognee to re-index the updated state while untouched historical runs are skipped (stats["skipped"]).

bash export PREFECT_API_URL=" http://localhost:4200/api "

bash export PREFECT_API_URL=" https://api.prefect.cloud/api/accounts/ /workspaces/" export PREFECT_API_KEY="pnu_..." Step 3: Ingest Orchestration Metadata with cognee.remember python import asyncio import os import cognee from cognee_community_connector_prefect import prefect_source async def main(): base_url = os.environ.get("PREFECT_API_URL", " http://localhost:4200/api" ) api_key = os.environ.get("PREFECT_API_KEY") # Ingest flows, deployments, and execution runs into memory source = prefect_source( base_url=base_url, api_key=api_key, resource_name="prefect_main", include=("flows", "deployments", "flow_runs"), # Optional: filter by tags or execution states flow_run_states=["COMPLETED", "FAILED", "CRASHED"], ) print("--- Starting Prefect Ingestion ---") await cognee.remember( source, dataset_name="orchestration_memory", primary_key="id", write_disposition="merge", max_rows_per_table=0, ) print("Sync Stats:", source.cognee_sync_stats) # Query your workflow memory print("\n--- Asking Cognee Memory ---") result = await cognee.search( query_text="What flows and deployments failed recently, and what are their schedules?", query_type=cognee.SearchType.GRAPH_COMPLETION, datasets=["orchestration_memory"], ) print("Orchestration Insights:\n", result) if name == " main ": asyncio.run(main())

To ensure rock-solid reliability without external network dependencies, the connector includes a 42-test offline suite:

Client Tests: Exponential backoff on HTTP 502/503, connection retry handling, rate-limit header parsing, and token redaction. State Transition Tests: Verifies that a flow run moving from PENDING to FAILED is recognized and re-emitted on the subsequent sync. Secret Redaction Tests: Verifies that test fixtures with sensitive passwords (db_password=supersecretpassword) never leak into document text. Forget-on-Delete Tests: Confirms single-endpoint lookups detect deleted entities and issue tombstones. Real SQLite DLT Pipeline: Verifies state survival and hard-deletion across consecutive pipeline runs. bash pytest packages/connector/prefect/tests -v ====================== 42 passed in 22.84s =======================

Connecting workflow orchestration engines like Prefect into Cognee bridges the gap between infrastructure telemetry and cognitive AI agents. Operators can now query complex pipeline states naturally, conduct automated failure root-cause analysis, and build resilient autonomous workflows.

Pull Request: topoteretes/cognee-community#446 Feature Issue: topoteretes/cognee#6119 Repository: Cognee Community Monorepo