From db6c6e599deea206ac3bfa91016c57eda34ba388 Mon Sep 17 00:00:00 2001 From: David Justice Date: Wed, 3 Jun 2026 17:16:50 -0400 Subject: [PATCH] feat: CrewAI incident response example with KD6 memory backend Add examples/crewai-memory/ demonstrating CrewAI multi-agent crews using KD6 as a persistent, layered memory backend via the StorageBackend protocol. Components: - Kd6StorageBackend: full StorageBackend protocol adapter translating CrewAI's path-based scopes to KD6's structured MemoryScope hierarchy, with upsert-based idempotent saves and metadata envelope preservation - Incident response simulation: 4-agent crew (Incident Commander, Platform Engineer, Backend Developer, Communications Lead) investigating a production checkout service failure - Smoke test: non-LLM connectivity test validating all CRUD operations - GitHub Models integration: both LLM inference and embeddings via GITHUB_TOKEN with openai/gpt-4o-mini and text-embedding-3-small Also adds Python-specific entries to .gitignore (__pycache__, .venv, .env). Signed-off-by: David Justice --- .gitignore | 7 + examples/crewai-memory/.env.example | 16 + examples/crewai-memory/README.md | 257 +++++++++++ examples/crewai-memory/pyproject.toml | 13 + examples/crewai-memory/src/__init__.py | 0 examples/crewai-memory/src/config.py | 79 ++++ examples/crewai-memory/src/kd6_backend.py | 506 ++++++++++++++++++++++ examples/crewai-memory/src/main.py | 293 +++++++++++++ examples/crewai-memory/src/smoke_test.py | 178 ++++++++ 9 files changed, 1349 insertions(+) create mode 100644 examples/crewai-memory/.env.example create mode 100644 examples/crewai-memory/README.md create mode 100644 examples/crewai-memory/pyproject.toml create mode 100644 examples/crewai-memory/src/__init__.py create mode 100644 examples/crewai-memory/src/config.py create mode 100644 examples/crewai-memory/src/kd6_backend.py create mode 100644 examples/crewai-memory/src/main.py create mode 100644 examples/crewai-memory/src/smoke_test.py diff --git a/.gitignore b/.gitignore index 6867282..1c0c2ae 100644 --- a/.gitignore +++ b/.gitignore @@ -25,6 +25,13 @@ target node_modules/ dist/ +# Python (examples) +__pycache__/ +*.pyc +.venv/ +*.egg-info/ +.env + # SQLite runtime files *.db *.db-shm diff --git a/examples/crewai-memory/.env.example b/examples/crewai-memory/.env.example new file mode 100644 index 0000000..e752bd7 --- /dev/null +++ b/examples/crewai-memory/.env.example @@ -0,0 +1,16 @@ +# CrewAI + KD6 example +# Copy this file to .env and fill in your values. + +# GitHub Personal Access Token with models:read scope +# See README.md for instructions on creating a token. +# Leave blank to auto-detect from `gh auth token`. +GITHUB_TOKEN= + +# KD6 server URL (default: local server) +KD6_URL=http://localhost:8080 + +# KD6 store name (auto-created on first run) +KD6_STORE_NAME=crewai-incident-response + +# KD6 tenant ID +KD6_TENANT_ID=crewai-demo diff --git a/examples/crewai-memory/README.md b/examples/crewai-memory/README.md new file mode 100644 index 0000000..b14862d --- /dev/null +++ b/examples/crewai-memory/README.md @@ -0,0 +1,257 @@ +# CrewAI + KD6: Incident Response Simulation + +A multi-agent incident response crew powered by [CrewAI](https://docs.crewai.com) +and [KD6](https://github.com/devigned/kd6), demonstrating how AI agents can +collaborate with persistent, layered, scoped memory. + +## What Happens + +Four AI agents work together to investigate and resolve a simulated production +incident at "ShopStream", a fictional e-commerce platform experiencing checkout +failures: + +| Agent | Role | Memory Usage | +|-------|------|-------------| +| **Incident Commander** | Coordinates the response, maintains timeline | Stores triage decisions, reads all agent findings | +| **Platform Engineer** | Investigates infrastructure metrics | Stores infrastructure findings (DB connections, Redis, memory) | +| **Backend Developer** | Analyzes code behavior, identifies root cause | Reads infra findings, stores root cause analysis | +| **Communications Lead** | Writes status updates and postmortem | Reads all findings, produces final documentation | + +### The Incident + +The checkout API is returning 503 errors for ~30% of requests. Key signals: +- PostgreSQL connections at 450/500 (near exhaustion) +- Redis cache hit rate dropped from 95% to 62% +- Memory usage elevated from 55% to 78% +- No recent deployments + +### How KD6 Memory Is Used + +**Memory layers** organize information by purpose: +- `semantic` — extracted facts, metrics, hypotheses +- `episodic` — timeline events, investigation steps +- `working` — active scratchpad for in-progress reasoning + +**Scoped visibility** controls who sees what: +- Each agent stores memories under `/incident/shopstream/{agent-scope}` +- The Incident Commander and Communications Lead can read across all scopes +- Private findings stay scoped to the investigating agent until shared + +**Knowledge graph** (optional, via the backend's `create_edge()` helper): +- Links related findings: "Redis degradation" → "DB connection spike" +- Enables traversal queries: "What evidence supports this root cause?" + +**Cross-session persistence** — run the simulation twice: +- First run: agents investigate from scratch +- Second run: agents recall findings from the first incident + +## Prerequisites + +1. **Rust toolchain** — to build and run KD6 +2. **Python 3.11+** — for CrewAI +3. **GitHub Personal Access Token** — see [Creating a token](#creating-a-github-token) below + +### Creating a GitHub token + +The example uses [GitHub Models](https://github.com/marketplace/models) for both +LLM inference and embeddings. You need a **fine-grained** Personal Access Token +with the `Models: Read` account permission. + +1. Go to [github.com/settings/tokens?type=beta](https://github.com/settings/tokens?type=beta) + (Fine-grained tokens) +2. Click **Generate new token** +3. Give it a name (e.g. "KD6 CrewAI example") and set an expiration +4. Under **Account permissions**, find **Models** and set it to **Read** +5. Click **Generate token** and copy the value + +Then copy the `.env` template and paste your token (or leave `GITHUB_TOKEN=` +blank to auto-detect from `gh auth token`): + +```bash +cp .env.example .env +# Edit .env to set GITHUB_TOKEN if not using the gh CLI +``` + +## Setup + +### 1. Start the KD6 server + +From the repository root: + +```bash +KD6_DATABASE_URL="sqlite:kd6-crewai.db?mode=rwc" \ +KD6_EMBEDDING_PROVIDER=openai-compatible \ +KD6_EMBEDDING_ENDPOINT=https://models.github.ai/inference \ +KD6_EMBEDDING_MODEL=openai/text-embedding-3-small \ +KD6_EMBEDDING_API_KEY=$GITHUB_TOKEN \ +cargo run -p kd6-server +``` + +The server starts on `http://localhost:8080`. + +### 2. Install Python dependencies + +```bash +cd examples/crewai-memory +python3.12 -m venv .venv +source .venv/bin/activate # or .venv\Scripts\activate on Windows +pip install -e . +``` + +### 3. Run the smoke test + +Validates KD6 connectivity without using any LLM calls: + +```bash +python -m src.smoke_test +``` + +Expected output: +``` +KD6 Storage Backend Smoke Test + Server: http://localhost:8080 + Store: crewai-smoke-test + + ✓ Connected to KD6 and ensured store exists + ✓ Saved 3 records + ✓ Listed 3 records + ✓ Retrieved record: Redis cache hit rate dropped... + ✓ Updated record + ✓ Count: 3 records total + ✓ Scopes: ['/incident', '/incident/test'] + ✓ Categories: {'infrastructure': 2, 'redis': 1, ...} + ✓ Deleted 2 records by ID + ✓ Reset complete (0 records remaining) + +All smoke tests passed! ✓ +``` + +### 4. Run the incident response + +```bash +python -m src.main +``` + +The crew runs through the incident sequentially. You'll see output like: + +``` +====================================================================== + ShopStream Incident Response Simulation + Powered by CrewAI + KD6 Memory +====================================================================== + +Starting incident response... + + [17:30:01][INFO]: Working Agent: Incident Commander + [17:30:01][INFO]: Starting Task: Review the incident alert... + +> Entering new CrewAgentExecutor chain... +Thought: I need to analyze the incident metrics and establish a timeline... + +Final Answer: +## Triage Report +1. Timeline: 14:23 UTC — automated alert on 5xx rate > 5% +2. Initial hypothesis: database connection pool exhaustion +... + + [17:30:15][INFO]: Working Agent: Platform Engineer + [17:30:15][INFO]: Starting Task: Investigate the infrastructure... +... + + [17:30:32][INFO]: Working Agent: Backend Developer + [17:30:32][INFO]: Starting Task: Analyze the application behavior... +... + + [17:30:48][INFO]: Working Agent: Communications Lead + [17:30:48][INFO]: Starting Task: Synthesize all findings... +... + +====================================================================== + INCIDENT RESPONSE COMPLETE +====================================================================== + +## Post-Incident Review: ShopStream Checkout Service +### Executive Summary +A connection pool leak in the checkout service caused cascading failures... +... + +Memories persisted in KD6. Run again to see cross-session recall. +``` + +Each agent shows its chain-of-thought reasoning before producing a final answer. +The actual content varies per run since it's LLM-generated. A full run typically +takes 1–2 minutes depending on model response times. + +## Architecture + +``` +┌──────────────────────────────────────────────┐ +│ CrewAI │ +│ ┌────────────┐ ┌─────────────────────────┐ │ +│ │ LLM (GPT) │ │ Memory System │ │ +│ │ via GitHub │ │ ┌───────────────────┐ │ │ +│ │ Models │ │ │ Kd6StorageBackend │──┼─┼──► KD6 REST API +│ └────────────┘ │ └───────────────────┘ │ │ (localhost:8080) +│ │ ┌───────────────────┐ │ │ +│ │ │ OpenAI Embedder │──┼─┼──► GitHub Models +│ │ │ (text-embed-3-sm) │ │ │ Embeddings API +│ │ └───────────────────┘ │ │ +│ └─────────────────────────┘ │ +└──────────────────────────────────────────────┘ + │ + ▼ +┌──────────────────────────────────────────────┐ +│ KD6 Server │ +│ ┌────────────┐ ┌───────────────────────┐ │ +│ │ REST API │ │ SQLite + FTS5 │ │ +│ │ (Axum) │ │ Vector Search │ │ +│ └────────────┘ │ Knowledge Graph │ │ +│ │ Audit Log │ │ +│ ┌────────────┐ └───────────────────────┘ │ +│ │ Embeddings │──► GitHub Models │ +│ │ (OpenAI- │ Embeddings API │ +│ │ compatible)│ │ +│ └────────────┘ │ +└──────────────────────────────────────────────┘ +``` + +## Key Files + +| File | Description | +|------|-------------| +| `src/kd6_backend.py` | `Kd6StorageBackend` — CrewAI StorageBackend adapter for KD6 | +| `src/config.py` | Environment configuration (GitHub Models, KD6 URL) | +| `src/main.py` | Incident response crew definition and runner | +| `src/smoke_test.py` | Non-LLM connectivity test for the KD6 backend | + +## Environment Variables + +| Variable | Default | Description | +|----------|---------|-------------| +| `GITHUB_TOKEN` | *(required)* | GitHub PAT with `models:read` scope | +| `CHAT_MODEL` | `openai/gpt-4o-mini` | Chat model for GitHub Models | +| `EMBEDDING_MODEL` | `openai/text-embedding-3-small` | Embedding model | +| `KD6_URL` | `http://localhost:8080` | KD6 server URL | +| `KD6_STORE_NAME` | `crewai-incident-response` | KD6 store name | +| `KD6_TENANT_ID` | `crewai-demo` | KD6 tenant for isolation | + +## How the Backend Adapter Works + +The `Kd6StorageBackend` translates between CrewAI's memory model and KD6's +OMS-compliant API: + +| CrewAI Concept | KD6 Equivalent | +|----------------|----------------| +| `MemoryRecord.scope` (path: `/org/team`) | `MemoryScope` (structured: `org_id`, `team_id`) | +| `MemoryRecord.categories` | `categories` | +| `MemoryRecord.importance` (0.0–1.0) | `confidence` (temporal metadata) | +| `MemoryRecord.private` | `AccessControl.policy` (`private` vs `public_read`) | +| `MemoryRecord.source` | `SourceReference.uri` | +| `MemoryRecord.id` | `upsert_key` (enables idempotent saves) | +| `MemoryRecord.metadata` | Stored in `content.crewai_metadata` envelope | + +The adapter also exposes KD6-specific features not in the StorageBackend +protocol: + +- `create_edge()` — create typed graph edges between memories +- `traverse_graph()` — BFS traversal of the knowledge graph diff --git a/examples/crewai-memory/pyproject.toml b/examples/crewai-memory/pyproject.toml new file mode 100644 index 0000000..c51fa8f --- /dev/null +++ b/examples/crewai-memory/pyproject.toml @@ -0,0 +1,13 @@ +[project] +name = "crewai-kd6-example" +version = "0.1.0" +description = "CrewAI multi-agent example using KD6 as the memory backend" +requires-python = ">=3.10,<3.13" +dependencies = [ + "crewai[tools]>=1.14.0", + "httpx>=0.27.0", +] + +[project.scripts] +run-crew = "src.main:main" +smoke-test = "src.smoke_test:main" diff --git a/examples/crewai-memory/src/__init__.py b/examples/crewai-memory/src/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/examples/crewai-memory/src/config.py b/examples/crewai-memory/src/config.py new file mode 100644 index 0000000..b1d8df8 --- /dev/null +++ b/examples/crewai-memory/src/config.py @@ -0,0 +1,79 @@ +"""Configuration for the CrewAI + KD6 example. + +Loads settings from environment variables with sensible defaults. +Uses GitHub Models for both LLM inference and embeddings. +""" + +from __future__ import annotations + +import os +import sys + + +def _require_env(name: str) -> str: + val = os.environ.get(name) + if not val: + # Try gh CLI as fallback + if name == "GITHUB_TOKEN": + import subprocess + + try: + result = subprocess.run( + ["gh", "auth", "token"], + capture_output=True, + text=True, + timeout=5, + ) + if result.returncode == 0 and result.stdout.strip(): + val = result.stdout.strip() + os.environ[name] = val + return val + except (FileNotFoundError, subprocess.TimeoutExpired): + pass + + print(f"Error: {name} environment variable is required.", file=sys.stderr) + print(f"Set it in your .env file or run: gh auth login --scopes models:read", file=sys.stderr) + sys.exit(1) + return val + + +# ── GitHub Models configuration ────────────────────────────────────────── + +GITHUB_TOKEN: str = _require_env("GITHUB_TOKEN") +GITHUB_MODELS_URL: str = "https://models.github.ai/inference" +CHAT_MODEL: str = os.environ.get("CHAT_MODEL", "openai/gpt-4o-mini") +EMBEDDING_MODEL: str = os.environ.get("EMBEDDING_MODEL", "openai/text-embedding-3-small") + +# ── KD6 configuration ──────────────────────────────────────────────────── + +KD6_URL: str = os.environ.get("KD6_URL", "http://localhost:8080") +KD6_STORE_NAME: str = os.environ.get("KD6_STORE_NAME", "crewai-incident-response") +KD6_TENANT_ID: str = os.environ.get("KD6_TENANT_ID", "crewai-demo") + + +def get_crewai_llm(): + """Create a CrewAI LLM instance configured for GitHub Models.""" + from crewai import LLM + + return LLM( + model=CHAT_MODEL, + base_url=GITHUB_MODELS_URL, + api_key=GITHUB_TOKEN, + ) + + +def get_crewai_embedder_config() -> dict: + """Return the embedder config dict for CrewAI Memory. + + CrewAI's Memory class accepts an embedder config that it passes to + its internal embedding factory. This configures it to use GitHub Models' + OpenAI-compatible embedding endpoint. + """ + return { + "provider": "openai", + "config": { + "model": EMBEDDING_MODEL, + "api_key": GITHUB_TOKEN, + "api_base": GITHUB_MODELS_URL, + }, + } diff --git a/examples/crewai-memory/src/kd6_backend.py b/examples/crewai-memory/src/kd6_backend.py new file mode 100644 index 0000000..15afff0 --- /dev/null +++ b/examples/crewai-memory/src/kd6_backend.py @@ -0,0 +1,506 @@ +"""KD6 storage backend for CrewAI's unified memory system. + +Implements CrewAI's StorageBackend protocol by translating memory operations +into KD6 REST API calls. This bridges CrewAI's path-based scoping model with +KD6's structured OMS scope hierarchy. + +Scope mapping: + CrewAI scope paths are mapped to KD6's MemoryScope fields: + / → tenant-level (no additional scope fields) + /{org} → org_id + /{org}/{team} → org_id + team_id + /{org}/{team}/{project} → org_id + team_id + project_id + /{org}/{team}/{project}/{user} → ... + user_id + etc. following: org > team > project > user > agent > session > run +""" + +from __future__ import annotations + +import logging +from datetime import datetime, timezone +from typing import Any +from uuid import UUID + +import httpx + +from crewai.memory.storage.backend import StorageBackend +from crewai.memory.types import MemoryRecord, ScopeInfo + +logger = logging.getLogger(__name__) + +# Maps CrewAI scope path segments to KD6 MemoryScope field names, in order. +_SCOPE_FIELDS = ("org_id", "team_id", "project_id", "user_id", "agent_id", "session_id", "run_id") + + +def _scope_path_to_kd6(path: str) -> dict[str, str]: + """Convert a CrewAI scope path like '/crew/research' to KD6 MemoryScope fields.""" + parts = [p for p in path.strip("/").split("/") if p] + scope: dict[str, str] = {} + for i, part in enumerate(parts): + if i < len(_SCOPE_FIELDS): + scope[_SCOPE_FIELDS[i]] = part + return scope + + +def _kd6_scope_to_path(scope: dict[str, Any]) -> str: + """Convert KD6 MemoryScope fields back to a CrewAI scope path.""" + parts: list[str] = [] + for field in _SCOPE_FIELDS: + val = scope.get(field) + if val: + parts.append(val) + else: + break # Stop at first gap to maintain hierarchy + return "/" + "/".join(parts) if parts else "/" + + +def _record_to_kd6_request( + record: MemoryRecord, + *, + layer: str = "semantic", + owner_agent_id: str = "crewai", +) -> dict[str, Any]: + """Convert a CrewAI MemoryRecord to a KD6 CreateMemoryRequest body.""" + scope = _scope_path_to_kd6(record.scope) + + # If the scope has an agent_id, use it as owner; otherwise use the default + effective_owner = scope.pop("agent_id", None) or owner_agent_id + if "agent_id" not in scope and effective_owner != owner_agent_id: + scope["agent_id"] = effective_owner + + body: dict[str, Any] = { + "layer": layer, + "content": { + "text": record.content, + "crewai_metadata": record.metadata, + }, + "owner_agent_id": effective_owner, + "scope": scope, + "categories": record.categories, + "access_control": { + "policy": "private" if record.private else "public_read", + }, + "confidence": record.importance, + "upsert_key": record.id, + } + + if record.embedding is not None: + body["embedding"] = record.embedding + + if record.source: + body["source"] = {"uri": record.source} + + return body + + +def _kd6_entry_to_record(entry: dict[str, Any]) -> MemoryRecord: + """Convert a KD6 MemoryEntry JSON to a CrewAI MemoryRecord.""" + content_val = entry.get("content", "") + crewai_metadata: dict[str, Any] = {} + + if isinstance(content_val, dict): + text = content_val.get("text", "") + crewai_metadata = content_val.get("crewai_metadata", {}) + elif isinstance(content_val, str): + text = content_val + else: + text = str(content_val) + + scope_dict = entry.get("scope", {}) + scope_path = _kd6_scope_to_path(scope_dict) + + access = entry.get("access_control", {}) + is_private = access.get("policy", "private") == "private" + + source_ref = entry.get("source") + source = source_ref.get("uri") if source_ref else None + + # Use upsert_key as the CrewAI record ID, falling back to KD6 UUID + record_id = entry.get("upsert_key") or str(entry.get("id", "")) + + # Store KD6 UUID in metadata for future operations + crewai_metadata["_kd6_id"] = str(entry.get("id", "")) + crewai_metadata["_kd6_version"] = entry.get("version", 1) + + created_at_str = entry.get("created_at", "") + try: + created_at = datetime.fromisoformat(created_at_str.replace("Z", "+00:00")) + except (ValueError, AttributeError): + created_at = datetime.now(timezone.utc) + + return MemoryRecord( + id=record_id, + content=text, + scope=scope_path, + categories=entry.get("categories", []), + metadata=crewai_metadata, + importance=entry.get("confidence") or 0.5, + created_at=created_at, + last_accessed=datetime.now(timezone.utc), + embedding=entry.get("embedding"), + source=source, + private=is_private, + ) + + +class Kd6StorageBackend: + """CrewAI StorageBackend backed by a KD6 memory service. + + This adapter translates CrewAI's memory operations into KD6 REST API calls, + enabling multi-agent crews to use KD6's layered, scoped, and graph-capable + memory system. + + Args: + kd6_url: Base URL of the KD6 server (e.g. "http://localhost:8080"). + store_name: Name of the KD6 store to use (auto-created if missing). + tenant_id: Tenant identifier for isolation. + default_layer: Default memory layer for new records. + default_owner: Default owner_agent_id when scope has no agent segment. + """ + + def __init__( + self, + kd6_url: str = "http://localhost:8080", + store_name: str = "crewai-memory", + tenant_id: str = "crewai", + default_layer: str = "semantic", + default_owner: str = "crewai", + ) -> None: + self.kd6_url = kd6_url.rstrip("/") + self.store_name = store_name + self.tenant_id = tenant_id + self.default_layer = default_layer + self.default_owner = default_owner + self._client = httpx.Client( + base_url=self.kd6_url, + headers={"X-Tenant-ID": self.tenant_id}, + timeout=30.0, + ) + self._ensure_store() + + def _ensure_store(self) -> None: + """Create the store if it doesn't exist.""" + r = self._client.get(f"/v1/stores/{self.store_name}") + if r.status_code == 200: + logger.info("Using existing KD6 store: %s", self.store_name) + return + if r.status_code == 404: + r2 = self._client.post("/v1/stores", json={"name": self.store_name}) + if r2.status_code == 201: + logger.info("Created KD6 store: %s", self.store_name) + return + # Handle race condition: another process created it + if r2.status_code == 409: + logger.info("Store already exists (race): %s", self.store_name) + return + r2.raise_for_status() + r.raise_for_status() + + @property + def _base(self) -> str: + return f"/v1/stores/{self.store_name}" + + # ── StorageBackend protocol methods ────────────────────────────────── + + def save(self, records: list[MemoryRecord]) -> None: + """Save records to KD6 via batch create with upsert.""" + if not records: + return + + entries = [ + _record_to_kd6_request(r, layer=self.default_layer, owner_agent_id=self.default_owner) + for r in records + ] + + r = self._client.post(f"{self._base}/memories/batch", json={"entries": entries}) + r.raise_for_status() + + resp = r.json() + errors = resp.get("errors", []) + if errors: + logger.warning("Batch save had %d errors: %s", len(errors), errors) + + def search( + self, + query_embedding: list[float], + scope_prefix: str | None = None, + categories: list[str] | None = None, + metadata_filter: dict[str, Any] | None = None, + limit: int = 10, + min_score: float = 0.0, + ) -> list[tuple[MemoryRecord, float]]: + """Search KD6 for memories by vector similarity.""" + body: dict[str, Any] = { + "query": "", + "embedding": query_embedding, + "top_k": limit, + "threshold": min_score, + } + + if scope_prefix: + body["scope"] = _scope_path_to_kd6(scope_prefix) + + filters: dict[str, Any] = {} + if categories: + filters["categories"] = categories + if metadata_filter and "owner_agent_id" in metadata_filter: + filters["owner_agent_id"] = metadata_filter["owner_agent_id"] + if filters: + body["filters"] = filters + + r = self._client.post(f"{self._base}/search", json=body) + r.raise_for_status() + + results: list[tuple[MemoryRecord, float]] = [] + for item in r.json(): + record = _kd6_entry_to_record(item["entry"]) + score = item.get("score", 0.0) + results.append((record, score)) + + return results + + def delete( + self, + scope_prefix: str | None = None, + categories: list[str] | None = None, + record_ids: list[str] | None = None, + older_than: datetime | None = None, + metadata_filter: dict[str, Any] | None = None, + ) -> int: + """Delete memories matching the given criteria.""" + if record_ids: + # Resolve KD6 UUIDs from record IDs + kd6_ids = self._resolve_kd6_ids(record_ids) + if not kd6_ids: + return 0 + r = self._client.post( + f"{self._base}/memories/batch/delete", + json={"memory_ids": kd6_ids}, + ) + r.raise_for_status() + return r.json().get("deleted", 0) + + # For scope/category/age-based deletes, list then batch-delete + records = self.list_records(scope_prefix=scope_prefix, limit=10000) + + to_delete: list[str] = [] + for rec in records: + if categories and not any(c in rec.categories for c in categories): + continue + if older_than and rec.created_at >= older_than: + continue + kd6_id = rec.metadata.get("_kd6_id") + if kd6_id: + to_delete.append(kd6_id) + + if not to_delete: + return 0 + + r = self._client.post( + f"{self._base}/memories/batch/delete", + json={"memory_ids": to_delete}, + ) + r.raise_for_status() + return r.json().get("deleted", 0) + + def update(self, record: MemoryRecord) -> None: + """Update an existing memory record in KD6.""" + kd6_id = record.metadata.get("_kd6_id") + if not kd6_id: + # Fall back to save (upsert) + self.save([record]) + return + + body: dict[str, Any] = { + "content": { + "text": record.content, + "crewai_metadata": { + k: v for k, v in record.metadata.items() if not k.startswith("_kd6_") + }, + }, + "categories": record.categories, + "access_control": { + "policy": "private" if record.private else "public_read", + }, + } + + r = self._client.patch(f"{self._base}/memories/{kd6_id}", json=body) + r.raise_for_status() + + def get_record(self, record_id: str) -> MemoryRecord | None: + """Retrieve a single memory record by ID.""" + # Try as KD6 UUID first + kd6_id = record_id + r = self._client.get(f"{self._base}/memories/{kd6_id}") + if r.status_code == 200: + return _kd6_entry_to_record(r.json()) + if r.status_code == 404: + # Search by upsert_key: list all and filter + records = self.list_records(limit=10000) + for rec in records: + if rec.id == record_id: + return rec + return None + r.raise_for_status() + return None + + def list_records( + self, + scope_prefix: str | None = None, + limit: int = 200, + offset: int = 0, + ) -> list[MemoryRecord]: + """List memory records, optionally filtered by scope prefix.""" + params: dict[str, Any] = {"limit": limit, "offset": offset} + if scope_prefix: + scope = _scope_path_to_kd6(scope_prefix) + for key, val in scope.items(): + params[f"scope.{key}"] = val + + r = self._client.get(f"{self._base}/memories", params=params) + r.raise_for_status() + + data = r.json() + items = data.get("items", data) if isinstance(data, dict) else data + return [_kd6_entry_to_record(entry) for entry in items] + + def get_scope_info(self, scope: str) -> ScopeInfo: + """Get statistics about a specific scope.""" + records = self.list_records(scope_prefix=scope, limit=10000) + cats: dict[str, int] = {} + for rec in records: + for cat in rec.categories: + cats[cat] = cats.get(cat, 0) + 1 + + return ScopeInfo( + scope=scope, + record_count=len(records), + categories=cats, + child_scopes=[], # Would need additional queries + ) + + def list_scopes(self, parent: str = "/") -> list[str]: + """List child scopes under a parent scope path.""" + records = self.list_records(scope_prefix=parent, limit=10000) + scopes: set[str] = set() + parent_depth = len([p for p in parent.strip("/").split("/") if p]) + for rec in records: + parts = [p for p in rec.scope.strip("/").split("/") if p] + if len(parts) > parent_depth: + child = "/" + "/".join(parts[: parent_depth + 1]) + scopes.add(child) + return sorted(scopes) + + def list_categories(self, scope_prefix: str | None = None) -> dict[str, int]: + """List categories with counts, optionally filtered by scope.""" + records = self.list_records(scope_prefix=scope_prefix, limit=10000) + cats: dict[str, int] = {} + for rec in records: + for cat in rec.categories: + cats[cat] = cats.get(cat, 0) + 1 + return cats + + def count(self, scope_prefix: str | None = None) -> int: + """Count records, optionally filtered by scope.""" + records = self.list_records(scope_prefix=scope_prefix, limit=10000) + return len(records) + + def reset(self, scope_prefix: str | None = None) -> None: + """Delete all records in the given scope.""" + deleted = self.delete(scope_prefix=scope_prefix) + logger.info("Reset: deleted %d records (scope=%s)", deleted, scope_prefix) + + # ── Async variants ─────────────────────────────────────────────────── + + async def asave(self, records: list[MemoryRecord]) -> None: + self.save(records) + + async def asearch( + self, + query_embedding: list[float], + scope_prefix: str | None = None, + categories: list[str] | None = None, + metadata_filter: dict[str, Any] | None = None, + limit: int = 10, + min_score: float = 0.0, + ) -> list[tuple[MemoryRecord, float]]: + return self.search( + query_embedding, scope_prefix, categories, metadata_filter, limit, min_score + ) + + async def adelete( + self, + scope_prefix: str | None = None, + categories: list[str] | None = None, + record_ids: list[str] | None = None, + older_than: datetime | None = None, + metadata_filter: dict[str, Any] | None = None, + ) -> int: + return self.delete(scope_prefix, categories, record_ids, older_than, metadata_filter) + + # ── Graph helpers (not part of StorageBackend, but showcase KD6) ───── + + def create_edge( + self, + source_id: str, + target_id: str, + relation_type: str, + *, + weight: float = 1.0, + metadata: dict[str, Any] | None = None, + ) -> dict[str, Any]: + """Create a typed edge between two memory entries in KD6's knowledge graph.""" + body: dict[str, Any] = { + "source_memory_id": source_id, + "target_memory_id": target_id, + "relation_type": relation_type, + "weight": weight, + } + if metadata: + body["metadata"] = metadata + + r = self._client.post(f"{self._base}/graph/edges", json=body) + r.raise_for_status() + return r.json() + + def traverse_graph( + self, + start_id: str, + relation_types: list[str] | None = None, + max_depth: int = 2, + ) -> list[dict[str, Any]]: + """Traverse the knowledge graph from a starting memory entry.""" + body: dict[str, Any] = { + "start_memory_id": start_id, + "max_depth": max_depth, + } + if relation_types: + body["relation_types"] = relation_types + + r = self._client.post(f"{self._base}/graph/traverse", json=body) + r.raise_for_status() + return r.json() + + # ── Internal helpers ───────────────────────────────────────────────── + + def _resolve_kd6_ids(self, record_ids: list[str]) -> list[str]: + """Resolve CrewAI record IDs to KD6 UUIDs.""" + kd6_ids: list[str] = [] + for rid in record_ids: + try: + UUID(rid) + kd6_ids.append(rid) + except ValueError: + # Not a UUID — search for it via listing + records = self.list_records(limit=10000) + for rec in records: + if rec.id == rid: + kd6_id = rec.metadata.get("_kd6_id") + if kd6_id: + kd6_ids.append(kd6_id) + break + return kd6_ids + + def close(self) -> None: + """Close the HTTP client.""" + self._client.close() diff --git a/examples/crewai-memory/src/main.py b/examples/crewai-memory/src/main.py new file mode 100644 index 0000000..8ad35e5 --- /dev/null +++ b/examples/crewai-memory/src/main.py @@ -0,0 +1,293 @@ +"""Incident Response Crew — a multi-agent simulation powered by KD6 memory. + +This example demonstrates a 4-agent crew handling a production incident. +Each agent uses KD6 as persistent, shared memory through CrewAI's Memory system. + +Scenario: + A fictional e-commerce platform "ShopStream" is experiencing elevated error + rates on its checkout service. The crew must investigate, diagnose, fix, + and document the incident. + +Agents: + 1. Incident Commander — coordinates response, tracks timeline, makes decisions + 2. Platform Engineer — investigates infrastructure, checks metrics and logs + 3. Backend Developer — analyzes code, identifies root cause, proposes fixes + 4. Communications Lead — drafts status updates and post-incident review + +KD6 features demonstrated: + - Memory layers: working (active investigation), semantic (extracted facts), + episodic (incident timeline events) + - Scoped visibility: each agent has its own scope, shared under /incident + - Knowledge graph: edges link related findings (e.g., "log entry" → "root cause") + - Cross-session persistence: run twice to see agents recall prior incidents +""" + +from __future__ import annotations + +import os +import sys +from pathlib import Path + +# Load .env if present +_env_file = Path(__file__).parent.parent / ".env" +if _env_file.exists(): + for line in _env_file.read_text().splitlines(): + line = line.strip() + if line and not line.startswith("#") and "=" in line: + key, _, value = line.partition("=") + os.environ.setdefault(key.strip(), value.strip()) + +from crewai import Agent, Crew, Process, Task + +from .config import ( + KD6_STORE_NAME, + KD6_TENANT_ID, + KD6_URL, + get_crewai_embedder_config, + get_crewai_llm, +) +from .kd6_backend import Kd6StorageBackend + +# ── Scenario context ───────────────────────────────────────────────────── + +INCIDENT_BRIEF = """\ +INCIDENT ALERT — ShopStream Checkout Service + +Severity: SEV-1 (customer-facing) +Started: 14:23 UTC +Detection: Automated alerting on 5xx error rate > 5% + +Symptoms: +- Checkout API returning HTTP 503 for ~30% of requests +- Payment processing timeouts increased from p99=200ms to p99=8000ms +- Cart abandonment rate spiked 4x in the last 15 minutes +- No recent deployments (last deploy was 6 hours ago) + +Affected services: checkout-api, payment-gateway, order-service +Infrastructure: Kubernetes cluster (us-east-1), PostgreSQL (RDS), Redis (ElastiCache) + +Initial metrics: +- checkout-api CPU: 45% (normal) +- checkout-api memory: 78% (elevated, usually 55%) +- PostgreSQL connections: 450/500 (near limit, usually 200) +- Redis hit rate: 62% (degraded, usually 95%) +""" + + +def build_crew() -> Crew: + """Construct the incident response crew with KD6 memory.""" + llm = get_crewai_llm() + + # ── KD6 memory backend ─────────────────────────────────────────── + kd6_backend = Kd6StorageBackend( + kd6_url=KD6_URL, + store_name=KD6_STORE_NAME, + tenant_id=KD6_TENANT_ID, + default_layer="semantic", + default_owner="incident-crew", + ) + + # ── Agents ─────────────────────────────────────────────────────── + + incident_commander = Agent( + role="Incident Commander", + goal=( + "Coordinate the incident response. Track the timeline of events, " + "delegate investigation tasks, and make decisions about mitigation " + "and resolution. Ensure all findings are documented." + ), + backstory=( + "You are a senior SRE who has led dozens of production incidents. " + "You know that clear communication and systematic investigation " + "are key. You maintain a detailed timeline and ensure nothing " + "falls through the cracks." + ), + llm=llm, + verbose=True, + ) + + platform_engineer = Agent( + role="Platform Engineer", + goal=( + "Investigate the infrastructure layer. Analyze metrics, check " + "resource utilization, connection pools, and network issues. " + "Report findings with specific data points." + ), + backstory=( + "You are an infrastructure specialist who lives in Grafana dashboards " + "and kubectl. You can read metrics patterns and correlate infrastructure " + "events with application behavior. You always cite specific numbers." + ), + llm=llm, + verbose=True, + ) + + backend_developer = Agent( + role="Backend Developer", + goal=( + "Analyze the application code and behavior. Identify the root cause " + "by correlating infrastructure findings with application patterns. " + "Propose specific code or configuration fixes." + ), + backstory=( + "You are a senior backend developer who wrote parts of the checkout " + "service. You understand connection pooling, caching strategies, " + "and common failure modes. You think in terms of code paths and " + "data flows." + ), + llm=llm, + verbose=True, + ) + + comms_lead = Agent( + role="Communications Lead", + goal=( + "Draft clear, accurate status updates for stakeholders and a " + "comprehensive post-incident review. Synthesize technical findings " + "into business-relevant language." + ), + backstory=( + "You are a technical program manager who bridges engineering and " + "business. You distill complex technical incidents into clear " + "narratives with actionable follow-ups. You always include " + "timeline, impact, root cause, and remediation." + ), + llm=llm, + verbose=True, + ) + + # ── Tasks (sequential investigation flow) ──────────────────────── + + triage = Task( + description=( + f"Review the incident alert and perform initial triage.\n\n" + f"{INCIDENT_BRIEF}\n\n" + "Create a structured timeline of events so far. Identify the most " + "likely failure domain (infrastructure, application, or external). " + "Assign specific investigation areas to the team. " + "Record your triage findings as structured observations." + ), + expected_output=( + "A triage report containing:\n" + "1. Incident timeline (what happened when)\n" + "2. Initial hypothesis for the failure domain\n" + "3. Specific investigation assignments\n" + "4. Severity confirmation and escalation status" + ), + agent=incident_commander, + ) + + investigate_infra = Task( + description=( + "Investigate the infrastructure based on the triage report. " + "Analyze the metrics provided in the incident alert:\n" + "- PostgreSQL connections at 450/500 (near exhaustion)\n" + "- Redis hit rate dropped from 95% to 62%\n" + "- Memory usage elevated from 55% to 78%\n\n" + "Determine: Is this a connection pool leak? A cache failure? " + "A resource exhaustion issue? Provide specific findings with " + "data points and your confidence level for each finding." + ), + expected_output=( + "Infrastructure investigation report with:\n" + "1. Database connection analysis (is the pool leaking?)\n" + "2. Redis cache degradation analysis\n" + "3. Memory pressure analysis\n" + "4. Correlation between these signals\n" + "5. Confidence-rated hypotheses" + ), + agent=platform_engineer, + context=[triage], + ) + + investigate_code = Task( + description=( + "Using the infrastructure findings, analyze the application " + "behavior to identify the root cause. Consider:\n" + "- Why would DB connections spike without a deployment?\n" + "- What could cause Redis hit rate to drop?\n" + "- How do these relate to the 503 errors?\n\n" + "Think about: connection pool configuration, cache invalidation " + "patterns, retry storms, and cascading failures. " + "Propose a specific root cause and a fix." + ), + expected_output=( + "Root cause analysis containing:\n" + "1. The specific root cause with supporting evidence\n" + "2. The failure cascade (what led to what)\n" + "3. Proposed immediate fix (mitigation)\n" + "4. Proposed permanent fix (prevention)\n" + "5. Risk assessment for each fix" + ), + agent=backend_developer, + context=[triage, investigate_infra], + ) + + write_postmortem = Task( + description=( + "Synthesize all findings into a post-incident review. " + "Include the full timeline, root cause, impact assessment, " + "resolution steps, and follow-up action items. " + "Write for both technical and non-technical stakeholders." + ), + expected_output=( + "Post-incident review document with:\n" + "1. Executive summary (2-3 sentences)\n" + "2. Timeline of events\n" + "3. Root cause explanation\n" + "4. Customer impact assessment\n" + "5. Resolution steps taken\n" + "6. Action items with owners and due dates\n" + "7. Lessons learned" + ), + agent=comms_lead, + context=[triage, investigate_infra, investigate_code], + ) + + # ── Build the crew ─────────────────────────────────────────────── + + crew = Crew( + agents=[incident_commander, platform_engineer, backend_developer, comms_lead], + tasks=[triage, investigate_infra, investigate_code, write_postmortem], + process=Process.sequential, + memory=True, + verbose=True, + ) + + # Wire up KD6 as the memory backend + from crewai.memory import Memory + + crew._memory = Memory( + storage=kd6_backend, + llm=llm, + embedder=get_crewai_embedder_config(), + root_scope="/incident/shopstream", + ) + + return crew + + +def main() -> None: + """Run the incident response crew.""" + print("=" * 70) + print(" ShopStream Incident Response Simulation") + print(" Powered by CrewAI + KD6 Memory") + print("=" * 70) + print() + + crew = build_crew() + + print("Starting incident response...\n") + result = crew.kickoff() + + print("\n" + "=" * 70) + print(" INCIDENT RESPONSE COMPLETE") + print("=" * 70) + print() + print(result) + print() + print("Memories persisted in KD6. Run again to see cross-session recall.") + + +if __name__ == "__main__": + main() diff --git a/examples/crewai-memory/src/smoke_test.py b/examples/crewai-memory/src/smoke_test.py new file mode 100644 index 0000000..8516541 --- /dev/null +++ b/examples/crewai-memory/src/smoke_test.py @@ -0,0 +1,178 @@ +"""Smoke test for the KD6 storage backend — no LLM calls required. + +Validates that the Kd6StorageBackend can perform CRUD operations against +a running KD6 server. Run this before the full crew to verify connectivity. + +Usage: + # Start KD6 server first: + KD6_DATABASE_URL="sqlite:kd6-test.db?mode=rwc" cargo run -p kd6-server + + # Then run smoke test: + python -m src.smoke_test +""" + +from __future__ import annotations + +import os +import sys +from datetime import datetime, timezone +from pathlib import Path + +# Load .env if present +_env_file = Path(__file__).parent.parent / ".env" +if _env_file.exists(): + for line in _env_file.read_text().splitlines(): + line = line.strip() + if line and not line.startswith("#") and "=" in line: + key, _, value = line.partition("=") + os.environ.setdefault(key.strip(), value.strip()) + +from crewai.memory.types import MemoryRecord + +from .kd6_backend import Kd6StorageBackend + +KD6_URL = os.environ.get("KD6_URL", "http://localhost:8080") +STORE_NAME = "crewai-smoke-test" +TENANT_ID = "smoke-test" + + +def _ok(msg: str) -> None: + print(f" ✓ {msg}") + + +def _fail(msg: str) -> None: + print(f" ✗ {msg}", file=sys.stderr) + sys.exit(1) + + +def main() -> None: + print("KD6 Storage Backend Smoke Test") + print(f" Server: {KD6_URL}") + print(f" Store: {STORE_NAME}") + print() + + # 1. Create backend (ensures store exists) + try: + backend = Kd6StorageBackend( + kd6_url=KD6_URL, + store_name=STORE_NAME, + tenant_id=TENANT_ID, + ) + _ok("Connected to KD6 and ensured store exists") + except Exception as e: + _fail(f"Failed to connect to KD6: {e}") + return + + # 2. Save records + records = [ + MemoryRecord( + id="smoke-1", + content="Redis cache hit rate dropped from 95% to 62%", + scope="/incident/test/infra", + categories=["infrastructure", "redis"], + importance=0.9, + source="smoke-test", + ), + MemoryRecord( + id="smoke-2", + content="PostgreSQL connections at 450 out of 500 limit", + scope="/incident/test/infra", + categories=["infrastructure", "database"], + importance=0.85, + source="smoke-test", + ), + MemoryRecord( + id="smoke-3", + content="Root cause identified: connection pool leak in checkout service", + scope="/incident/test/code", + categories=["root-cause", "code"], + importance=1.0, + source="smoke-test", + private=True, + ), + ] + + try: + backend.save(records) + _ok(f"Saved {len(records)} records") + except Exception as e: + _fail(f"Failed to save records: {e}") + + # 3. List records + try: + all_records = backend.list_records() + assert len(all_records) >= 3, f"Expected >= 3 records, got {len(all_records)}" + _ok(f"Listed {len(all_records)} records") + except Exception as e: + _fail(f"Failed to list records: {e}") + + # 4. Get single record + try: + rec = all_records[0] + kd6_id = rec.metadata.get("_kd6_id") + fetched = backend.get_record(kd6_id) + assert fetched is not None, "get_record returned None" + assert fetched.content, "Record has no content" + _ok(f"Retrieved record: {fetched.content[:50]}...") + except Exception as e: + _fail(f"Failed to get record: {e}") + + # 5. Update record + try: + rec = all_records[0] + rec.content = rec.content + " [UPDATED]" + backend.update(rec) + _ok("Updated record") + except Exception as e: + _fail(f"Failed to update record: {e}") + + # 6. Count records + try: + total = backend.count() + assert total >= 3, f"Expected >= 3, got {total}" + _ok(f"Count: {total} records total") + except Exception as e: + _fail(f"Failed to count: {e}") + + # 7. List scopes + try: + scopes = backend.list_scopes("/") + assert len(scopes) > 0, "No scopes found" + _ok(f"Scopes: {scopes}") + except Exception as e: + _fail(f"Failed to list scopes: {e}") + + # 8. List categories + try: + cats = backend.list_categories() + assert len(cats) > 0, "No categories found" + _ok(f"Categories: {cats}") + except Exception as e: + _fail(f"Failed to list categories: {e}") + + # 9. Delete specific records + try: + kd6_ids = [r.metadata["_kd6_id"] for r in all_records[:2]] + deleted = backend.delete(record_ids=kd6_ids) + assert deleted >= 2, f"Expected >= 2 deleted, got {deleted}" + _ok(f"Deleted {deleted} records by ID") + except Exception as e: + _fail(f"Failed to delete: {e}") + + # 10. Reset (clean up) + try: + backend.reset() + remaining = backend.count() + _ok(f"Reset complete ({remaining} records remaining)") + except Exception as e: + _fail(f"Failed to reset: {e}") + + backend.close() + + print() + print("All smoke tests passed! ✓") + print("The KD6 backend is ready for the CrewAI example.") + + +if __name__ == "__main__": + main()