Pluggable Observability for Semantic Workflows: cognee × Keywords AI

We love building graph-native, data-aware AI systems. But when you stitch together multiple async tasks, agents, and data pipelines, your system can become a black box, with behavior that’s hard to trace, measure, and debug.

You need span‑level visibility across workflows—without littering your code with tracing boilerplate or getting locked into a tool.

That’s why we’ve designed cognee / Memory for AI agents to run on modular, scalable pipelines. Our workflows load data, create semantic graph nodes, and trigger downstream tasks. We wanted clean, consistent tracing across all of it—and Keywords AI, via its lightweight keywordsai-tracing SDK, gives us exactly that, with a single decorator: @observe.

Keywords AI / LLM observability for AI teams provides real‑time traces, logs, and metrics across LLM calls and agent runs: task/workflow spans, token and latency stats, and error visibility. Wrapping our async pipelines with @observe gives every span a meaningful name and stable ID—so when something veers off course, we can debug fast, and with zero changes to our business logic.

In this article, we’ll cover how cognee standardizes observability, how the Keywords AI adapter works, and give you a practical example you can run by yourself locally.

Who This Is For:

• ML/agent teams who need to see every tool call and model hop.
• Platform/SRE folks who want consistent tracing across services.
• Developers who want observability without vendor lock‑in or heavy integration work.

Outcome: Name every span. Follow every hop. Fix issues faster.

One Decorator, Many Backends: cognee’s Pluggable Observability

cognee is an open source memory engine with a semantic graph at its core. You use tasks and workflows via its Python SDK to manipulate DataPoint models based on your data and to connect them into a knowledge graph. For observability, cognee exposes a single surface: @observe.

• Unified abstraction: Decorate tasks with @observe; decorate workflows with @observe(workflow=True).
• Pluggable backends: Choose your monitoring tool via config or environment variables.
• Zero vendor lock-in: The interface remains the same regardless of the telemetry provider.

Integrating Keywords AI: Spanning with a Single Import

cognee-community is our extension hub. Adapters for third‑party tools and databases, pipelines, and community‑contributed tasks live here. Hosting integrations in this repo gives you:

