holetron-lab/fleet-memory
1
0
Fork 0
Code Releases Packages Activity Actions

feat: add Pydantic AI integration for persistent agent memory (#441)

Browse source 
* feat: add Pydantic AI integration for persistent agent memory

Adds hindsight-pydantic-ai package providing Hindsight-backed memory
tools for Pydantic AI agents. Since Pydantic AI is async-native, tools
use the hindsight-client async API directly (no thread-pool compat layer).

- create_hindsight_tools(): factory returning retain/recall/reflect Tool instances
- memory_instructions(): auto-injects relevant memories via Agent instructions
- Global configure()/get_config()/reset_config() following existing integration pattern

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* doc: add README for Pydantic AI integration

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Ben 2026-03-02 09:14:00 -05:00 committed by GitHub
parent ab70da1ead
commit cab5a40f3a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 2432 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

191
hindsight-integrations/pydantic-ai/README.md Normal file
View file

@ -0,0 +1,191 @@
# hindsight-pydantic-ai
Persistent memory tools for Pydantic AI agents via Hindsight. Give your agents long-term memory with retain, recall, and reflect — all async-native with no thread-pool hacks.
## Features
- **Async-Native Tools** - Uses Pydantic AI's async tool interface directly (`aretain`, `arecall`, `areflect`)
- **Memory Instructions** - Auto-inject relevant memories into every agent run via `instructions=[...]`
- **Three Memory Tools** - Retain (store), Recall (search), Reflect (synthesize) — include any combination
- **Simple Configuration** - Configure once globally, or pass a client directly
- **Lightweight** - Depends on `pydantic-ai-slim` to avoid pulling in all model providers
## Installation
```bash
pip install hindsight-pydantic-ai
```
## Quick Start
```python
from hindsight_client import Hindsight
from hindsight_pydantic_ai import create_hindsight_tools, memory_instructions
from pydantic_ai import Agent
client = Hindsight(base_url="http://localhost:8888")
agent = Agent(
"openai:gpt-4o",
tools=create_hindsight_tools(client=client, bank_id="user-123"),
instructions=[memory_instructions(client=client, bank_id="user-123")],
)
result = await agent.run("What do you remember about my preferences?")
print(result.output)
```
The agent now has three tools it can call:
- **`hindsight_retain`** — Store information to long-term memory
- **`hindsight_recall`** — Search long-term memory for relevant facts
- **`hindsight_reflect`** — Synthesize a reasoned answer from memories
The `memory_instructions` callable automatically recalls relevant memories and injects them into the system prompt on every run.
## Tools Only (No Auto-Injection)
If you want the agent to decide when to use memory (rather than always injecting context):
```python
agent = Agent(
"openai:gpt-4o",
tools=create_hindsight_tools(client=client, bank_id="user-123"),
)
```
## Instructions Only (No Tools)
If you just want memories auto-injected without giving the agent explicit memory tools:
```python
agent = Agent(
"openai:gpt-4o",
instructions=[memory_instructions(client=client, bank_id="user-123")],
)
```
## Selecting Tools
Include only the tools you need:
```python
tools = create_hindsight_tools(
client=client,
bank_id="user-123",
include_retain=True,
include_recall=True,
include_reflect=False, # Omit reflect
)
```
## Global Configuration
Instead of passing a client to every call, configure once:
```python
from hindsight_pydantic_ai import configure, create_hindsight_tools
configure(
hindsight_api_url="http://localhost:8888",
api_key="your-api-key", # Or set HINDSIGHT_API_KEY env var
budget="mid", # Recall budget: low/mid/high
max_tokens=4096, # Max tokens for recall results
tags=["env:prod"], # Tags for stored memories
recall_tags=["scope:global"], # Tags to filter recall
recall_tags_match="any", # Tag match mode: any/all/any_strict/all_strict
)
# Now create tools without passing client — uses global config
tools = create_hindsight_tools(bank_id="user-123")
```
## Per-Tool Overrides
Constructor arguments override global configuration:
```python
tools = create_hindsight_tools(
bank_id="user-123",
budget="high", # Override global budget
max_tokens=8192, # Override global max_tokens
tags=["session:abc"], # Override global tags
)
```
## Memory Instructions Options
Customize what memories get injected and how:
```python
instructions_fn = memory_instructions(
client=client,
bank_id="user-123",
query="relevant context about the user", # What to search for
budget="low", # Keep it fast
max_results=5, # Limit injected memories
max_tokens=4096, # Max recall tokens
prefix="Relevant memories:\n", # Text before the memory list
tags=["scope:global"], # Filter by tags
tags_match="any", # Tag match mode
)
```
## Configuration Reference
### `create_hindsight_tools()`
| Parameter | Default | Description |
|---|---|---|
| `bank_id` | *required* | Hindsight memory bank ID |
| `client` | `None` | Pre-configured Hindsight client |
| `hindsight_api_url` | `None` | API URL (used if no client provided) |
| `api_key` | `None` | API key (used if no client provided) |
| `budget` | `"mid"` | Recall/reflect budget level (low/mid/high) |
| `max_tokens` | `4096` | Maximum tokens for recall results |
| `tags` | `None` | Tags applied when storing memories |
| `recall_tags` | `None` | Tags to filter when searching |
| `recall_tags_match` | `"any"` | Tag matching mode |
| `include_retain` | `True` | Include the retain (store) tool |
| `include_recall` | `True` | Include the recall (search) tool |
| `include_reflect` | `True` | Include the reflect (synthesize) tool |
### `memory_instructions()`
| Parameter | Default | Description |
|---|---|---|
| `bank_id` | *required* | Hindsight memory bank ID |
| `client` | `None` | Pre-configured Hindsight client |
| `hindsight_api_url` | `None` | API URL (used if no client provided) |
| `api_key` | `None` | API key (used if no client provided) |
| `query` | `"relevant context about the user"` | Recall query for memory injection |
| `budget` | `"low"` | Recall budget level |
| `max_results` | `5` | Maximum memories to inject |
| `max_tokens` | `4096` | Maximum tokens for recall results |
| `prefix` | `"Relevant memories:\n"` | Text prepended before memory list |
| `tags` | `None` | Tags to filter recall results |
| `tags_match` | `"any"` | Tag matching mode |
### `configure()`
| Parameter | Default | Description |
|---|---|---|
| `hindsight_api_url` | Production API | Hindsight API URL |
| `api_key` | `HINDSIGHT_API_KEY` env | API key for authentication |
| `budget` | `"mid"` | Default recall budget level |
| `max_tokens` | `4096` | Default max tokens for recall |
| `tags` | `None` | Default tags for retain operations |
| `recall_tags` | `None` | Default tags to filter recall |
| `recall_tags_match` | `"any"` | Default tag matching mode |
| `verbose` | `False` | Enable verbose logging |
## Requirements
- Python >= 3.10
- pydantic-ai-slim >= 1.0.0
- hindsight-client >= 0.4.0
- A running Hindsight API server
## License
MIT

42
hindsight-integrations/pydantic-ai/hindsight_pydantic_ai/__init__.py Normal file
View file

@ -0,0 +1,42 @@
"""Hindsight-Pydantic AI: Persistent memory tools for AI agents.
Provides Hindsight-backed tools and instructions for Pydantic AI agents,
giving them long-term memory across runs.
Basic usage::
from hindsight_client import Hindsight
from hindsight_pydantic_ai import create_hindsight_tools, memory_instructions
from pydantic_ai import Agent
client = Hindsight(base_url="http://localhost:8888")
agent = Agent(
"openai:gpt-4o",
tools=create_hindsight_tools(client=client, bank_id="user-123"),
instructions=[memory_instructions(client=client, bank_id="user-123")],
)
result = await agent.run("What do you remember about my preferences?")
"""
from .config import (
HindsightPydanticAIConfig,
configure,
get_config,
reset_config,
)
from .errors import HindsightError
from .tools import create_hindsight_tools, memory_instructions
__version__ = "0.1.0"
__all__ = [
"configure",
"get_config",
"reset_config",
"HindsightPydanticAIConfig",
"HindsightError",
"create_hindsight_tools",
"memory_instructions",
]

92
hindsight-integrations/pydantic-ai/hindsight_pydantic_ai/config.py Normal file
View file

@ -0,0 +1,92 @@
"""Global configuration for Hindsight-Pydantic AI integration."""
from __future__ import annotations
import os
from dataclasses import dataclass
DEFAULT_HINDSIGHT_API_URL = "https://api.hindsight.vectorize.io"
HINDSIGHT_API_KEY_ENV = "HINDSIGHT_API_KEY"
@dataclass
class HindsightPydanticAIConfig:
"""Connection and default settings for the Pydantic AI integration.
Attributes:
hindsight_api_url: URL of the Hindsight API server.
api_key: API key for Hindsight authentication.
budget: Default recall budget level (low/mid/high).
max_tokens: Default maximum tokens for recall results.
tags: Default tags applied when storing memories.
recall_tags: Default tags to filter when searching memories.
recall_tags_match: Tag matching mode (any/all/any_strict/all_strict).
verbose: Enable verbose logging.
"""
hindsight_api_url: str = DEFAULT_HINDSIGHT_API_URL
api_key: str | None = None
budget: str = "mid"
max_tokens: int = 4096
tags: list[str] | None = None
recall_tags: list[str] | None = None
recall_tags_match: str = "any"
verbose: bool = False
_global_config: HindsightPydanticAIConfig | None = None
def configure(
hindsight_api_url: str | None = None,
api_key: str | None = None,
budget: str = "mid",
max_tokens: int = 4096,
tags: list[str] | None = None,
recall_tags: list[str] | None = None,
recall_tags_match: str = "any",
verbose: bool = False,
) -> HindsightPydanticAIConfig:
"""Configure Hindsight connection and default settings.
Args:
hindsight_api_url: Hindsight API URL (default: production).
api_key: API key. Falls back to HINDSIGHT_API_KEY env var.
budget: Default recall budget (low/mid/high).
max_tokens: Default max tokens for recall.
tags: Default tags for retain operations.
recall_tags: Default tags to filter recall/search.
recall_tags_match: Tag matching mode.
verbose: Enable verbose logging.
Returns:
The configured HindsightPydanticAIConfig.
"""
global _global_config
resolved_url = hindsight_api_url or DEFAULT_HINDSIGHT_API_URL
resolved_key = api_key or os.environ.get(HINDSIGHT_API_KEY_ENV)
_global_config = HindsightPydanticAIConfig(
hindsight_api_url=resolved_url,
api_key=resolved_key,
budget=budget,
max_tokens=max_tokens,
tags=tags,
recall_tags=recall_tags,
recall_tags_match=recall_tags_match,
verbose=verbose,
)
return _global_config
def get_config() -> HindsightPydanticAIConfig | None:
"""Get the current global configuration."""
return _global_config
def reset_config() -> None:
"""Reset global configuration to None."""
global _global_config
_global_config = None

7
hindsight-integrations/pydantic-ai/hindsight_pydantic_ai/errors.py Normal file
View file

@ -0,0 +1,7 @@
"""Hindsight-Pydantic AI error types."""
class HindsightError(Exception):
"""Exception raised when a Hindsight memory operation fails."""
pass

243
hindsight-integrations/pydantic-ai/hindsight_pydantic_ai/tools.py Normal file
View file

@ -0,0 +1,243 @@
"""Pydantic AI tools for Hindsight memory operations.
Provides factory functions that create Pydantic AI ``Tool`` instances
backed by Hindsight's retain/recall/reflect APIs.
"""
from __future__ import annotations
import logging
from collections.abc import Awaitable, Callable
from typing import Any
from hindsight_client import Hindsight
from pydantic_ai import RunContext, Tool
from .config import get_config
from .errors import HindsightError
logger = logging.getLogger(__name__)
def _resolve_client(
client: Hindsight | None,
hindsight_api_url: str | None,
api_key: str | None,
) -> Hindsight:
"""Resolve a Hindsight client from explicit args or global config."""
if client is not None:
return client
config = get_config()
url = hindsight_api_url or (config.hindsight_api_url if config else None)
key = api_key or (config.api_key if config else None)
if url is None:
raise HindsightError(
"No Hindsight API URL configured. "
"Pass client= or hindsight_api_url=, or call configure() first."
)
kwargs: dict[str, Any] = {"base_url": url, "timeout": 30.0}
if key:
kwargs["api_key"] = key
return Hindsight(**kwargs)
def create_hindsight_tools(
*,
bank_id: str,
client: Hindsight | None = None,
hindsight_api_url: str | None = None,
api_key: str | None = None,
budget: str = "mid",
max_tokens: int = 4096,
tags: list[str] | None = None,
recall_tags: list[str] | None = None,
recall_tags_match: str = "any",
include_retain: bool = True,
include_recall: bool = True,
include_reflect: bool = True,
) -> list[Tool]:
"""Create Hindsight memory tools for a Pydantic AI agent.
Returns a list of ``Tool`` instances that can be passed directly to
``Agent(tools=...)``. Each tool is an async closure that captures
the Hindsight client — no ``RunContext`` or deps modification needed.
Args:
bank_id: The Hindsight memory bank to operate on.
client: Pre-configured Hindsight client (preferred).
hindsight_api_url: API URL (used if no client provided).
api_key: API key (used if no client provided).
budget: Recall/reflect budget level (low/mid/high).
max_tokens: Maximum tokens for recall results.
tags: Tags applied when storing memories via retain.
recall_tags: Tags to filter when searching memories.
recall_tags_match: Tag matching mode (any/all/any_strict/all_strict).
include_retain: Include the retain (store) tool.
include_recall: Include the recall (search) tool.
include_reflect: Include the reflect (synthesize) tool.
Returns:
List of Pydantic AI Tool instances.
Raises:
HindsightError: If no client or API URL can be resolved.
"""
resolved_client = _resolve_client(client, hindsight_api_url, api_key)
# Resolve defaults from global config
config = get_config()
effective_tags = tags if tags is not None else (config.tags if config else None)
effective_recall_tags = (
recall_tags if recall_tags is not None else (config.recall_tags if config else None)
)
effective_recall_tags_match = recall_tags_match or (config.recall_tags_match if config else "any")
effective_budget = budget or (config.budget if config else "mid")
effective_max_tokens = max_tokens or (config.max_tokens if config else 4096)
tools: list[Tool] = []
if include_retain:
async def hindsight_retain(content: str) -> str:
"""Store information to long-term memory for later retrieval.
Use this to save important facts, user preferences, decisions,
or any information that should be remembered across conversations.
"""
try:
retain_kwargs: dict[str, Any] = {"bank_id": bank_id, "content": content}
if effective_tags:
retain_kwargs["tags"] = effective_tags
await resolved_client.aretain(**retain_kwargs)
return "Memory stored successfully."
except Exception as e:
logger.error(f"Retain failed: {e}")
raise HindsightError(f"Retain failed: {e}") from e
tools.append(Tool(hindsight_retain, takes_ctx=False))
if include_recall:
async def hindsight_recall(query: str) -> str:
"""Search long-term memory for relevant information.
Use this to find previously stored facts, preferences, or context.
Returns a numbered list of matching memories.
"""
try:
recall_kwargs: dict[str, Any] = {
"bank_id": bank_id,
"query": query,
"budget": effective_budget,
"max_tokens": effective_max_tokens,
}
if effective_recall_tags:
recall_kwargs["tags"] = effective_recall_tags
recall_kwargs["tags_match"] = effective_recall_tags_match
response = await resolved_client.arecall(**recall_kwargs)
if not response.results:
return "No relevant memories found."
lines = []
for i, result in enumerate(response.results, 1):
lines.append(f"{i}. {result.text}")
return "\n".join(lines)
except Exception as e:
logger.error(f"Recall failed: {e}")
raise HindsightError(f"Recall failed: {e}") from e
tools.append(Tool(hindsight_recall, takes_ctx=False))
if include_reflect:
async def hindsight_reflect(query: str) -> str:
"""Synthesize a thoughtful answer from long-term memories.
Use this when you need a coherent summary or reasoned response
about what you know, rather than raw memory facts.
"""
try:
reflect_kwargs: dict[str, Any] = {
"bank_id": bank_id,
"query": query,
"budget": effective_budget,
}
response = await resolved_client.areflect(**reflect_kwargs)
return response.text or "No relevant memories found."
except Exception as e:
logger.error(f"Reflect failed: {e}")
raise HindsightError(f"Reflect failed: {e}") from e
tools.append(Tool(hindsight_reflect, takes_ctx=False))
return tools
def memory_instructions(
*,
bank_id: str,
client: Hindsight | None = None,
hindsight_api_url: str | None = None,
api_key: str | None = None,
query: str = "relevant context about the user",
budget: str = "low",
max_results: int = 5,
max_tokens: int = 4096,
prefix: str = "Relevant memories:\n",
tags: list[str] | None = None,
tags_match: str = "any",
) -> Callable[[RunContext[Any]], Awaitable[str]]:
"""Create an instructions function that auto-injects relevant memories.
Returns an async callable suitable for use with Pydantic AI's
``instructions`` parameter. Because instructions are re-evaluated
on every run, memories stay fresh even when ``message_history``
is reused.
Args:
bank_id: The Hindsight memory bank to recall from.
client: Pre-configured Hindsight client (preferred).
hindsight_api_url: API URL (used if no client provided).
api_key: API key (used if no client provided).
query: The recall query to find relevant memories.
budget: Recall budget level (low/mid/high).
max_results: Maximum number of memories to include.
max_tokens: Maximum tokens for recall results.
prefix: Text prepended before the memory list.
tags: Tags to filter recall results.
tags_match: Tag matching mode (any/all/any_strict/all_strict).
Returns:
An async function compatible with ``Agent(instructions=[...])``.
Raises:
HindsightError: If no client or API URL can be resolved.
"""
resolved_client = _resolve_client(client, hindsight_api_url, api_key)
async def _instructions(ctx: RunContext[Any]) -> str:
try:
recall_kwargs: dict[str, Any] = {
"bank_id": bank_id,
"query": query,
"budget": budget,
"max_tokens": max_tokens,
}
if tags:
recall_kwargs["tags"] = tags
recall_kwargs["tags_match"] = tags_match
response = await resolved_client.arecall(**recall_kwargs)
results = response.results[:max_results] if response.results else []
if not results:
return ""
lines = [prefix]
for i, result in enumerate(results, 1):
lines.append(f"{i}. {result.text}")
return "\n".join(lines)
except Exception:
# Silently return empty — instructions failures shouldn't block the agent
return ""
return _instructions

60
hindsight-integrations/pydantic-ai/pyproject.toml Normal file
View file

@ -0,0 +1,60 @@
[project]
name = "hindsight-pydantic-ai"
version = "0.1.0"
description = "Pydantic AI integration for Hindsight - persistent memory tools for AI agents"
requires-python = ">=3.10"
license = { text = "MIT" }
authors = [
{ name = "Vectorize", email = "support@vectorize.io" }
]
keywords = [
"ai",
"memory",
"pydantic-ai",
"agents",
"hindsight",
]
classifiers = [
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Topic :: Scientific/Engineering :: Artificial Intelligence",
]
dependencies = [
"pydantic-ai-slim>=1.0.0",
"hindsight-client>=0.4.0",
]
[project.optional-dependencies]
dev = [
"pytest>=7.0.0",
"pytest-mock>=3.10.0",
"pytest-asyncio>=0.23.0",
]
[project.urls]
Homepage = "https://github.com/vectorize-io/hindsight"
Documentation = "https://github.com/vectorize-io/hindsight/tree/main/hindsight-integrations/pydantic-ai"
Repository = "https://github.com/vectorize-io/hindsight"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["hindsight_pydantic_ai"]
[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"
[dependency-groups]
dev = [
"pytest>=9.0.2",
"pytest-asyncio>=0.23.0",
]

0
hindsight-integrations/pydantic-ai/tests/__init__.py Normal file
View file

76
hindsight-integrations/pydantic-ai/tests/test_config.py Normal file
View file

@ -0,0 +1,76 @@
"""Unit tests for hindsight_pydantic_ai configuration."""
import os
from unittest.mock import patch
from hindsight_pydantic_ai import configure, get_config, reset_config
from hindsight_pydantic_ai.config import (
DEFAULT_HINDSIGHT_API_URL,
HINDSIGHT_API_KEY_ENV,
)
class TestDefaults:
def test_default_api_url(self):
assert DEFAULT_HINDSIGHT_API_URL == "https://api.hindsight.vectorize.io"
def test_env_var_name(self):
assert HINDSIGHT_API_KEY_ENV == "HINDSIGHT_API_KEY"
class TestConfigure:
def setup_method(self):
reset_config()
def teardown_method(self):
reset_config()
def test_configure_with_no_arguments(self):
config = configure()
assert config.hindsight_api_url == DEFAULT_HINDSIGHT_API_URL
assert config.budget == "mid"
assert config.max_tokens == 4096
assert config.verbose is False
def test_configure_reads_api_key_from_env(self):
with patch.dict(os.environ, {HINDSIGHT_API_KEY_ENV: "test-key"}):
config = configure()
assert config.api_key == "test-key"
def test_configure_explicit_overrides_env(self):
with patch.dict(os.environ, {HINDSIGHT_API_KEY_ENV: "env-key"}):
config = configure(api_key="explicit-key")
assert config.api_key == "explicit-key"
def test_configure_all_options(self):
config = configure(
hindsight_api_url="http://custom:8888",
api_key="my-key",
budget="high",
max_tokens=2048,
tags=["env:test"],
recall_tags=["scope:global"],
recall_tags_match="all",
verbose=True,
)
assert config.hindsight_api_url == "http://custom:8888"
assert config.api_key == "my-key"
assert config.budget == "high"
assert config.max_tokens == 2048
assert config.tags == ["env:test"]
assert config.recall_tags == ["scope:global"]
assert config.recall_tags_match == "all"
assert config.verbose is True
def test_get_config_returns_none_without_configure(self):
assert get_config() is None
def test_get_config_returns_config_after_configure(self):
configure()
assert get_config() is not None
def test_reset_config(self):
configure()
assert get_config() is not None
reset_config()
assert get_config() is None

468
hindsight-integrations/pydantic-ai/tests/test_tools.py Normal file
View file

@ -0,0 +1,468 @@
"""Unit tests for Hindsight Pydantic AI tools."""
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from hindsight_pydantic_ai import (
configure,
create_hindsight_tools,
memory_instructions,
reset_config,
)
from hindsight_pydantic_ai.errors import HindsightError
def _mock_client():
"""Create a mock Hindsight client with async methods."""
client = MagicMock()
client.aretain = AsyncMock()
client.arecall = AsyncMock()
client.areflect = AsyncMock()
return client
def _mock_recall_result(text: str):
"""Create a mock RecallResult."""
result = MagicMock()
result.text = text
return result
def _mock_recall_response(texts: list[str]):
"""Create a mock RecallResponse with results."""
response = MagicMock()
response.results = [_mock_recall_result(t) for t in texts]
return response
def _mock_reflect_response(text: str):
"""Create a mock ReflectResponse."""
response = MagicMock()
response.text = text
return response
def _mock_retain_response():
"""Create a mock RetainResponse."""
response = MagicMock()
response.success = True
return response
class TestCreateHindsightTools:
def setup_method(self):
reset_config()
def teardown_method(self):
reset_config()
def test_returns_three_tools_by_default(self):
client = _mock_client()
tools = create_hindsight_tools(bank_id="test", client=client)
assert len(tools) == 3
def test_include_retain_only(self):
client = _mock_client()
tools = create_hindsight_tools(
bank_id="test",
client=client,
include_retain=True,
include_recall=False,
include_reflect=False,
)
assert len(tools) == 1
assert tools[0].name == "hindsight_retain"
def test_include_recall_only(self):
client = _mock_client()
tools = create_hindsight_tools(
bank_id="test",
client=client,
include_retain=False,
include_recall=True,
include_reflect=False,
)
assert len(tools) == 1
assert tools[0].name == "hindsight_recall"
def test_include_reflect_only(self):
client = _mock_client()
tools = create_hindsight_tools(
bank_id="test",
client=client,
include_retain=False,
include_recall=False,
include_reflect=True,
)
assert len(tools) == 1
assert tools[0].name == "hindsight_reflect"
def test_no_tools_when_all_excluded(self):
client = _mock_client()
tools = create_hindsight_tools(
bank_id="test",
client=client,
include_retain=False,
include_recall=False,
include_reflect=False,
)
assert len(tools) == 0
def test_raises_without_client_or_config(self):
with pytest.raises(HindsightError, match="No Hindsight API URL"):
create_hindsight_tools(bank_id="test")
def test_falls_back_to_global_config(self):
configure(hindsight_api_url="http://localhost:8888")
with patch("hindsight_pydantic_ai.tools.Hindsight") as mock_cls:
mock_cls.return_value = _mock_client()
tools = create_hindsight_tools(bank_id="test")
assert len(tools) == 3
mock_cls.assert_called_once_with(
base_url="http://localhost:8888", timeout=30.0
)
def test_explicit_url_overrides_config(self):
configure(hindsight_api_url="http://config:8888")
with patch("hindsight_pydantic_ai.tools.Hindsight") as mock_cls:
mock_cls.return_value = _mock_client()
create_hindsight_tools(
bank_id="test", hindsight_api_url="http://explicit:9999"
)
mock_cls.assert_called_once_with(
base_url="http://explicit:9999", timeout=30.0
)
class TestRetainTool:
@pytest.mark.asyncio
async def test_retain_stores_memory(self):
client = _mock_client()
client.aretain.return_value = _mock_retain_response()
tools = create_hindsight_tools(
bank_id="test-bank",
client=client,
include_recall=False,
include_reflect=False,
)
tool_fn = tools[0].function
result = await tool_fn("The user likes Python")
assert result == "Memory stored successfully."
client.aretain.assert_called_once_with(bank_id="test-bank", content="The user likes Python")
@pytest.mark.asyncio
async def test_retain_passes_tags(self):
client = _mock_client()
client.aretain.return_value = _mock_retain_response()
tools = create_hindsight_tools(
bank_id="test-bank",
client=client,
tags=["source:chat"],
include_recall=False,
include_reflect=False,
)
tool_fn = tools[0].function
await tool_fn("some content")
call_kwargs = client.aretain.call_args[1]
assert call_kwargs["tags"] == ["source:chat"]
@pytest.mark.asyncio
async def test_retain_raises_hindsight_error(self):
client = _mock_client()
client.aretain.side_effect = RuntimeError("connection refused")
tools = create_hindsight_tools(
bank_id="test",
client=client,
include_recall=False,
include_reflect=False,
)
tool_fn = tools[0].function
with pytest.raises(HindsightError, match="Retain failed"):
await tool_fn("content")
class TestRecallTool:
@pytest.mark.asyncio
async def test_recall_returns_numbered_results(self):
client = _mock_client()
client.arecall.return_value = _mock_recall_response(
["User likes Python", "User is in NYC"]
)
tools = create_hindsight_tools(
bank_id="test-bank",
client=client,
include_retain=False,
include_reflect=False,
)
tool_fn = tools[0].function
result = await tool_fn("user preferences")
assert "1. User likes Python" in result
assert "2. User is in NYC" in result
client.arecall.assert_called_once()
@pytest.mark.asyncio
async def test_recall_empty_results(self):
client = _mock_client()
client.arecall.return_value = _mock_recall_response([])
tools = create_hindsight_tools(
bank_id="test",
client=client,
include_retain=False,
include_reflect=False,
)
tool_fn = tools[0].function
result = await tool_fn("anything")
assert result == "No relevant memories found."
@pytest.mark.asyncio
async def test_recall_passes_budget_and_max_tokens(self):
client = _mock_client()
client.arecall.return_value = _mock_recall_response(["fact"])
tools = create_hindsight_tools(
bank_id="test",
client=client,
budget="high",
max_tokens=2048,
include_retain=False,
include_reflect=False,
)
tool_fn = tools[0].function
await tool_fn("query")
call_kwargs = client.arecall.call_args[1]
assert call_kwargs["budget"] == "high"
assert call_kwargs["max_tokens"] == 2048
@pytest.mark.asyncio
async def test_recall_passes_tags(self):
client = _mock_client()
client.arecall.return_value = _mock_recall_response(["fact"])
tools = create_hindsight_tools(
bank_id="test",
client=client,
recall_tags=["scope:user"],
recall_tags_match="all",
include_retain=False,
include_reflect=False,
)
tool_fn = tools[0].function
await tool_fn("query")
call_kwargs = client.arecall.call_args[1]
assert call_kwargs["tags"] == ["scope:user"]
assert call_kwargs["tags_match"] == "all"
@pytest.mark.asyncio
async def test_recall_raises_hindsight_error(self):
client = _mock_client()
client.arecall.side_effect = RuntimeError("timeout")
tools = create_hindsight_tools(
bank_id="test",
client=client,
include_retain=False,
include_reflect=False,
)
tool_fn = tools[0].function
with pytest.raises(HindsightError, match="Recall failed"):
await tool_fn("query")
class TestReflectTool:
@pytest.mark.asyncio
async def test_reflect_returns_text(self):
client = _mock_client()
client.areflect.return_value = _mock_reflect_response(
"The user is a Python developer who prefers functional patterns."
)
tools = create_hindsight_tools(
bank_id="test-bank",
client=client,
include_retain=False,
include_recall=False,
)
tool_fn = tools[0].function
result = await tool_fn("What do you know about the user?")
assert result == "The user is a Python developer who prefers functional patterns."
client.areflect.assert_called_once()
@pytest.mark.asyncio
async def test_reflect_empty_returns_fallback(self):
client = _mock_client()
client.areflect.return_value = _mock_reflect_response("")
tools = create_hindsight_tools(
bank_id="test",
client=client,
include_retain=False,
include_recall=False,
)
tool_fn = tools[0].function
result = await tool_fn("anything")
assert result == "No relevant memories found."
@pytest.mark.asyncio
async def test_reflect_passes_budget(self):
client = _mock_client()
client.areflect.return_value = _mock_reflect_response("answer")
tools = create_hindsight_tools(
bank_id="test",
client=client,
budget="high",
include_retain=False,
include_recall=False,
)
tool_fn = tools[0].function
await tool_fn("query")
call_kwargs = client.areflect.call_args[1]
assert call_kwargs["budget"] == "high"
@pytest.mark.asyncio
async def test_reflect_raises_hindsight_error(self):
client = _mock_client()
client.areflect.side_effect = RuntimeError("timeout")
tools = create_hindsight_tools(
bank_id="test",
client=client,
include_retain=False,
include_recall=False,
)
tool_fn = tools[0].function
with pytest.raises(HindsightError, match="Reflect failed"):
await tool_fn("query")
class TestMemoryInstructions:
@pytest.mark.asyncio
async def test_returns_formatted_memories(self):
client = _mock_client()
client.arecall.return_value = _mock_recall_response(
["Likes Python", "Lives in NYC", "Prefers dark mode"]
)
instructions_fn = memory_instructions(
bank_id="test-bank", client=client
)
# Instructions functions receive RunContext — mock it
mock_ctx = MagicMock()
result = await instructions_fn(mock_ctx)
assert "Relevant memories:" in result
assert "1. Likes Python" in result
assert "2. Lives in NYC" in result
assert "3. Prefers dark mode" in result
@pytest.mark.asyncio
async def test_respects_max_results(self):
client = _mock_client()
client.arecall.return_value = _mock_recall_response(
["fact1", "fact2", "fact3", "fact4", "fact5"]
)
instructions_fn = memory_instructions(
bank_id="test", client=client, max_results=2
)
mock_ctx = MagicMock()
result = await instructions_fn(mock_ctx)
assert "1. fact1" in result
assert "2. fact2" in result
assert "3." not in result
@pytest.mark.asyncio
async def test_custom_prefix(self):
client = _mock_client()
client.arecall.return_value = _mock_recall_response(["fact"])
instructions_fn = memory_instructions(
bank_id="test", client=client, prefix="Memory context:\n"
)
mock_ctx = MagicMock()
result = await instructions_fn(mock_ctx)
assert result.startswith("Memory context:")
@pytest.mark.asyncio
async def test_empty_results_returns_empty_string(self):
client = _mock_client()
client.arecall.return_value = _mock_recall_response([])
instructions_fn = memory_instructions(
bank_id="test", client=client
)
mock_ctx = MagicMock()
result = await instructions_fn(mock_ctx)
assert result == ""
@pytest.mark.asyncio
async def test_error_returns_empty_string(self):
client = _mock_client()
client.arecall.side_effect = RuntimeError("connection error")
instructions_fn = memory_instructions(
bank_id="test", client=client
)
mock_ctx = MagicMock()
result = await instructions_fn(mock_ctx)
assert result == ""
@pytest.mark.asyncio
async def test_passes_query_and_budget(self):
client = _mock_client()
client.arecall.return_value = _mock_recall_response(["fact"])
instructions_fn = memory_instructions(
bank_id="test",
client=client,
query="user preferences and context",
budget="high",
)
mock_ctx = MagicMock()
await instructions_fn(mock_ctx)
call_kwargs = client.arecall.call_args[1]
assert call_kwargs["query"] == "user preferences and context"
assert call_kwargs["budget"] == "high"
@pytest.mark.asyncio
async def test_passes_tags(self):
client = _mock_client()
client.arecall.return_value = _mock_recall_response(["fact"])
instructions_fn = memory_instructions(
bank_id="test",
client=client,
tags=["scope:user"],
tags_match="all",
)
mock_ctx = MagicMock()
await instructions_fn(mock_ctx)
call_kwargs = client.arecall.call_args[1]
assert call_kwargs["tags"] == ["scope:user"]
assert call_kwargs["tags_match"] == "all"
def test_raises_without_client_or_config(self):
reset_config()
with pytest.raises(HindsightError, match="No Hindsight API URL"):
memory_instructions(bank_id="test")

1253
hindsight-integrations/pydantic-ai/uv.lock Normal file
View file

File diff suppressed because it is too large Load diff