From 82e90e9bea0da26e20d2d37add94b9cc29b42af7 Mon Sep 17 00:00:00 2001 From: ferhimedamine Date: Wed, 1 Jul 2026 10:08:18 +0000 Subject: [PATCH] examples: add persistent memory cookbook with Dakera integration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a new cookbook entry showing how to give any Instructor application persistent, decay-weighted long-term memory using Dakera (https://dakera.ai) — a self-hosted vector memory server. ## What's added ### `examples/persistent_memory_dakera/` - `dakera_memory.py` — integration helper: - `DakeraMemory`: sync httpx client for `/v1/memory/store`, `/v1/memory/search`, `/v1/memory/forget` - `AsyncDakeraMemory`: async variant using `httpx.AsyncClient` - `DakeraMemoryHook`: attaches to any instructor client via `client.on()`; auto-recalls before each call (`completion:kwargs`), auto-stores after (`completion:response`); Dakera errors are silently swallowed so they never crash an LLM call - `build_context_messages(messages, hits)`: inject recalled memories as a leading system message - `run.py` — four runnable examples: 1. Manual store + recall with `build_context_messages` 2. Zero-boilerplate hook-based approach (`hook.attach(client)`) 3. Structured extraction pipeline: `UserProfile` Pydantic model 4. Multi-turn CLI chat loop that persists across restarts - `test_dakera_memory.py` — 31 tests, all passing (fully mocked, no live server needed) ### `docs/examples/persistent_memory_dakera.md` Cookbook page with usage examples, API reference, and Quick Start. ### `mkdocs.yml` Entry added under `Cookbook:` alongside `Tracing with Langfuse`. ## Quick start ```bash docker run -p 3300:3300 -e DAKERA_API_KEY=demo ghcr.io/dakera-ai/dakera:latest pip install instructor openai httpx python examples/persistent_memory_dakera/run.py ``` Co-Authored-By: Paperclip --- docs/examples/persistent_memory_dakera.md | 328 +++++++++++++ .../persistent_memory_dakera/dakera_memory.py | 450 ++++++++++++++++++ examples/persistent_memory_dakera/run.py | 341 +++++++++++++ .../test_dakera_memory.py | 361 ++++++++++++++ mkdocs.yml | 1 + 5 files changed, 1481 insertions(+) create mode 100644 docs/examples/persistent_memory_dakera.md create mode 100644 examples/persistent_memory_dakera/dakera_memory.py create mode 100644 examples/persistent_memory_dakera/run.py create mode 100644 examples/persistent_memory_dakera/test_dakera_memory.py diff --git a/docs/examples/persistent_memory_dakera.md b/docs/examples/persistent_memory_dakera.md new file mode 100644 index 000000000..04c95c504 --- /dev/null +++ b/docs/examples/persistent_memory_dakera.md @@ -0,0 +1,328 @@ +--- +title: Persistent Memory with Dakera +description: Add long-term, decay-weighted memory to Instructor applications using the Dakera open-source memory server. +--- + +# Persistent Memory with Dakera + +By default, every `instructor.create()` call is stateless — the LLM has no +memory of previous conversations. This cookbook shows how to give your +Instructor-powered application **persistent, semantic memory** using +[Dakera](https://dakera.ai), a self-hosted vector memory server with +decay-weighted importance scoring. + +## Why Dakera? + +| Feature | Dakera | +|---------|--------| +| Self-hosted | Yes — Docker image, no external SaaS required | +| Decay weighting | Access-weighted importance; stale facts fade naturally | +| Semantic search | HNSW vector index for similarity recall | +| Auth | Optional Bearer token | +| Languages | REST API — any HTTP client works | + +## Prerequisites + +Start Dakera locally (requires Docker): + +```bash +docker run -p 3300:3300 -e DAKERA_API_KEY=demo \ + ghcr.io/dakera-ai/dakera:latest +``` + +Install dependencies: + +```bash +pip install instructor openai httpx +``` + +## The Dakera REST API + +Three endpoints are all you need: + +| Endpoint | Purpose | +|----------|---------| +| `POST /v1/memory/store` | Store a memory | +| `POST /v1/memory/search` | Semantic search over stored memories | +| `POST /v1/memory/forget` | Delete memories | + +## Integration Helper + +Save the following as `dakera_memory.py` alongside your project +([full source on GitHub](https://github.com/instructor-ai/instructor/blob/main/examples/persistent_memory_dakera/dakera_memory.py)): + +```python +import os +from dataclasses import dataclass, field +from typing import Any +import httpx + +@dataclass +class DakeraMemory: + base_url: str = field( + default_factory=lambda: os.environ.get("DAKERA_BASE_URL", "http://localhost:3300") + ) + api_key: str = field( + default_factory=lambda: os.environ.get("DAKERA_API_KEY", "") + ) + agent_id: str = "default" + session_id: str | None = None + timeout: float = 10.0 + + def __post_init__(self) -> None: + headers = {"Authorization": f"Bearer {self.api_key}"} if self.api_key else {} + self._client = httpx.Client(base_url=self.base_url, headers=headers, timeout=self.timeout) + + def store(self, content: str, *, importance: float | None = None, tags: list[str] | None = None) -> dict[str, Any]: + payload: dict[str, Any] = {"content": content, "agent_id": self.agent_id} + if self.session_id: + payload["session_id"] = self.session_id + if importance is not None: + payload["importance"] = importance + if tags is not None: + payload["tags"] = tags + resp = self._client.post("/v1/memory/store", json=payload) + resp.raise_for_status() + return resp.json().get("memory", {}) + + def search(self, query: str, *, top_k: int = 5) -> list[dict[str, Any]]: + resp = self._client.post("/v1/memory/search", json={"agent_id": self.agent_id, "query": query, "top_k": top_k}) + resp.raise_for_status() + return resp.json().get("memories", []) + + def forget(self, memory_ids: list[str] | None = None) -> None: + payload: dict[str, Any] = {"agent_id": self.agent_id} + if memory_ids is not None: + payload["memory_ids"] = memory_ids + self._client.post("/v1/memory/forget", json=payload).raise_for_status() + + def close(self) -> None: + self._client.close() + + def __enter__(self): return self + def __exit__(self, *_): self.close() +``` + +## Approach 1 — Manual Store and Recall + +Full explicit control: store facts yourself, recall them before each LLM call, +decide what to remember afterwards. + +```python +import instructor +from openai import OpenAI +from pydantic import BaseModel, Field +from typing import Optional +from dakera_memory import DakeraMemory + +class ChatReply(BaseModel): + text: str + should_remember: bool = Field(description="True if this contains a long-term fact") + memory_hint: Optional[str] = Field(None, description="Compact version of the fact to store") + +client = instructor.from_openai(OpenAI()) +mem = DakeraMemory(api_key="demo", agent_id="alice") + +# Store some facts from a previous session +mem.store("The user prefers Python over JavaScript.", importance=0.9, tags=["preference"]) +mem.store("The user is building an e-commerce chatbot.", importance=0.8, tags=["project"]) + +# New query — recall relevant memories first +query = "What Python framework should I use for my chatbot backend?" +hits = mem.search(query, top_k=3) + +# Build context: inject memories as a leading system message +memory_block = "\n".join( + f"- [{h['score']:.2f}] {h['memory']['content']}" + for h in hits +) +messages = [ + {"role": "system", "content": f"Relevant memories:\n{memory_block}"}, + {"role": "user", "content": query}, +] + +reply = client.create(model="gpt-4.1-mini", response_model=ChatReply, messages=messages) +print(reply.text) +# → "Given that you prefer Python and are building an e-commerce chatbot, +# FastAPI is an excellent choice — it's fast, type-safe, and pairs well +# with Pydantic models that instructor already uses." + +# Store anything worth keeping +if reply.should_remember and reply.memory_hint: + mem.store(reply.memory_hint, importance=0.6, tags=["advice"]) +``` + +## Approach 2 — Hook-Based (Zero Boilerplate) + +Attach a hook once; every subsequent `create()` call automatically recalls +relevant memories and stores the reply. + +```python +import instructor +from openai import OpenAI +from pydantic import BaseModel +from dakera_memory import DakeraMemory, DakeraMemoryHook + +class Reply(BaseModel): + text: str + +client = instructor.from_openai(OpenAI()) +mem = DakeraMemory(api_key="demo", agent_id="bob") +hook = DakeraMemoryHook(mem, top_k=3, auto_recall=True, auto_store=True) +hook.attach(client) + +# Memory is injected automatically — no extra code needed +reply = client.create( + model="gpt-4.1-mini", + response_model=Reply, + messages=[{"role": "user", "content": "What did I tell you about my project?"}], +) +print(reply.text) + +# Clean up when done +hook.detach(client) +mem.close() +``` + +### How the Hook Works + +The `DakeraMemoryHook` registers two instructor event handlers: + +| Hook event | Action | +|-----------|--------| +| `completion:kwargs` | Extracts the last user message, searches Dakera, prepends a system memory block | +| `completion:response` | Extracts the assistant reply text, stores it with configurable importance | + +```python +# Hook constructor options +hook = DakeraMemoryHook( + mem, + top_k=5, # how many memories to inject + auto_recall=True, # inject memories before each call + auto_store=True, # store replies after each call + store_importance=0.6, # decay weight for auto-stored replies + store_tags=["instructor"], +) +``` + +## Approach 3 — Structured Extraction Pipeline + +Use Instructor's structured output to extract a `UserProfile` Pydantic model, +then persist each field as a separate searchable memory for fine-grained recall. + +```python +import instructor +from openai import OpenAI +from pydantic import BaseModel, Field +from typing import Optional +from dakera_memory import DakeraMemory + +class UserProfile(BaseModel): + name: str + interests: list[str] + location: Optional[str] = None + profession: Optional[str] = None + +client = instructor.from_openai(OpenAI()) +mem = DakeraMemory(api_key="demo", agent_id="carol") + +intro = "I'm Sam, a data engineer in Amsterdam. I love hiking, cooking, and Rust." +profile = client.create( + model="gpt-4.1-mini", + response_model=UserProfile, + messages=[{"role": "user", "content": f"Extract a profile from: {intro}"}], +) + +# Store each fact separately for fine-grained retrieval +for interest in profile.interests: + mem.store(f"{profile.name} is interested in {interest}.", importance=0.7, tags=["interest"]) +if profile.profession: + mem.store(f"{profile.name} works as {profile.profession}.", importance=0.9, tags=["profession"]) +if profile.location: + mem.store(f"{profile.name} lives in {profile.location}.", importance=0.6, tags=["location"]) +``` + +## Approach 4 — Persistent Multi-Turn Chat + +A complete CLI chatbot that remembers facts across restarts. +Run it twice: what you say in session 1 is recalled in session 2. + +```python +import instructor +from openai import OpenAI +from pydantic import BaseModel, Field +from typing import Optional +from dakera_memory import DakeraMemory, build_context_messages + +class ChatReply(BaseModel): + text: str + should_remember: bool + memory_hint: Optional[str] = None + +client = instructor.from_openai(OpenAI()) +mem = DakeraMemory(api_key="demo", agent_id="persistent-chat") +history = [] +system = {"role": "system", "content": "You are a helpful assistant with long-term memory."} + +while True: + user_input = input("You: ").strip() + if user_input.lower() in ("quit", "exit"): + break + + history.append({"role": "user", "content": user_input}) + hits = mem.search(user_input, top_k=3) + messages = build_context_messages([system] + history, hits) + + reply = client.create(model="gpt-4.1-mini", response_model=ChatReply, messages=messages) + print(f"Assistant: {reply.text}") + history.append({"role": "assistant", "content": reply.text}) + + if reply.should_remember and reply.memory_hint: + mem.store(reply.memory_hint, importance=0.75, tags=["chat"]) + print(f" [Remembered: {reply.memory_hint}]") + +mem.close() +``` + +## Complete Working Example + +The full source code for all four approaches, plus a test suite, is available at: + +- `examples/persistent_memory_dakera/dakera_memory.py` — core helpers +- `examples/persistent_memory_dakera/run.py` — cookbook examples 1–4 +- `examples/persistent_memory_dakera/test_dakera_memory.py` — pytest suite + +## Key Patterns + +### Memory decay + +Dakera weights memories by access frequency and configurable importance (0–1). +Rarely-accessed memories naturally become less influential without manual cleanup. +Use higher importance values for critical user facts: + +```python +mem.store("User's name is Alice", importance=0.95) # durable +mem.store("User asked about the weather", importance=0.2) # ephemeral +``` + +### Session scoping + +Pass `session_id` to scope memories to the current conversation, while +`agent_id` gives cross-session identity: + +```python +mem = DakeraMemory(api_key="demo", agent_id="alice", session_id="session-2026-07-01") +``` + +### Error resilience + +Never let a memory failure crash your LLM call. The hook implementation +catches all exceptions from Dakera and logs a warning: + +```python +try: + hits = self.memory.search(user_text, top_k=self.top_k) +except Exception as exc: + logger.warning("Dakera memory search failed: %s", exc) + return # continue without memory context +``` diff --git a/examples/persistent_memory_dakera/dakera_memory.py b/examples/persistent_memory_dakera/dakera_memory.py new file mode 100644 index 000000000..ac75f93d6 --- /dev/null +++ b/examples/persistent_memory_dakera/dakera_memory.py @@ -0,0 +1,450 @@ +""" +dakera_memory.py — Dakera persistent memory integration for Instructor. + +This module provides: +- DakeraMemory: a lightweight client wrapping the Dakera REST API +- DakeraMemoryHook: a hook-based mixin that auto-stores + auto-retrieves + memories around every instructor.create() / instructor.chat.completions.create() call +- build_context_messages(): a helper to inject recalled memories as a + system message into a message list before calling the LLM + +Usage +----- +```python +import instructor +from dakera_memory import DakeraMemory, DakeraMemoryHook + +client = instructor.from_provider("openai/gpt-4.1-mini") +mem = DakeraMemory(base_url="http://localhost:3300", api_key="demo", agent_id="my-agent") +hook = DakeraMemoryHook(mem) +hook.attach(client) + +from pydantic import BaseModel + +class Reply(BaseModel): + text: str + +messages = [{"role": "user", "content": "What did I have for breakfast?"}] +messages = hook.recall_into_messages(messages) + +reply = client.create(messages=messages, response_model=Reply) +hook.store("What did I have for breakfast?", reply.text) +``` +""" + +from __future__ import annotations + +import logging +import os +from dataclasses import dataclass, field +from typing import Any + +import httpx + +logger = logging.getLogger(__name__) + +_DEFAULT_BASE_URL = "http://localhost:3300" +_DEFAULT_TOP_K = 5 + + +# --------------------------------------------------------------------------- +# Core REST client +# --------------------------------------------------------------------------- + + +@dataclass +class DakeraMemory: + """Thin wrapper around the Dakera REST API. + + Args: + base_url: Base URL of the Dakera server (default ``http://localhost:3300``). + api_key: Bearer token. Falls back to the ``DAKERA_API_KEY`` env var. + agent_id: Logical agent / user identifier used to namespace memories. + session_id: Optional session scope for the current conversation. + timeout: HTTP timeout in seconds (default 10). + """ + + base_url: str = field(default_factory=lambda: os.environ.get("DAKERA_BASE_URL", _DEFAULT_BASE_URL)) + api_key: str = field(default_factory=lambda: os.environ.get("DAKERA_API_KEY", "")) + agent_id: str = "default" + session_id: str | None = None + timeout: float = 10.0 + + def __post_init__(self) -> None: + self._client = httpx.Client( + base_url=self.base_url, + headers=self._auth_headers(), + timeout=self.timeout, + ) + + def _auth_headers(self) -> dict[str, str]: + if self.api_key: + return {"Authorization": f"Bearer {self.api_key}"} + return {} + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def store( + self, + content: str, + *, + importance: float | None = None, + tags: list[str] | None = None, + ) -> dict[str, Any]: + """Store a memory and return the created record. + + Args: + content: The text to remember. + importance: Optional 0-1 weight hint for the decay engine. + tags: Optional list of tag strings. + + Returns: + The full memory record dict as returned by the server. + """ + payload: dict[str, Any] = { + "content": content, + "agent_id": self.agent_id, + } + if self.session_id is not None: + payload["session_id"] = self.session_id + if importance is not None: + payload["importance"] = importance + if tags is not None: + payload["tags"] = tags + + resp = self._client.post("/v1/memory/store", json=payload) + resp.raise_for_status() + return resp.json().get("memory", {}) + + def search(self, query: str, *, top_k: int = _DEFAULT_TOP_K) -> list[dict[str, Any]]: + """Search memories by semantic similarity. + + Args: + query: Natural-language query string. + top_k: Maximum number of results to return (default 5). + + Returns: + List of ``{"memory": {...}, "score": float}`` dicts, most relevant first. + """ + payload: dict[str, Any] = { + "agent_id": self.agent_id, + "query": query, + "top_k": top_k, + } + resp = self._client.post("/v1/memory/search", json=payload) + resp.raise_for_status() + return resp.json().get("memories", []) + + def forget(self, memory_ids: list[str] | None = None) -> None: + """Delete specific memories (or all memories for the agent if omitted). + + Args: + memory_ids: List of memory IDs to delete. Pass ``None`` to forget everything. + """ + payload: dict[str, Any] = {"agent_id": self.agent_id} + if memory_ids is not None: + payload["memory_ids"] = memory_ids + resp = self._client.post("/v1/memory/forget", json=payload) + resp.raise_for_status() + + def close(self) -> None: + """Close the underlying HTTP connection pool.""" + self._client.close() + + def __enter__(self) -> "DakeraMemory": + return self + + def __exit__(self, *_: Any) -> None: + self.close() + + +# --------------------------------------------------------------------------- +# Async variant +# --------------------------------------------------------------------------- + + +@dataclass +class AsyncDakeraMemory: + """Async counterpart of :class:`DakeraMemory` — use inside ``async`` code.""" + + base_url: str = field(default_factory=lambda: os.environ.get("DAKERA_BASE_URL", _DEFAULT_BASE_URL)) + api_key: str = field(default_factory=lambda: os.environ.get("DAKERA_API_KEY", "")) + agent_id: str = "default" + session_id: str | None = None + timeout: float = 10.0 + + def __post_init__(self) -> None: + self._client = httpx.AsyncClient( + base_url=self.base_url, + headers={"Authorization": f"Bearer {self.api_key}"} if self.api_key else {}, + timeout=self.timeout, + ) + + async def store( + self, + content: str, + *, + importance: float | None = None, + tags: list[str] | None = None, + ) -> dict[str, Any]: + payload: dict[str, Any] = {"content": content, "agent_id": self.agent_id} + if self.session_id is not None: + payload["session_id"] = self.session_id + if importance is not None: + payload["importance"] = importance + if tags is not None: + payload["tags"] = tags + resp = await self._client.post("/v1/memory/store", json=payload) + resp.raise_for_status() + return resp.json().get("memory", {}) + + async def search(self, query: str, *, top_k: int = _DEFAULT_TOP_K) -> list[dict[str, Any]]: + payload: dict[str, Any] = {"agent_id": self.agent_id, "query": query, "top_k": top_k} + resp = await self._client.post("/v1/memory/search", json=payload) + resp.raise_for_status() + return resp.json().get("memories", []) + + async def forget(self, memory_ids: list[str] | None = None) -> None: + payload: dict[str, Any] = {"agent_id": self.agent_id} + if memory_ids is not None: + payload["memory_ids"] = memory_ids + resp = await self._client.post("/v1/memory/forget", json=payload) + resp.raise_for_status() + + async def aclose(self) -> None: + await self._client.aclose() + + async def __aenter__(self) -> "AsyncDakeraMemory": + return self + + async def __aexit__(self, *_: Any) -> None: + await self.aclose() + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def build_context_messages( + messages: list[dict[str, Any]], + memories: list[dict[str, Any]], + *, + header: str = "Relevant memories from previous conversations:", + max_memories: int = 5, +) -> list[dict[str, Any]]: + """Prepend a system message containing recalled memories to *messages*. + + The injected system message is inserted **before** any existing system + messages so the LLM sees the memory context first. + + Args: + messages: The original message list to be passed to the LLM. + memories: Result of :meth:`DakeraMemory.search` — list of + ``{"memory": {"content": ...}, "score": float}`` dicts. + header: Heading line that introduces the memory block. + max_memories: Cap on how many memories to inject (default 5). + + Returns: + A new message list with a system memory block prepended. + """ + if not memories: + return messages + + items = memories[:max_memories] + lines = [header, ""] + for i, hit in enumerate(items, 1): + content = hit.get("memory", {}).get("content", "") + score = hit.get("score", 0.0) + lines.append(f"{i}. [{score:.2f}] {content}") + + memory_message: dict[str, Any] = { + "role": "system", + "content": "\n".join(lines), + } + + # Keep existing system messages; prepend the memory block. + return [memory_message] + list(messages) + + +# --------------------------------------------------------------------------- +# Hook-based integration +# --------------------------------------------------------------------------- + + +class DakeraMemoryHook: + """Instructor hook that automatically stores and retrieves memories. + + Attach to any instructor-patched client once; thereafter every + ``create()`` call will have past memories woven into the system context. + + Typical usage:: + + client = instructor.from_provider("openai/gpt-4.1-mini") + mem = DakeraMemory(api_key="demo", agent_id="alice") + hook = DakeraMemoryHook(mem, top_k=3) + hook.attach(client) + + When a ``completion:kwargs`` event fires: + + 1. The hook extracts the last user message as the query. + 2. It searches Dakera for relevant memories. + 3. It injects those memories as a leading system message. + + When a ``completion:response`` event fires: + + 1. The hook extracts the assistant's reply text. + 2. It stores that reply as a new memory (with ``importance=0.6``). + + You can disable auto-recall or auto-store independently:: + + hook = DakeraMemoryHook(mem, auto_recall=True, auto_store=False) + + If you prefer full manual control, skip :meth:`attach` and call + :meth:`recall_into_messages` / :meth:`store` yourself. + """ + + def __init__( + self, + memory: DakeraMemory, + *, + top_k: int = _DEFAULT_TOP_K, + auto_recall: bool = True, + auto_store: bool = True, + store_importance: float = 0.6, + store_tags: list[str] | None = None, + ) -> None: + self.memory = memory + self.top_k = top_k + self.auto_recall = auto_recall + self.auto_store = auto_store + self.store_importance = store_importance + self.store_tags = store_tags or ["instructor"] + + # ------------------------------------------------------------------ + # Instructor hook handlers + # ------------------------------------------------------------------ + + def _on_completion_kwargs(self, *args: Any, **kwargs: Any) -> None: + """Fires before each LLM call — inject memory context.""" + if not self.auto_recall: + return + messages: list[dict[str, Any]] = kwargs.get("messages", []) + if not messages: + return + + # Use the last user message as the semantic query. + user_text = next( + (m.get("content", "") for m in reversed(messages) if m.get("role") == "user"), + None, + ) + if not user_text: + return + + try: + hits = self.memory.search(user_text, top_k=self.top_k) + except Exception as exc: # never crash the LLM call on a memory error + logger.warning("Dakera memory search failed: %s", exc) + return + + if not hits: + return + + enriched = build_context_messages(messages, hits, max_memories=self.top_k) + # Mutate the messages list in-place so instructor sees the update. + messages.clear() + messages.extend(enriched) + + def _on_completion_response(self, response: Any) -> None: + """Fires after a successful LLM call — persist the reply.""" + if not self.auto_store: + return + try: + # Works with openai ChatCompletion objects and most other providers. + choice = response.choices[0] + text: str = "" + if hasattr(choice, "message") and hasattr(choice.message, "content"): + text = choice.message.content or "" + elif hasattr(choice, "text"): + text = choice.text or "" + if not text: + return + self.memory.store( + text, + importance=self.store_importance, + tags=self.store_tags, + ) + except Exception as exc: + logger.warning("Dakera memory store failed: %s", exc) + + def attach(self, client: Any) -> None: + """Register hooks on an instructor-patched *client*. + + Args: + client: Any object returned by ``instructor.from_provider()``, + ``instructor.from_openai()``, etc. + """ + client.on("completion:kwargs", self._on_completion_kwargs) + client.on("completion:response", self._on_completion_response) + + def detach(self, client: Any) -> None: + """Remove previously registered hooks from *client*.""" + client.off("completion:kwargs", self._on_completion_kwargs) + client.off("completion:response", self._on_completion_response) + + # ------------------------------------------------------------------ + # Manual helpers (bypass hooks) + # ------------------------------------------------------------------ + + def recall_into_messages( + self, + messages: list[dict[str, Any]], + query: str | None = None, + ) -> list[dict[str, Any]]: + """Return *messages* with a prepended memory-context system message. + + Useful when you want manual control instead of automatic hooks. + + Args: + messages: The original message list. + query: Query to search memories with. Defaults to the last + user message in *messages*. + + Returns: + A new message list (original is not mutated). + """ + if query is None: + query = next( + (m.get("content", "") for m in reversed(messages) if m.get("role") == "user"), + None, + ) + if not query: + return messages + + hits = self.memory.search(query, top_k=self.top_k) + return build_context_messages(messages, hits, max_memories=self.top_k) + + def store( + self, + content: str, + *, + importance: float | None = None, + tags: list[str] | None = None, + ) -> dict[str, Any]: + """Manually store *content* as a memory. + + Args: + content: Text to persist. + importance: Override the default importance weight. + tags: Override the default tags. + + Returns: + The created memory record. + """ + return self.memory.store( + content, + importance=importance if importance is not None else self.store_importance, + tags=tags if tags is not None else self.store_tags, + ) diff --git a/examples/persistent_memory_dakera/run.py b/examples/persistent_memory_dakera/run.py new file mode 100644 index 000000000..c0a7990da --- /dev/null +++ b/examples/persistent_memory_dakera/run.py @@ -0,0 +1,341 @@ +""" +Persistent Memory with Dakera — Instructor Cookbook +===================================================== + +This example shows how to give an Instructor-powered chatbot **long-term +memory across sessions** using Dakera, a self-hosted decay-weighted vector +memory server. + +Prerequisites +------------- +1. Start Dakera locally (Docker): + + docker run -p 3300:3300 -e DAKERA_API_KEY=demo \ + ghcr.io/dakera-ai/dakera:latest + +2. Install dependencies: + + pip install instructor openai httpx + +3. Export your LLM key: + + export OPENAI_API_KEY=sk-... + +What this example demonstrates +------------------------------- +- Storing facts as memories after each LLM exchange +- Retrieving relevant memories and injecting them as context +- Structured extraction (Pydantic models) combined with persistent memory +- A multi-turn CLI chat loop that remembers across restarts +- The "hook" approach (automatic) vs the "manual" approach +""" + +from __future__ import annotations + +import os +import textwrap +from typing import Optional + +import instructor +from openai import OpenAI +from pydantic import BaseModel, Field + +from dakera_memory import DakeraMemory, DakeraMemoryHook, build_context_messages + +# --------------------------------------------------------------------------- +# Configuration — override via environment variables or edit here. +# --------------------------------------------------------------------------- +DAKERA_URL = os.environ.get("DAKERA_BASE_URL", "http://localhost:3300") +DAKERA_KEY = os.environ.get("DAKERA_API_KEY", "demo") +AGENT_ID = os.environ.get("DAKERA_AGENT_ID", "instructor-demo") +LLM_MODEL = os.environ.get("LLM_MODEL", "gpt-4.1-mini") + + +# --------------------------------------------------------------------------- +# Pydantic models +# --------------------------------------------------------------------------- + + +class UserProfile(BaseModel): + """Structured profile extracted from a user's self-introduction.""" + + name: str = Field(description="The user's name") + interests: list[str] = Field(description="List of the user's interests or hobbies") + location: Optional[str] = Field(None, description="Where the user lives, if mentioned") + profession: Optional[str] = Field(None, description="The user's job or profession, if mentioned") + + +class ChatReply(BaseModel): + """A single assistant reply in a conversation.""" + + text: str = Field(description="The assistant's reply to the user") + should_remember: bool = Field( + description="True if this exchange contains information worth remembering long-term" + ) + memory_hint: Optional[str] = Field( + None, + description="A compact rephrasing of the key fact(s) to store if should_remember is True", + ) + + +# --------------------------------------------------------------------------- +# Example 1 — Manual store/recall (explicit control) +# --------------------------------------------------------------------------- + + +def example_manual_store_and_recall() -> None: + """Demonstrate manually storing and recalling memories.""" + print("=" * 60) + print("Example 1: Manual store + recall") + print("=" * 60) + + client = instructor.from_openai(OpenAI()) + mem = DakeraMemory(base_url=DAKERA_URL, api_key=DAKERA_KEY, agent_id=AGENT_ID) + + # --- Store some facts ------------------------------------------------- + facts = [ + "The user prefers Python over JavaScript.", + "The user is building a chatbot for their e-commerce startup.", + "The user mentioned they have trouble with async code in Python.", + "The user's name is Alex and they live in Berlin.", + ] + for fact in facts: + record = mem.store(fact, importance=0.8, tags=["profile", "preferences"]) + print(f"Stored: {record.get('id', '?')} — {fact[:60]}") + + print() + + # --- Recall on a new question ----------------------------------------- + question = "How should I structure my Python chatbot project?" + hits = mem.search(question, top_k=3) + messages = build_context_messages( + [{"role": "user", "content": question}], + hits, + ) + + print("Messages sent to LLM:") + for m in messages: + role = m["role"].upper() + content = textwrap.shorten(m["content"], width=120, placeholder="...") + print(f" [{role}] {content}") + print() + + reply = client.create( + model=LLM_MODEL, + response_model=ChatReply, + messages=messages, + ) + print(f"Assistant: {reply.text}") + print(f"Should remember: {reply.should_remember}") + if reply.memory_hint: + stored = mem.store(reply.memory_hint, importance=0.5, tags=["assistant-advice"]) + print(f"Stored advice memory: {stored.get('id', '?')}") + + mem.close() + print() + + +# --------------------------------------------------------------------------- +# Example 2 — Hook-based (automatic, zero boilerplate) +# --------------------------------------------------------------------------- + + +def example_hook_based_memory() -> None: + """Demonstrate the automatic hook approach.""" + print("=" * 60) + print("Example 2: Hook-based automatic memory") + print("=" * 60) + + raw_client = OpenAI() + client = instructor.from_openai(raw_client) + + mem = DakeraMemory(base_url=DAKERA_URL, api_key=DAKERA_KEY, agent_id=f"{AGENT_ID}-hook") + hook = DakeraMemoryHook(mem, top_k=3, auto_recall=True, auto_store=True) + hook.attach(client) + + # The hook automatically injects recalled memories before this call + # and stores the assistant's reply afterwards. + reply = client.create( + model=LLM_MODEL, + response_model=ChatReply, + messages=[ + { + "role": "system", + "content": "You are a helpful programming assistant. Be concise.", + }, + { + "role": "user", + "content": "What's the best way to handle rate limits in an async Python HTTP client?", + }, + ], + ) + print(f"Assistant (hook): {reply.text[:200]}...") + print() + + # Detach when done so the hook doesn't fire on unrelated calls. + hook.detach(client) + mem.close() + + +# --------------------------------------------------------------------------- +# Example 3 — Structured extraction → memory pipeline +# --------------------------------------------------------------------------- + + +def example_structured_extraction_to_memory() -> None: + """Extract a structured UserProfile and persist it as a memory.""" + print("=" * 60) + print("Example 3: Structured extraction → memory") + print("=" * 60) + + client = instructor.from_openai(OpenAI()) + mem = DakeraMemory(base_url=DAKERA_URL, api_key=DAKERA_KEY, agent_id=f"{AGENT_ID}-profile") + + introduction = ( + "Hi! I'm Sam, a data engineer at a fintech company in Amsterdam. " + "I love hiking, cooking Italian food, and I've been learning Rust lately. " + "Most of my work involves building data pipelines with Apache Spark." + ) + + profile = client.create( + model=LLM_MODEL, + response_model=UserProfile, + messages=[ + {"role": "user", "content": f"Extract a user profile from: {introduction}"}, + ], + ) + print(f"Extracted profile: {profile.model_dump_json(indent=2)}") + + # Store each interest separately for fine-grained retrieval. + for interest in profile.interests: + mem.store( + f"{profile.name} is interested in {interest}.", + importance=0.7, + tags=["profile", "interest"], + ) + if profile.profession: + mem.store( + f"{profile.name} works as {profile.profession}.", + importance=0.9, + tags=["profile", "profession"], + ) + if profile.location: + mem.store( + f"{profile.name} lives in {profile.location}.", + importance=0.6, + tags=["profile", "location"], + ) + + print(f"\nStored {len(profile.interests) + 2} memories for {profile.name}") + + # Later: retrieve context for a relevant query + query = "What programming topics might this user want help with?" + hits = mem.search(query, top_k=4) + messages = build_context_messages( + [{"role": "user", "content": query}], + hits, + ) + recommendation = client.create( + model=LLM_MODEL, + response_model=ChatReply, + messages=messages, + ) + print(f"\nPersonalized recommendation: {recommendation.text}") + + mem.close() + print() + + +# --------------------------------------------------------------------------- +# Example 4 — Multi-turn CLI chat loop with persistent memory +# --------------------------------------------------------------------------- + + +def chat_loop() -> None: + """Interactive chat loop that remembers facts across restarts. + + Run this script twice: facts you share in session 1 will be recalled + in session 2, because memories persist in Dakera between runs. + """ + print("=" * 60) + print("Example 4: Multi-turn chat with persistent memory") + print("Type 'quit' to exit, 'forget' to wipe all memories.") + print("=" * 60) + + client = instructor.from_openai(OpenAI()) + mem = DakeraMemory( + base_url=DAKERA_URL, + api_key=DAKERA_KEY, + agent_id=f"{AGENT_ID}-chat", + ) + + history: list[dict[str, str]] = [] + system_prompt = { + "role": "system", + "content": ( + "You are a helpful assistant with long-term memory. " + "When you see 'Relevant memories' at the top, use them to personalise your replies. " + "Be concise and friendly." + ), + } + + while True: + try: + user_input = input("\nYou: ").strip() + except (EOFError, KeyboardInterrupt): + print("\nBye!") + break + + if not user_input: + continue + if user_input.lower() == "quit": + break + if user_input.lower() == "forget": + mem.forget() + history.clear() + print("[All memories cleared]") + continue + + history.append({"role": "user", "content": user_input}) + + # Recall relevant past memories and inject them. + hits = mem.search(user_input, top_k=3) + messages_with_context = build_context_messages( + [system_prompt] + history, + hits, + ) + + reply = client.create( + model=LLM_MODEL, + response_model=ChatReply, + messages=messages_with_context, + ) + + print(f"Assistant: {reply.text}") + history.append({"role": "assistant", "content": reply.text}) + + # Decide what to persist. + if reply.should_remember and reply.memory_hint: + stored = mem.store(reply.memory_hint, importance=0.75, tags=["chat"]) + print(f" [Remembered: {reply.memory_hint[:80]}]") + + mem.close() + + +# --------------------------------------------------------------------------- +# Entrypoint +# --------------------------------------------------------------------------- + +if __name__ == "__main__": + import sys + + mode = sys.argv[1] if len(sys.argv) > 1 else "all" + + if mode in ("all", "1"): + example_manual_store_and_recall() + if mode in ("all", "2"): + example_hook_based_memory() + if mode in ("all", "3"): + example_structured_extraction_to_memory() + if mode in ("all", "chat", "4"): + chat_loop() diff --git a/examples/persistent_memory_dakera/test_dakera_memory.py b/examples/persistent_memory_dakera/test_dakera_memory.py new file mode 100644 index 000000000..fd3c04277 --- /dev/null +++ b/examples/persistent_memory_dakera/test_dakera_memory.py @@ -0,0 +1,361 @@ +""" +Tests for dakera_memory.py — no live server or LLM required. + +Run with: + pytest test_dakera_memory.py -v +""" + +from __future__ import annotations + +import json +from unittest.mock import MagicMock, patch, call + +import httpx +import pytest + +from dakera_memory import ( + DakeraMemory, + DakeraMemoryHook, + build_context_messages, +) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +def _mock_response(payload: dict, status_code: int = 200) -> httpx.Response: + """Return a minimal httpx.Response wrapping *payload*. + + httpx.Response.raise_for_status() requires ``._request`` to be set; + we attach a minimal dummy request so the check doesn't blow up. + """ + resp = httpx.Response( + status_code=status_code, + content=json.dumps(payload).encode(), + headers={"content-type": "application/json"}, + ) + # Attach a dummy request so raise_for_status() can inspect it. + resp.request = httpx.Request("POST", "http://localhost:3300/v1/memory/store") + return resp + + +@pytest.fixture() +def mem(monkeypatch): + """DakeraMemory instance with a mocked httpx.Client.""" + with patch("dakera_memory.httpx.Client") as mock_client_cls: + mock_client = MagicMock() + mock_client_cls.return_value = mock_client + instance = DakeraMemory( + base_url="http://localhost:3300", + api_key="test-key", + agent_id="test-agent", + ) + instance._client = mock_client # expose for assertions + yield instance + + +# --------------------------------------------------------------------------- +# DakeraMemory unit tests +# --------------------------------------------------------------------------- + + +class TestDakeraMemoryStore: + def test_store_calls_correct_endpoint(self, mem): + mem._client.post.return_value = _mock_response( + {"memory": {"id": "abc123", "content": "hello"}} + ) + result = mem.store("hello") + + mem._client.post.assert_called_once_with( + "/v1/memory/store", + json={ + "content": "hello", + "agent_id": "test-agent", + }, + ) + assert result["id"] == "abc123" + + def test_store_with_all_optional_fields(self, mem): + mem._client.post.return_value = _mock_response( + {"memory": {"id": "xyz"}} + ) + mem.store( + "some content", + importance=0.9, + tags=["a", "b"], + ) + _, kwargs = mem._client.post.call_args + payload = kwargs["json"] + assert payload["importance"] == 0.9 + assert payload["tags"] == ["a", "b"] + + def test_store_with_session_id(self, mem): + mem.session_id = "session-42" + mem._client.post.return_value = _mock_response({"memory": {}}) + mem.store("content") + _, kwargs = mem._client.post.call_args + assert kwargs["json"]["session_id"] == "session-42" + + def test_store_raises_on_http_error(self, mem): + mem._client.post.return_value = _mock_response({}, status_code=500) + with pytest.raises(httpx.HTTPStatusError): + mem.store("bad") + + +class TestDakeraMemorySearch: + def test_search_calls_correct_endpoint(self, mem): + mem._client.post.return_value = _mock_response( + {"memories": [{"memory": {"content": "Python tip"}, "score": 0.91}]} + ) + results = mem.search("Python", top_k=3) + + mem._client.post.assert_called_once_with( + "/v1/memory/search", + json={"agent_id": "test-agent", "query": "Python", "top_k": 3}, + ) + assert len(results) == 1 + assert results[0]["score"] == 0.91 + + def test_search_returns_empty_list_when_no_memories(self, mem): + mem._client.post.return_value = _mock_response({"memories": []}) + assert mem.search("anything") == [] + + def test_search_default_top_k_is_five(self, mem): + mem._client.post.return_value = _mock_response({"memories": []}) + mem.search("query") + _, kwargs = mem._client.post.call_args + assert kwargs["json"]["top_k"] == 5 + + +class TestDakeraMemoryForget: + def test_forget_all(self, mem): + mem._client.post.return_value = _mock_response({}) + mem.forget() + mem._client.post.assert_called_once_with( + "/v1/memory/forget", + json={"agent_id": "test-agent"}, + ) + + def test_forget_specific_ids(self, mem): + mem._client.post.return_value = _mock_response({}) + mem.forget(memory_ids=["id1", "id2"]) + _, kwargs = mem._client.post.call_args + assert kwargs["json"]["memory_ids"] == ["id1", "id2"] + + +class TestDakeraMemoryContextManager: + def test_context_manager_calls_close(self): + with patch("dakera_memory.httpx.Client") as mock_cls: + mock_client = MagicMock() + mock_cls.return_value = mock_client + with DakeraMemory(api_key="x", agent_id="y") as m: + pass + mock_client.close.assert_called_once() + + +# --------------------------------------------------------------------------- +# build_context_messages tests +# --------------------------------------------------------------------------- + + +class TestBuildContextMessages: + def _make_hits(self, *contents: str) -> list[dict]: + return [ + {"memory": {"content": c}, "score": round(0.99 - i * 0.1, 2)} + for i, c in enumerate(contents) + ] + + def test_prepends_system_message(self): + messages = [{"role": "user", "content": "hello"}] + hits = self._make_hits("Fact A") + result = build_context_messages(messages, hits) + + assert len(result) == 2 + assert result[0]["role"] == "system" + assert "Fact A" in result[0]["content"] + assert result[1] == messages[0] + + def test_returns_original_when_no_memories(self): + messages = [{"role": "user", "content": "hi"}] + result = build_context_messages(messages, []) + assert result is messages # same object, not a copy + + def test_respects_max_memories(self): + messages = [{"role": "user", "content": "x"}] + hits = self._make_hits("A", "B", "C", "D", "E") + result = build_context_messages(messages, hits, max_memories=2) + system_text = result[0]["content"] + assert "A" in system_text + assert "B" in system_text + assert "C" not in system_text + + def test_custom_header(self): + messages = [{"role": "user", "content": "x"}] + hits = self._make_hits("Foo") + result = build_context_messages(messages, hits, header="My custom header:") + assert "My custom header:" in result[0]["content"] + + def test_does_not_mutate_original_messages(self): + original = [{"role": "user", "content": "q"}] + build_context_messages(original, self._make_hits("X")) + assert len(original) == 1 # unchanged + + def test_score_appears_in_output(self): + messages = [{"role": "user", "content": "q"}] + hits = [{"memory": {"content": "tip"}, "score": 0.87}] + result = build_context_messages(messages, hits) + assert "0.87" in result[0]["content"] + + +# --------------------------------------------------------------------------- +# DakeraMemoryHook tests +# --------------------------------------------------------------------------- + + +class TestDakeraMemoryHookRecall: + def _make_mem_mock(self, hits=None): + m = MagicMock(spec=DakeraMemory) + m.search.return_value = hits or [] + m.store.return_value = {"id": "new-mem"} + return m + + def test_attach_registers_handlers(self): + client = MagicMock() + hook = DakeraMemoryHook(self._make_mem_mock()) + hook.attach(client) + assert client.on.call_count == 2 + calls = {c.args[0] for c in client.on.call_args_list} + assert "completion:kwargs" in calls + assert "completion:response" in calls + + def test_detach_removes_handlers(self): + client = MagicMock() + hook = DakeraMemoryHook(self._make_mem_mock()) + hook.attach(client) + hook.detach(client) + assert client.off.call_count == 2 + + def test_on_completion_kwargs_injects_memory(self): + hits = [{"memory": {"content": "User likes Python"}, "score": 0.95}] + mem_mock = self._make_mem_mock(hits=hits) + hook = DakeraMemoryHook(mem_mock, top_k=3) + + messages = [{"role": "user", "content": "Which language should I use?"}] + kwargs = {"messages": messages} + hook._on_completion_kwargs(**kwargs) + + mem_mock.search.assert_called_once_with( + "Which language should I use?", top_k=3 + ) + # Messages list is mutated in-place + assert messages[0]["role"] == "system" # memory block prepended + assert any(m["role"] == "user" for m in messages) + + def test_on_completion_kwargs_skips_when_no_user_message(self): + mem_mock = self._make_mem_mock() + hook = DakeraMemoryHook(mem_mock) + hook._on_completion_kwargs(messages=[{"role": "system", "content": "x"}]) + mem_mock.search.assert_not_called() + + def test_on_completion_kwargs_skips_when_no_hits(self): + mem_mock = self._make_mem_mock(hits=[]) + hook = DakeraMemoryHook(mem_mock) + messages = [{"role": "user", "content": "hello"}] + hook._on_completion_kwargs(messages=messages) + assert messages[0]["role"] == "user" # untouched + + def test_on_completion_kwargs_handles_memory_error_gracefully(self): + mem_mock = self._make_mem_mock() + mem_mock.search.side_effect = RuntimeError("timeout") + hook = DakeraMemoryHook(mem_mock) + messages = [{"role": "user", "content": "q"}] + # Should not raise + hook._on_completion_kwargs(messages=messages) + assert messages[0]["role"] == "user" + + def test_on_completion_kwargs_disabled_when_auto_recall_false(self): + mem_mock = self._make_mem_mock(hits=[{"memory": {"content": "x"}, "score": 1.0}]) + hook = DakeraMemoryHook(mem_mock, auto_recall=False) + messages = [{"role": "user", "content": "q"}] + hook._on_completion_kwargs(messages=messages) + mem_mock.search.assert_not_called() + + +class TestDakeraMemoryHookStore: + def _make_response(self, content: str) -> MagicMock: + resp = MagicMock() + resp.choices = [MagicMock()] + resp.choices[0].message.content = content + return resp + + def _make_mem_mock(self): + m = MagicMock(spec=DakeraMemory) + m.store.return_value = {"id": "stored"} + return m + + def test_on_completion_response_stores_reply(self): + mem_mock = self._make_mem_mock() + hook = DakeraMemoryHook(mem_mock, store_importance=0.7, store_tags=["test"]) + hook._on_completion_response(self._make_response("Great answer!")) + mem_mock.store.assert_called_once_with( + "Great answer!", importance=0.7, tags=["test"] + ) + + def test_on_completion_response_skips_empty_content(self): + mem_mock = self._make_mem_mock() + hook = DakeraMemoryHook(mem_mock) + hook._on_completion_response(self._make_response("")) + mem_mock.store.assert_not_called() + + def test_on_completion_response_handles_error_gracefully(self): + mem_mock = self._make_mem_mock() + mem_mock.store.side_effect = RuntimeError("network error") + hook = DakeraMemoryHook(mem_mock) + # Should not raise + hook._on_completion_response(self._make_response("content")) + + def test_on_completion_response_disabled_when_auto_store_false(self): + mem_mock = self._make_mem_mock() + hook = DakeraMemoryHook(mem_mock, auto_store=False) + hook._on_completion_response(self._make_response("content")) + mem_mock.store.assert_not_called() + + +class TestDakeraMemoryHookManual: + def test_recall_into_messages_uses_last_user_message(self): + mem_mock = MagicMock(spec=DakeraMemory) + mem_mock.search.return_value = [] + hook = DakeraMemoryHook(mem_mock) + messages = [ + {"role": "system", "content": "sys"}, + {"role": "user", "content": "first"}, + {"role": "assistant", "content": "resp"}, + {"role": "user", "content": "second"}, + ] + hook.recall_into_messages(messages) + mem_mock.search.assert_called_once_with("second", top_k=5) + + def test_recall_into_messages_accepts_explicit_query(self): + mem_mock = MagicMock(spec=DakeraMemory) + mem_mock.search.return_value = [] + hook = DakeraMemoryHook(mem_mock) + messages = [{"role": "user", "content": "ignored"}] + hook.recall_into_messages(messages, query="explicit query") + mem_mock.search.assert_called_once_with("explicit query", top_k=5) + + def test_manual_store_delegates_to_memory(self): + mem_mock = MagicMock(spec=DakeraMemory) + mem_mock.store.return_value = {"id": "m1"} + hook = DakeraMemoryHook(mem_mock, store_importance=0.5, store_tags=["default"]) + result = hook.store("a fact") + mem_mock.store.assert_called_once_with("a fact", importance=0.5, tags=["default"]) + assert result == {"id": "m1"} + + def test_manual_store_allows_overriding_importance_and_tags(self): + mem_mock = MagicMock(spec=DakeraMemory) + mem_mock.store.return_value = {} + hook = DakeraMemoryHook(mem_mock, store_importance=0.5, store_tags=["default"]) + hook.store("fact", importance=0.99, tags=["custom"]) + mem_mock.store.assert_called_once_with("fact", importance=0.99, tags=["custom"]) diff --git a/mkdocs.yml b/mkdocs.yml index 8487fdac3..2a6bc0888 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -227,6 +227,7 @@ nav: - "Action Items Extraction": 'examples/action_items.md' - "Contact Information Extraction": 'examples/extract_contact_info.md' - "Knowledge Graph Building": 'examples/building_knowledge_graphs.md' + - "Persistent Memory with Dakera": 'examples/persistent_memory_dakera.md' - "Tracing with Langfuse": 'examples/tracing_with_langfuse.md' - "Multiple Classification Tasks": 'examples/multiple_classification.md' - "Pandas DataFrame Integration": 'examples/pandas_df.md'