Ingesting Mixpanel Event Dictionaries & Cohorts into Cognee Knowledge Graphs
When engineering AI agents to analyze user churn or feature adoption, the common knee-jerk reaction is to dump millions of raw event logs into a vector database.
That almost never works. Millions of rows like {"event": "checkout_clicked", "time": 1728144000} quickly blow past token budgets, bloat vector indexes, and return noisy semantic matches.
The real high-leverage intelligence in Mixpanel lives in Lexicon Schemas, User Cohort Criteria, and Saved Report Bookmarks.
Here is how we built the official Mixpanel Connector for Cognee to turn that semantic product knowledge into an autonomous graph memory.
1. What We Ingest: Context Over Noise
Instead of flooding your graph with every repetitive click timestamp, our connector focuses on the core conceptual primitives:
-
Lexicon Schemas: The data dictionary of your product (what properties exist on
workspace_created, what each property represents, and its active status). - User Cohorts: Dynamic user groupings (e.g. "Power Users active > 5 days a week" or "Dormant Enterprise Accounts").
- Saved Reports / Insights: The queries and metrics your growth and product teams care about most.
flowchart LR
MP[Mixpanel API] -->|Lexicon / Cohorts / Bookmarks| Client[MixpanelClient]
Client -->|yield Semantic Markdown| DLT[DLT Pipeline]
DLT -->|DOCUMENT_SOURCE_ATTR| Cognee[Cognee Cognify]
Cognee -->|Graph Nodes & Relations| DB[(Graph DB + Vector Storage)]2. Engineering the Connector
Resilience & Rate-Limit Handling
Mixpanel APIs enforce strict query limits. We built an exponential backoff loop with status code inspection to gracefully handle HTTP 429 and network blips:
for attempt in range(MAX_RETRIES):
try:
response = http.request(method, url, auth=self.auth, params=params)
if response.status_code == 429:
retry_after = float(response.headers.get("Retry-After", 2**attempt))
time.sleep(retry_after)
continue
response.raise_for_status()
return response.json()
except (httpx.TransportError, httpx.TimeoutException):
if attempt == MAX_RETRIES - 1:
raise
time.sleep(2**attempt)
Semantic Prose Serialization
Each event schema is parsed into a clean document capturing its data dictionary:
# Mixpanel Lexicon Event: payment_failed
- **Event Name:** payment_failed
- **Status:** active
- **Tags:** billing, alerts, core_checkout
### Description
Fired whenever a Stripe or PayPal payment intent returns an error during checkout.
### Event Properties & Data Dictionary
- `error_code` (string) - Raw gateway error code
- `retry_count` (number) - Number of automatic retry attempts
- `plan_tier` (string) - Target subscription tier (starter, pro, enterprise)
When Cognee indexes this, questions like "What triggers a payment failure event and what properties are logged?" are answered instantly with graph precision.
3. Quickstart Example
import asyncio
import cognee
from cognee_community_connector_mixpanel import mixpanel_source
async def main():
source = mixpanel_source(
project_id="YOUR_MIXPANEL_PROJECT_ID",
secret="YOUR_SERVICE_ACCOUNT_SECRET",
include_schemas=True,
include_cohorts=True,
include_reports=True,
)
await cognee.add(source, dataset_name="mixpanel_knowledge")
await cognee.cognify(dataset_name="mixpanel_knowledge")
answer = await cognee.search(
"Which user cohorts track dormant enterprise accounts?",
dataset_name="mixpanel_knowledge"
)
print(answer)
if __name__ == "__main__":
asyncio.run(main())
4. Testing & Verification
100% offline mocked pytest tests verify schema parsing, cohort mapping, date filters, and Cognee ingestion without hitting live network endpoints:
uv run pytest packages/connector/mixpanel/tests/test_mixpanel.py -v
Top comments (0)