Cognee

1
import asyncio
2
import cognee_community_observability_respan  # noqa: F401 — patches Cognee on import
3
from cognee.modules.observability.get_observe import get_observe
4
5
observe = get_observe()  # returns Respan-backed decorator
6
7
@observe
8
def process_documents(documents: list[str]):
9
    return [f"Processed: {doc}" for doc in documents]
10
11
@observe
12
def create_knowledge_graph(processed_docs: list[str]):
13
    return [{"id": i, "content": doc} for i, doc in enumerate(processed_docs)]
14
15
@observe(workflow=True)
16
async def semantic_workflow():
17
    documents = ["Document 1", "Document 2", "Document 3"]
18
    processed = process_documents(documents)
19
    return create_knowledge_graph(processed)
20
21
result = asyncio.run(semantic_workflow())
22
print(f"Created knowledge graph with {len(result)} nodes")