• Community license and independent evolution: Adapters can iterate at their own pace, independently of core releases.
• Slim, targeted installs: Pull in only what you need, avoiding heavyweight drivers for users who don’t need them.
• Seamless interoperability with core: A small registration shim wires the package into cognee’s abstraction layer.
• Predictable layout: Under *packages/**, each subfolder is a self‑contained provider implemented in the same pattern you’ll follow.

Keywords AI integration lives in the community repo as cognee-community-observability-keywordsai.

Note on the approach: The adapter intentionally monkey‑patches get_observe() at import time. This keeps the @observe surface stable while letting you swap backends via config.

Quick Start (3 Minutes)

Prerequisites:

• Python 3.10+
• Keywords AI API key
• LLM API key (e.g., OpenAI)
• A clean virtual environment

1) Clone the community repo:

 git clone https://github.com/topoteretes/cognee-community.git
 cd cognee-community

2. Get your Keywords AI API key here and set the following environment variables:

# Required for Keywords AI setup
export MONITORING_TOOL=keywordsai
export KEYWORDSAI_API_KEY=<your_KeywordsAI_key>

# Required for cognee
export LLM_API_KEY=<your_OpenAI_key>

3. Navigate to the Keywords AI integration directory and install dependencies:

cd packages/observability/keywordsai

uv sync --all-extras

Prefer pip? Use your env’s installer to install this package and its extras.

4. The adapter patches cognee’s get_observe() at import time, so you get Keywords AI spans with no code changes to your tasks:

# packages/observability/keywordsai/cognee_community_observability_keywordsai/register.py

from importlib import import_module
from typing import Callable, Any

from .keywordsai_adapter import get_keywordsai_observe
from cognee.base_config import get_base_config

get_observe_mod = import_module("cognee.modules.observability.get_observe")
_orig_get_observe: Callable[..., Any] = get_observe_mod.get_observe  # save

def _patched_get_observe():
    tool = get_base_config().monitoring_tool
    value = str(getattr(tool, "value", tool)).lower().replace("_", "")

    if value.startswith("observer."):
        value = value.split(".", 1)[1]

    if value == "keywordsai":
        return get_keywordsai_observe()
    return _orig_get_observe()

get_observe_mod.get_observe = _patched_get_observe  # type: ignore[attr-defined]

Under the hood, get_keywordsai_observe() maps the cognee @observe surface to Keywords AI’s @task and @workflow decorators:

# packages/observability/keywordsai/cognee_community_observability_keywordsai/keywordsai_adapter.py
"""
Expose `get_keywordsai_observe()` – a decorator factory that mimics Cognee's
`@observe`, but sends spans to Keywords AI using `keywordsai-tracing`.
"""

from typing import Any, Callable

from keywordsai_tracing.decorators import task, workflow
from keywordsai_tracing.main import KeywordsAITelemetry

KeywordsAITelemetry()

def get_keywordsai_observe() -> Callable[..., Any]:
    """
    Return a decorator that supports the same surface as Cognee's @observe :

      @observe                       -> Keywords AI task span
      @observe(workflow=True)        -> Keywords AI workflow span
      @observe(name="X", ...)        -> span named "X"
    """

    def observe(func: Any | None = None, **kw):  # noqa: ANN401
        def _wrap(f):
            name = kw.get("name") or kw.get("as_type") or f.__name__
            if kw.get("workflow"):
                return workflow(name=name)(f)
            return task(name=name)(f)

        return _wrap if func is None else _wrap(func)

    return observe

• Automatic telemetry bootstrap: KeywordsAITelemetry() initializes the SDK once.
• Named spans: Use @observe(name="my_span") to override defaults.
  • Tip: Use @observe(name="ingest_files") to enforce consistent span naming.

Example: Trace Every Step in a cognee Pipeline

To demonstrate how Keywords AI helps trace every step in cognee, we’ve created a simple, custom pipeline that shows how to build DataPoint nodes from JSON input files (like people.json or companies.json), ingest them, and render a graph—all while emitting task and workflow spans to Keywords AI.

# packages/observability/keywordsai/examples/observability_keywordsai_demo/example.py
import cognee_community_observability_keywordsai  # noqa: F401  (patches Cognee)

import os, json, asyncio
from typing import List, Any

from cognee import prune, visualize_graph
from cognee.low_level import setup, DataPoint
from cognee.modules.data.methods import load_or_create_datasets
from cognee.modules.users.methods import get_default_user
from cognee.pipelines import run_tasks, Task
from cognee.tasks.storage import add_data_points as _add_data_points

from cognee.modules.observability.get_observe import get_observe
observe = get_observe()

class Person(DataPoint):
    name: str
    metadata: dict[str, Any] = {"index_fields": ["name"]}

class Department(DataPoint):
    name: str
    employees: list[Person]
    metadata: dict[str, Any] = {"index_fields": ["name"]}

class CompanyType(DataPoint):
    name: str = "Company"
    metadata: dict[str, Any] = {"index_fields": ["name"]}

class Company(DataPoint):
    name: str
    departments: list[Department]
    is_type: CompanyType
    metadata: dict[str, Any] = {"index_fields": ["name"]}

# ---------- Task: build DataPoints ---------- #
@observe                    # task span
def ingest_files(data: List[Any]):
    people_dp, dept_dp, company_dp = {}, {}, {}
    for item in data:
        people, companies = item["people"], item["companies"]
        for p in people:
            person = people_dp[p["name"]] = Person(name=p["name"])
            dept_dp.setdefault(
                p["department"], Department(name=p["department"], employees=[])
            ).employees.append(person)

        ctype = CompanyType()
        for c in companies:
            comp = company_dp[c["name"]] = Company(
                name=c["name"], departments=[], is_type=ctype
            )
            for d in c["departments"]:
                comp.departments.append(
                    dept_dp.setdefault(d, Department(name=d, employees=[]))
                )
    return company_dp.values()

# Wrap to trace
add_data_points = observe(_add_data_points)

# ---------- Workflow ---------- #
@observe(workflow=True)     # workflow span
async def main():
    await prune.prune_data(); await prune.prune_system(metadata=True)
    await setup()

    user = await get_default_user()
    ds = await load_or_create_datasets(["test_dataset"], [], user)

    base = os.path.dirname(__file__)
    companies = json.load(open(os.path.join(base, "companies.json")))
    people    = json.load(open(os.path.join(base, "people.json")))
    data = [{"companies": companies, "people": people}]

    pipeline = run_tasks(
        [Task(ingest_files), Task(add_data_points)],
        dataset_id=ds[0].id,
        data=data,
    )
    async for s in pipeline:
        print(s)

    html_out = os.path.join(base, ".artifacts/graph_visualization.html")
    await visualize_graph(html_out)
    return html_out

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

Key points:

• Decorate tasks: @observe creates a Keywords AI task span for ingest_files.
• Decorate workflows: @observe(workflow=True) wraps the whole run as a workflow span.
• Wrap existing functions: observe(_add_data_points) instruments an existing cognee task without changing its code.

What You Get in Keywords AI

• Workflow and task spans with consistent naming.
• Timing, errors, and context tied back to your semantic pipeline.
• No custom plumbing—instrumentation is a single decorator.

Step Into the Future: Measurable, Graph-Native AI Systems

Observability should help you move faster, not slow you down. With cognee’s unified @observe surface and Keywords AI’s lightweight tracing, you get robust telemetry for complex, graph-driven pipelines—without holding development back.

Have any questions or want to join the conversation? Drop into our Discord communities:

• cognee
• Keywords AI