Compare commits

..
Author SHA1 Message Date
Nicolò Boschi 4017b3ad8e chore: regenerate client SDKs after Vertex AI support 2026-01-29 19:03:13 +01:00
Nicolò Boschi eef8f16ead fix: add index-strategy to root pyproject.toml for workspace-level uv resolution 2026-01-29 18:54:27 +01:00
Nicolò Boschi 26d773e8d1 fix: add uv index-strategy to resolve dependency conflicts with pytorch index
When using pytorch index for faster torch downloads in CI,
filelock dependency resolution was failing because pytorch index
only has older versions. Adding unsafe-best-match strategy allows
uv to search all configured indexes.

Also fix type checking warnings from ty.
2026-01-29 18:46:18 +01:00
Nicolò Boschi 42ff43a712 fix 2026-01-29 18:46:18 +01:00
Nicolò Boschi fea6f67ac5 feat: support vertex as llm provider 2026-01-29 18:46:18 +01:00
48 changed files with 2650 additions and 2291 deletions
+19 -7
View File
@@ -92,7 +92,8 @@ class RecallRequest(BaseModel):
query: str
types: list[str] | None = Field(
default=None,
description="List of fact types to recall: 'world', 'experience', 'observation'. Defaults to world and experience if not specified.",
description="List of fact types to recall: 'world', 'experience', 'observation'. Defaults to world and experience if not specified. "
"Note: 'opinion' is accepted but ignored (opinions are excluded from recall).",
)
budget: Budget = Budget.MID
max_tokens: int = 4096
@@ -503,6 +504,13 @@ class ReflectRequest(BaseModel):
)
class OpinionItem(BaseModel):
"""Model for an opinion with confidence score."""
text: str
confidence: float
class ReflectFact(BaseModel):
"""A fact used in think response."""
@@ -521,7 +529,7 @@ class ReflectFact(BaseModel):
id: str | None = None
text: str
type: str | None = None # fact type: world, experience, observation
type: str | None = None # fact type: world, experience, opinion
context: str | None = None
occurred_start: str | None = None
occurred_end: str | None = None
@@ -1404,10 +1412,9 @@ def create_app(
worker_id=worker_id,
executor=memory.execute_task,
poll_interval_ms=config.worker_poll_interval_ms,
batch_size=config.worker_batch_size,
max_retries=config.worker_max_retries,
tenant_extension=getattr(memory, "_tenant_extension", None),
max_slots=config.worker_max_slots,
consolidation_max_slots=config.worker_consolidation_max_slots,
)
poller_task = asyncio.create_task(poller.run())
logging.info(f"Worker poller started (worker_id={worker_id})")
@@ -1700,7 +1707,9 @@ def _register_routes(app: FastAPI):
description="Recall memory using semantic similarity and spreading activation.\n\n"
"The type parameter is optional and must be one of:\n"
"- `world`: General knowledge about people, places, events, and things that happen\n"
"- `experience`: Memories about experience, conversations, actions taken, and tasks performed",
"- `experience`: Memories about experience, conversations, actions taken, and tasks performed\n"
"- `opinion`: The bank's formed beliefs, perspectives, and viewpoints\n\n"
"Set `include_entities=true` to get entity observations alongside recall results.",
operation_id="recall_memories",
tags=["Memory"],
)
@@ -1714,8 +1723,10 @@ def _register_routes(app: FastAPI):
metrics = get_metrics_collector()
try:
# Default to world and experience if not specified (exclude observation)
# Default to world and experience if not specified (exclude observation and opinion)
# Filter out 'opinion' even if requested - opinions are excluded from recall
fact_types = request.types if request.types else list(VALID_RECALL_FACT_TYPES)
fact_types = [ft for ft in fact_types if ft != "opinion"]
# Parse query_timestamp if provided
question_date = None
@@ -1847,7 +1858,8 @@ def _register_routes(app: FastAPI):
"2. Retrieves world facts relevant to the query\n"
"3. Retrieves existing opinions (bank's perspectives)\n"
"4. Uses LLM to formulate a contextual answer\n"
"5. Returns plain text answer and the facts used",
"5. Extracts and stores any new opinions formed\n"
"6. Returns plain text answer, the facts used, and new opinions",
operation_id="reflect",
tags=["Memory"],
)
+5 -45
View File
@@ -29,26 +29,15 @@ logger = logging.getLogger(__name__)
# Default bank_id from environment variable
DEFAULT_BANK_ID = os.environ.get("HINDSIGHT_MCP_BANK_ID", "default")
# MCP authentication token (optional - if set, Bearer token auth is required)
MCP_AUTH_TOKEN = os.environ.get("HINDSIGHT_API_MCP_AUTH_TOKEN")
# Context variable to hold the current bank_id
_current_bank_id: ContextVar[str | None] = ContextVar("current_bank_id", default=None)
# Context variable to hold the current API key (for tenant auth propagation)
_current_api_key: ContextVar[str | None] = ContextVar("current_api_key", default=None)
def get_current_bank_id() -> str | None:
"""Get the current bank_id from context."""
return _current_bank_id.get()
def get_current_api_key() -> str | None:
"""Get the current API key from context."""
return _current_api_key.get()
def create_mcp_server(memory: MemoryEngine) -> FastMCP:
"""
Create and configure the Hindsight MCP server.
@@ -65,7 +54,6 @@ def create_mcp_server(memory: MemoryEngine) -> FastMCP:
# Configure and register tools using shared module
config = MCPToolsConfig(
bank_id_resolver=get_current_bank_id,
api_key_resolver=get_current_api_key, # Propagate API key for tenant auth
include_bank_id_param=True, # HTTP MCP supports multi-bank via parameter
tools=None, # All tools
retain_fire_and_forget=False, # HTTP MCP supports sync/async modes
@@ -77,11 +65,7 @@ def create_mcp_server(memory: MemoryEngine) -> FastMCP:
class MCPMiddleware:
"""ASGI middleware that handles authentication and extracts bank_id from header or path.
Authentication:
If HINDSIGHT_API_MCP_AUTH_TOKEN is set, all requests must include a valid
Authorization header with Bearer token or direct token matching the configured value.
"""ASGI middleware that extracts bank_id from header or path and sets context.
Bank ID can be provided via:
1. X-Bank-Id header (recommended for Claude Code)
@@ -90,7 +74,7 @@ class MCPMiddleware:
For Claude Code, configure with:
claude mcp add --transport http hindsight http://localhost:8888/mcp \\
--header "X-Bank-Id: my-bank" --header "Authorization: Bearer <token>"
--header "X-Bank-Id: my-bank"
"""
def __init__(self, app, memory: MemoryEngine):
@@ -114,22 +98,6 @@ class MCPMiddleware:
await self.mcp_app(scope, receive, send)
return
# Extract auth token from header (for tenant auth propagation)
auth_header = self._get_header(scope, "Authorization")
auth_token: str | None = None
if auth_header:
# Support both "Bearer <token>" and direct token
auth_token = auth_header[7:].strip() if auth_header.startswith("Bearer ") else auth_header.strip()
# Authenticate if MCP_AUTH_TOKEN is configured
if MCP_AUTH_TOKEN:
if not auth_token:
await self._send_error(send, 401, "Authorization header required")
return
if auth_token != MCP_AUTH_TOKEN:
await self._send_error(send, 401, "Invalid authentication token")
return
path = scope.get("path", "")
# Strip any mount prefix (e.g., /mcp) that FastAPI might not have stripped
@@ -164,10 +132,8 @@ class MCPMiddleware:
bank_id = DEFAULT_BANK_ID
logger.debug(f"Using default bank_id: {bank_id}")
# Set bank_id and api_key context
bank_id_token = _current_bank_id.set(bank_id)
# Store the auth token for tenant extension to validate
api_key_token = _current_api_key.set(auth_token) if auth_token else None
# Set bank_id context
token = _current_bank_id.set(bank_id)
try:
new_scope = scope.copy()
new_scope["path"] = new_path
@@ -186,9 +152,7 @@ class MCPMiddleware:
await self.mcp_app(new_scope, receive, send_wrapper)
finally:
_current_bank_id.reset(bank_id_token)
if api_key_token is not None:
_current_api_key.reset(api_key_token)
_current_bank_id.reset(token)
async def _send_error(self, send, status: int, message: str):
"""Send an error response."""
@@ -212,10 +176,6 @@ def create_mcp_app(memory: MemoryEngine):
"""
Create an ASGI app that handles MCP requests.
Authentication:
Set HINDSIGHT_API_MCP_AUTH_TOKEN to require Bearer token authentication.
If not set, MCP endpoint is open (for local development).
Bank ID can be provided via:
1. X-Bank-Id header: claude mcp add --transport http hindsight http://localhost:8888/mcp --header "X-Bank-Id: my-bank"
2. URL path: /mcp/{bank_id}/
+11 -10
View File
@@ -119,6 +119,7 @@ ENV_RETAIN_CHUNK_SIZE = "HINDSIGHT_API_RETAIN_CHUNK_SIZE"
ENV_RETAIN_EXTRACT_CAUSAL_LINKS = "HINDSIGHT_API_RETAIN_EXTRACT_CAUSAL_LINKS"
ENV_RETAIN_EXTRACTION_MODE = "HINDSIGHT_API_RETAIN_EXTRACTION_MODE"
ENV_RETAIN_CUSTOM_INSTRUCTIONS = "HINDSIGHT_API_RETAIN_CUSTOM_INSTRUCTIONS"
ENV_RETAIN_OBSERVATIONS_ASYNC = "HINDSIGHT_API_RETAIN_OBSERVATIONS_ASYNC"
# Observations settings (consolidated knowledge from facts)
ENV_ENABLE_OBSERVATIONS = "HINDSIGHT_API_ENABLE_OBSERVATIONS"
@@ -143,9 +144,8 @@ ENV_WORKER_ENABLED = "HINDSIGHT_API_WORKER_ENABLED"
ENV_WORKER_ID = "HINDSIGHT_API_WORKER_ID"
ENV_WORKER_POLL_INTERVAL_MS = "HINDSIGHT_API_WORKER_POLL_INTERVAL_MS"
ENV_WORKER_MAX_RETRIES = "HINDSIGHT_API_WORKER_MAX_RETRIES"
ENV_WORKER_BATCH_SIZE = "HINDSIGHT_API_WORKER_BATCH_SIZE"
ENV_WORKER_HTTP_PORT = "HINDSIGHT_API_WORKER_HTTP_PORT"
ENV_WORKER_MAX_SLOTS = "HINDSIGHT_API_WORKER_MAX_SLOTS"
ENV_WORKER_CONSOLIDATION_MAX_SLOTS = "HINDSIGHT_API_WORKER_CONSOLIDATION_MAX_SLOTS"
# Reflect agent settings
ENV_REFLECT_MAX_ITERATIONS = "HINDSIGHT_API_REFLECT_MAX_ITERATIONS"
@@ -210,6 +210,7 @@ DEFAULT_RETAIN_EXTRACT_CAUSAL_LINKS = True # Extract causal links between facts
DEFAULT_RETAIN_EXTRACTION_MODE = "concise" # Extraction mode: "concise", "verbose", or "custom"
RETAIN_EXTRACTION_MODES = ("concise", "verbose", "custom") # Allowed extraction modes
DEFAULT_RETAIN_CUSTOM_INSTRUCTIONS = None # Custom extraction guidelines (only used when mode="custom")
DEFAULT_RETAIN_OBSERVATIONS_ASYNC = False # Run observation generation async (after retain completes)
# Observations defaults (consolidated knowledge from facts)
DEFAULT_ENABLE_OBSERVATIONS = True # Observations enabled by default
@@ -230,9 +231,8 @@ DEFAULT_WORKER_ENABLED = True # API runs worker by default (standalone mode)
DEFAULT_WORKER_ID = None # Will use hostname if not specified
DEFAULT_WORKER_POLL_INTERVAL_MS = 500 # Poll database every 500ms
DEFAULT_WORKER_MAX_RETRIES = 3 # Max retries before marking task failed
DEFAULT_WORKER_BATCH_SIZE = 10 # Tasks to claim per poll cycle
DEFAULT_WORKER_HTTP_PORT = 8889 # HTTP port for worker metrics/health
DEFAULT_WORKER_MAX_SLOTS = 10 # Total concurrent tasks per worker
DEFAULT_WORKER_CONSOLIDATION_MAX_SLOTS = 2 # Max concurrent consolidation tasks per worker
# Reflect agent settings
DEFAULT_REFLECT_MAX_ITERATIONS = 10 # Max tool call iterations before forcing response
@@ -397,6 +397,7 @@ class HindsightConfig:
retain_extract_causal_links: bool
retain_extraction_mode: str
retain_custom_instructions: str | None
retain_observations_async: bool
# Observations settings (consolidated knowledge from facts)
enable_observations: bool
@@ -421,9 +422,8 @@ class HindsightConfig:
worker_id: str | None
worker_poll_interval_ms: int
worker_max_retries: int
worker_batch_size: int
worker_http_port: int
worker_max_slots: int
worker_consolidation_max_slots: int
# Reflect agent settings
reflect_max_iterations: int
@@ -565,6 +565,10 @@ class HindsightConfig:
os.getenv(ENV_RETAIN_EXTRACTION_MODE, DEFAULT_RETAIN_EXTRACTION_MODE)
),
retain_custom_instructions=os.getenv(ENV_RETAIN_CUSTOM_INSTRUCTIONS) or DEFAULT_RETAIN_CUSTOM_INSTRUCTIONS,
retain_observations_async=os.getenv(
ENV_RETAIN_OBSERVATIONS_ASYNC, str(DEFAULT_RETAIN_OBSERVATIONS_ASYNC)
).lower()
== "true",
# Observations settings (consolidated knowledge from facts)
enable_observations=os.getenv(ENV_ENABLE_OBSERVATIONS, str(DEFAULT_ENABLE_OBSERVATIONS)).lower() == "true",
consolidation_batch_size=int(
@@ -585,11 +589,8 @@ class HindsightConfig:
worker_id=os.getenv(ENV_WORKER_ID) or DEFAULT_WORKER_ID,
worker_poll_interval_ms=int(os.getenv(ENV_WORKER_POLL_INTERVAL_MS, str(DEFAULT_WORKER_POLL_INTERVAL_MS))),
worker_max_retries=int(os.getenv(ENV_WORKER_MAX_RETRIES, str(DEFAULT_WORKER_MAX_RETRIES))),
worker_batch_size=int(os.getenv(ENV_WORKER_BATCH_SIZE, str(DEFAULT_WORKER_BATCH_SIZE))),
worker_http_port=int(os.getenv(ENV_WORKER_HTTP_PORT, str(DEFAULT_WORKER_HTTP_PORT))),
worker_max_slots=int(os.getenv(ENV_WORKER_MAX_SLOTS, str(DEFAULT_WORKER_MAX_SLOTS))),
worker_consolidation_max_slots=int(
os.getenv(ENV_WORKER_CONSOLIDATION_MAX_SLOTS, str(DEFAULT_WORKER_CONSOLIDATION_MAX_SLOTS))
),
# Reflect agent settings
reflect_max_iterations=int(os.getenv(ENV_REFLECT_MAX_ITERATIONS, str(DEFAULT_REFLECT_MAX_ITERATIONS))),
)
@@ -865,14 +865,7 @@ Focus on DURABLE knowledge that serves this mission, not ephemeral state.
)
# Parse JSON response - should be an array
if isinstance(result, str):
# Strip markdown code fences (some models wrap JSON in ```json ... ```)
clean = result.strip()
if clean.startswith("```"):
clean = clean.split("\n", 1)[1] if "\n" in clean else clean[3:]
if clean.endswith("```"):
clean = clean[:-3]
clean = clean.strip()
result = json.loads(clean)
result = json.loads(result)
# Ensure result is a list
if isinstance(result, list):
return result
@@ -442,6 +442,49 @@ class MemoryEngineInterface(ABC):
"""
...
@abstractmethod
async def get_entity_observations(
self,
bank_id: str,
entity_id: str,
*,
limit: int = 10,
request_context: "RequestContext",
) -> list[Any]:
"""
Get observations for an entity.
Args:
bank_id: The memory bank ID.
entity_id: The entity ID.
limit: Maximum observations.
request_context: Request context for authentication.
Returns:
List of EntityObservation objects.
"""
...
@abstractmethod
async def regenerate_entity_observations(
self,
bank_id: str,
entity_id: str,
entity_name: str,
*,
request_context: "RequestContext",
) -> None:
"""
Regenerate observations for an entity.
Args:
bank_id: The memory bank ID.
entity_id: The entity ID.
entity_name: The entity's canonical name.
request_context: Request context for authentication.
"""
...
# =========================================================================
# Statistics & Operations
# =========================================================================
@@ -105,6 +105,9 @@ class LLMProvider:
self._mock_calls: list[dict] = []
self._mock_response: Any = None
# Vertex AI token refresher
self._vertexai_refresher: Any = None
# Set default base URLs
if not self.base_url:
if self.provider == "groq":
@@ -114,48 +117,62 @@ class LLMProvider:
elif self.provider == "lmstudio":
self.base_url = "http://localhost:1234/v1"
# Vertex AI config — stored for client creation below
self._vertexai_project_id: str | None = None
self._vertexai_region: str | None = None
self._vertexai_credentials: Any = None
# Handle Vertex AI provider
if self.provider == "vertexai":
if not VERTEXAI_AVAILABLE:
raise ValueError("Vertex AI requires 'google-auth' package. Install with: pip install google-auth")
from ..config import get_config
config = get_config()
self._vertexai_project_id = config.llm_vertexai_project_id
if not self._vertexai_project_id:
project_id = config.llm_vertexai_project_id
if not project_id:
raise ValueError(
"HINDSIGHT_API_LLM_VERTEXAI_PROJECT_ID is required for Vertex AI provider. "
"Set it to your GCP project ID."
)
self._vertexai_region = config.llm_vertexai_region or "us-central1"
region = config.llm_vertexai_region or "us-central1"
service_account_key = config.llm_vertexai_service_account_key
# Load explicit service account credentials if provided
if service_account_key:
if not VERTEXAI_AVAILABLE:
raise ValueError(
"Vertex AI service account auth requires 'google-auth' package. "
"Install with: pip install google-auth"
# Try ADC first
credentials = None
auth_method = None
try:
credentials, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"])
auth_method = "ADC"
logger.info("Vertex AI: Using Application Default Credentials")
except google.auth.exceptions.DefaultCredentialsError:
logger.debug("Vertex AI: ADC not available, trying service account")
# Fall back to service account key file
if credentials is None and service_account_key:
try:
credentials = service_account.Credentials.from_service_account_file(
service_account_key,
scopes=["https://www.googleapis.com/auth/cloud-platform"],
)
self._vertexai_credentials = service_account.Credentials.from_service_account_file(
service_account_key,
scopes=["https://www.googleapis.com/auth/cloud-platform"],
auth_method = "Service Account"
logger.info(f"Vertex AI: Using service account key: {service_account_key}")
except Exception as e:
logger.error(f"Vertex AI: Failed to load service account key: {e}")
if credentials is None:
raise ValueError(
"Vertex AI authentication failed. Either:\n"
" 1. Set up ADC: gcloud auth application-default login\n"
" 2. Set HINDSIGHT_API_LLM_VERTEXAI_SERVICE_ACCOUNT_KEY to path of service account JSON key"
)
logger.info(f"Vertex AI: Using service account key: {service_account_key}")
# Strip google/ prefix from model name — native SDK uses bare names
# e.g. "google/gemini-2.0-flash-lite-001" -> "gemini-2.0-flash-lite-001"
if self.model.startswith("google/"):
self.model = self.model[len("google/") :]
# Initialize token refresher
from .vertexai_token_refresher import VertexAITokenRefresher
logger.info(
f"Vertex AI: project={self._vertexai_project_id}, region={self._vertexai_region}, "
f"model={self.model}, auth={'service_account' if service_account_key else 'ADC'}"
)
self._vertexai_refresher = VertexAITokenRefresher(credentials, project_id, region)
self.base_url = self._vertexai_refresher.get_base_url()
logger.info(f"Vertex AI: project={project_id}, region={region}, auth={auth_method}")
# Validate API key (not needed for ollama, lmstudio, vertexai, or mock)
if self.provider not in ("ollama", "lmstudio", "vertexai", "mock") and not self.api_key:
@@ -185,16 +202,30 @@ class LLMProvider:
anthropic_kwargs["timeout"] = self.timeout
self._anthropic_client = AsyncAnthropic(**anthropic_kwargs)
elif self.provider == "vertexai":
# Native genai SDK with Vertex AI — handles ADC automatically,
# or uses explicit service account credentials if provided
# Custom transport for token injection
class TokenInjectingTransport(httpx.AsyncHTTPTransport):
def __init__(self, refresher, *args, **kwargs):
super().__init__(*args, **kwargs)
self._refresher = refresher
async def handle_async_request(self, request):
token = self._refresher.get_token()
request.headers["Authorization"] = f"Bearer {token}"
return await super().handle_async_request(request)
transport = TokenInjectingTransport(self._vertexai_refresher)
client_kwargs = {
"vertexai": True,
"project": self._vertexai_project_id,
"location": self._vertexai_region,
"api_key": "dummy", # Required by AsyncOpenAI but unused (we inject token via transport)
"base_url": self.base_url,
"max_retries": 0,
"http_client": httpx.AsyncClient(transport=transport),
}
if self._vertexai_credentials is not None:
client_kwargs["credentials"] = self._vertexai_credentials
self._gemini_client = genai.Client(**client_kwargs)
if self.timeout:
client_kwargs["timeout"] = self.timeout
self._client = AsyncOpenAI(**client_kwargs)
# Start background refresh
self._vertexai_refresher.start_refresh_task()
elif self.provider in ("ollama", "lmstudio"):
# Use dummy key if not provided for local
api_key = self.api_key or "local"
@@ -286,8 +317,8 @@ class LLMProvider:
return_usage,
)
# Handle Gemini and Vertex AI providers (both use native genai SDK)
if self.provider in ("gemini", "vertexai"):
# Handle Gemini provider separately
if self.provider == "gemini":
return await self._call_gemini(
messages,
response_format,
@@ -651,8 +682,8 @@ class LLMProvider:
messages, tools, max_completion_tokens, max_retries, initial_backoff, max_backoff, start_time, scope
)
# Handle Gemini and Vertex AI (convert to Gemini tool format)
if self.provider in ("gemini", "vertexai"):
# Handle Gemini (convert to Gemini tool format)
if self.provider == "gemini":
return await self._call_with_tools_gemini(
messages, tools, max_retries, initial_backoff, max_backoff, start_time, scope
)
@@ -1572,8 +1603,10 @@ class LLMProvider:
self._mock_calls = []
async def cleanup(self) -> None:
"""Clean up resources."""
pass
"""Clean up resources (e.g., stop token refresh tasks)."""
if self._vertexai_refresher is not None:
await self._vertexai_refresher.stop()
logger.debug("Vertex AI token refresher stopped")
@classmethod
def for_memory(cls) -> "LLMProvider":
@@ -504,11 +504,12 @@ class MemoryEngine(MemoryEngineInterface):
if request_context is None:
raise AuthenticationError("RequestContext is required when tenant extension is configured")
# For internal/background operations (e.g., worker tasks), skip extension authentication.
# The task was already authenticated at submission time, and execute_task sets _current_schema
# from the task's _schema field. For public schema tasks, _current_schema keeps its default "public".
# For internal/background operations (e.g., worker tasks), skip extension authentication
# if the schema has already been set by execute_task via the _schema field.
if request_context.internal:
return _current_schema.get()
current = _current_schema.get()
if current and current != "public":
return current
# Let AuthenticationError propagate - HTTP layer will convert to 401
tenant_context = await self._tenant_extension.authenticate(request_context)
@@ -888,23 +889,6 @@ class MemoryEngine(MemoryEngineInterface):
# Use configured database schema for migrations (defaults to "public")
run_migrations(self.db_url, schema=get_config().database_schema)
# Migrate all existing tenant schemas (if multi-tenant)
if self._tenant_extension is not None:
try:
tenants = await self._tenant_extension.list_tenants()
if tenants:
logger.info(f"Running migrations on {len(tenants)} tenant schemas...")
for tenant in tenants:
schema = tenant.schema
if schema and schema != "public":
try:
run_migrations(self.db_url, schema=schema)
except Exception as e:
logger.warning(f"Failed to migrate tenant schema {schema}: {e}")
logger.info("Tenant schema migrations completed")
except Exception as e:
logger.warning(f"Failed to run tenant schema migrations: {e}")
# Ensure embedding column dimension matches the model's dimension
# This is done after migrations and after embeddings.initialize()
ensure_embedding_dimension(self.db_url, self.embeddings.dimension, schema=get_config().database_schema)
@@ -1191,8 +1175,8 @@ class MemoryEngine(MemoryEngineInterface):
context: Context about when/why this memory was formed
event_date: When the event occurred (defaults to now)
document_id: Optional document ID for tracking (always upserts if document already exists)
fact_type_override: Override fact type ('world', 'experience')
confidence_score: Confidence score (0.0 to 1.0)
fact_type_override: Override fact type ('world', 'experience', 'opinion')
confidence_score: Confidence score for opinions (0.0 to 1.0)
request_context: Request context for authentication.
Returns:
@@ -1247,8 +1231,8 @@ class MemoryEngine(MemoryEngineInterface):
- "document_id" (optional): Document ID for this specific content item
document_id: **DEPRECATED** - Use "document_id" key in each content dict instead.
Applies the same document_id to ALL content items that don't specify their own.
fact_type_override: Override fact type for all facts ('world', 'experience')
confidence_score: Confidence score (0.0 to 1.0)
fact_type_override: Override fact type for all facts ('world', 'experience', 'opinion')
confidence_score: Confidence score for opinions (0.0 to 1.0)
return_usage: If True, returns tuple of (unit_ids, TokenUsage). Default False for backward compatibility.
Returns:
@@ -1570,16 +1554,16 @@ class MemoryEngine(MemoryEngineInterface):
if fact_type is None:
fact_type = list(VALID_RECALL_FACT_TYPES)
# Filter out 'opinion' early (deprecated, silently ignore)
fact_type = [ft for ft in fact_type if ft != "opinion"]
# Validate fact types
# Validate fact types early
invalid_types = set(fact_type) - VALID_RECALL_FACT_TYPES
if invalid_types:
raise ValueError(
f"Invalid fact type(s): {', '.join(sorted(invalid_types))}. "
f"Must be one of: {', '.join(sorted(VALID_RECALL_FACT_TYPES))}"
)
# Filter out 'opinion' - opinions are no longer returned from recall
fact_type = [ft for ft in fact_type if ft != "opinion"]
if not fact_type:
# All requested types were opinions - return empty result
return RecallResultModel(results=[], entities={}, chunks={})
@@ -2235,15 +2219,44 @@ class MemoryEngine(MemoryEngineInterface):
)
top_results_dicts.append(result_dict)
# Get entities for each fact if include_entities is requested
fact_entity_map = {} # unit_id -> list of (entity_id, entity_name)
if include_entities and top_scored:
unit_ids = [uuid.UUID(sr.id) for sr in top_scored]
if unit_ids:
async with acquire_with_retry(pool) as entity_conn:
entity_rows = await entity_conn.fetch(
f"""
SELECT ue.unit_id, e.id as entity_id, e.canonical_name
FROM {fq_table("unit_entities")} ue
JOIN {fq_table("entities")} e ON ue.entity_id = e.id
WHERE ue.unit_id = ANY($1::uuid[])
""",
unit_ids,
)
for row in entity_rows:
unit_id = str(row["unit_id"])
if unit_id not in fact_entity_map:
fact_entity_map[unit_id] = []
fact_entity_map[unit_id].append(
{"entity_id": str(row["entity_id"]), "canonical_name": row["canonical_name"]}
)
# Convert results to MemoryFact objects
memory_facts = []
for result_dict in top_results_dicts:
result_id = str(result_dict.get("id"))
# Get entity names for this fact
entity_names = None
if include_entities and result_id in fact_entity_map:
entity_names = [e["canonical_name"] for e in fact_entity_map[result_id]]
memory_facts.append(
MemoryFact(
id=str(result_dict.get("id")),
id=result_id,
text=result_dict.get("text"),
fact_type=result_dict.get("fact_type", "world"),
entities=None, # Entity observations removed
entities=entity_names,
context=result_dict.get("context"),
occurred_start=result_dict.get("occurred_start"),
occurred_end=result_dict.get("occurred_end"),
@@ -2254,12 +2267,38 @@ class MemoryEngine(MemoryEngineInterface):
)
)
# Entity observations removed - always set to None
# Fetch entity observations if requested
entities_dict = None
total_entity_tokens = 0
total_chunk_tokens = 0
if include_entities and fact_entity_map:
# Collect unique entities in order of fact relevance (preserving order from top_scored)
# Use a list to maintain order, but track seen entities to avoid duplicates
entities_ordered = [] # list of (entity_id, entity_name) tuples
seen_entity_ids = set()
# Iterate through facts in relevance order
for sr in top_scored:
unit_id = sr.id
if unit_id in fact_entity_map:
for entity in fact_entity_map[unit_id]:
entity_id = entity["entity_id"]
entity_name = entity["canonical_name"]
if entity_id not in seen_entity_ids:
entities_ordered.append((entity_id, entity_name))
seen_entity_ids.add(entity_id)
# Return entities with empty observations (summaries now live in mental models)
entities_dict = {}
for entity_id, entity_name in entities_ordered:
entities_dict[entity_name] = EntityState(
entity_id=entity_id,
canonical_name=entity_name,
observations=[], # Mental models provide this now
)
# Fetch chunks if requested
chunks_dict = None
total_chunk_tokens = 0
if include_chunks and top_scored:
from .response_models import ChunkInfo
@@ -2328,6 +2367,7 @@ class MemoryEngine(MemoryEngineInterface):
# Log final recall stats
total_time = time.time() - recall_start
num_chunks = len(chunks_dict) if chunks_dict else 0
num_entities = len(entities_dict) if entities_dict else 0
# Include wait times in log if significant
wait_parts = []
if semaphore_wait > 0.01:
@@ -2336,7 +2376,7 @@ class MemoryEngine(MemoryEngineInterface):
wait_parts.append(f"conn={max_conn_wait:.3f}s")
wait_info = f" | waits: {', '.join(wait_parts)}" if wait_parts else ""
log_buffer.append(
f"[RECALL {recall_id}] Complete: {len(top_scored)} facts ({total_tokens} tok), {num_chunks} chunks ({total_chunk_tokens} tok) | {fact_type_summary} | {total_time:.3f}s{wait_info}"
f"[RECALL {recall_id}] Complete: {len(top_scored)} facts ({total_tokens} tok), {num_chunks} chunks ({total_chunk_tokens} tok), {num_entities} entities ({total_entity_tokens} tok) | {fact_type_summary} | {total_time:.3f}s{wait_info}"
)
if not quiet:
logger.info("\n" + "\n".join(log_buffer))
@@ -3510,6 +3550,7 @@ class MemoryEngine(MemoryEngineInterface):
ReflectResult containing:
- text: Plain text answer
- based_on: Empty dict (agent retrieves facts dynamically)
- new_opinions: Empty list
- structured_output: None (not yet supported for agentic reflect)
"""
# Use cached LLM config
@@ -3834,6 +3875,7 @@ class MemoryEngine(MemoryEngineInterface):
result = ReflectResult(
text=agent_result.text,
based_on=based_on,
new_opinions=[], # Learnings stored as mental models
structured_output=agent_result.structured_output,
usage=usage,
tool_trace=tool_trace_result,
@@ -3862,6 +3904,32 @@ class MemoryEngine(MemoryEngineInterface):
return result
async def get_entity_observations(
self,
bank_id: str,
entity_id: str,
*,
limit: int = 10,
request_context: "RequestContext",
) -> list[Any]:
"""
Get observations for an entity.
NOTE: Entity observations/summaries have been moved to mental models.
This method returns an empty list. Use mental models for entity summaries.
Args:
bank_id: bank IDentifier
entity_id: Entity UUID to get observations for
limit: Ignored (kept for backwards compatibility)
request_context: Request context for authentication.
Returns:
Empty list (observations now in mental models)
"""
await self._authenticate_tenant(request_context)
return []
async def list_entities(
self,
bank_id: str,
@@ -4048,6 +4116,36 @@ class MemoryEngine(MemoryEngineInterface):
await self._authenticate_tenant(request_context)
return EntityState(entity_id=entity_id, canonical_name=entity_name, observations=[])
async def regenerate_entity_observations(
self,
bank_id: str,
entity_id: str,
entity_name: str,
*,
version: str | None = None,
conn=None,
request_context: "RequestContext",
) -> list[str]:
"""
Regenerate observations for an entity.
NOTE: Entity observations/summaries have been moved to mental models.
This method is now a no-op and returns an empty list.
Args:
bank_id: bank IDentifier
entity_id: Entity UUID
entity_name: Canonical name of the entity
version: Entity's last_seen timestamp when task was created (for deduplication)
conn: Optional database connection (ignored)
request_context: Request context for authentication.
Returns:
Empty list (observations now in mental models)
"""
await self._authenticate_tenant(request_context)
return []
# =========================================================================
# Statistics & Operations (for HTTP API layer)
# =========================================================================
@@ -4158,6 +4256,9 @@ class MemoryEngine(MemoryEngineInterface):
if not entity_row:
return None
# Get observations for the entity
observations = await self.get_entity_observations(bank_id, entity_id, limit=20, request_context=request_context)
return {
"id": str(entity_row["id"]),
"canonical_name": entity_row["canonical_name"],
@@ -4165,7 +4266,7 @@ class MemoryEngine(MemoryEngineInterface):
"first_seen": entity_row["first_seen"].isoformat() if entity_row["first_seen"] else None,
"last_seen": entity_row["last_seen"].isoformat() if entity_row["last_seen"] else None,
"metadata": entity_row["metadata"] or {},
"observations": [],
"observations": observations,
}
def _parse_observations(self, observations_raw: list):
@@ -263,6 +263,7 @@ class ReflectResult(BaseModel):
}
],
},
"new_opinions": ["Machine learning has great potential in healthcare"],
"structured_output": {"summary": "ML in healthcare", "confidence": 0.9},
"usage": {"input_tokens": 1500, "output_tokens": 500, "total_tokens": 2000},
}
@@ -271,8 +272,9 @@ class ReflectResult(BaseModel):
text: str = Field(description="The formulated answer text")
based_on: dict[str, Any] = Field(
description="Facts used to formulate the answer, organized by type (world, experience, mental_models, directives)"
description="Facts used to formulate the answer, organized by type (world, experience, opinion, mental_models, directives)"
)
new_opinions: list[str] = Field(default_factory=list, description="List of newly formed opinions during reflection")
structured_output: dict[str, Any] | None = Field(
default=None,
description="Structured output parsed according to the provided response schema. Only present when response_schema was provided.",
@@ -295,6 +297,24 @@ class ReflectResult(BaseModel):
)
class Opinion(BaseModel):
"""
An opinion with confidence score.
Opinions represent the bank's formed perspectives on topics,
with a confidence level indicating strength of belief.
"""
model_config = ConfigDict(
json_schema_extra={
"example": {"text": "Machine learning has great potential in healthcare", "confidence": 0.85}
}
)
text: str = Field(description="The opinion text")
confidence: float = Field(description="Confidence score between 0.0 and 1.0")
class EntityObservation(BaseModel):
"""
An observation about an entity.
@@ -693,6 +693,7 @@ async def _extract_facts_from_chunk(
context: str,
llm_config: "LLMConfig",
agent_name: str = None,
extract_opinions: bool = False,
) -> tuple[list[dict[str, str]], TokenUsage]:
"""
Extract facts from a single chunk (internal helper for parallel processing).
@@ -706,9 +707,17 @@ async def _extract_facts_from_chunk(
logger = logging.getLogger(__name__)
# Determine which fact types to extract
memory_bank_context = f"\n- Your name: {agent_name}" if agent_name and extract_opinions else ""
# Determine which fact types to extract based on the flag
# Note: We use "assistant" in the prompt but convert to "bank" for storage
fact_types_instruction = "Extract ONLY 'world' and 'assistant' type facts."
if extract_opinions:
# Opinion extraction uses a separate prompt (not this one)
fact_types_instruction = "Extract ONLY 'opinion' type facts (formed opinions, beliefs, and perspectives). DO NOT extract 'world' or 'assistant' facts."
else:
fact_types_instruction = (
"Extract ONLY 'world' and 'assistant' type facts. DO NOT extract opinions - those are extracted separately."
)
# Check config for extraction mode and causal link extraction
config = get_config()
@@ -761,6 +770,7 @@ async def _extract_facts_from_chunk(
# Format event_date with day of week for better temporal reasoning
event_date_formatted = event_date.strftime("%A, %B %d, %Y") # e.g., "Monday, June 10, 2024"
user_message = f"""Extract facts from the following text chunk.
{memory_bank_context}
Chunk: {chunk_index + 1}/{total_chunks}
Event Date: {event_date_formatted} ({event_date.isoformat()})
@@ -1019,6 +1029,7 @@ async def _extract_facts_with_auto_split(
context: str,
llm_config: LLMConfig,
agent_name: str = None,
extract_opinions: bool = False,
) -> tuple[list[dict[str, str]], TokenUsage]:
"""
Extract facts from a chunk with automatic splitting if output exceeds token limits.
@@ -1034,6 +1045,7 @@ async def _extract_facts_with_auto_split(
context: Context about the conversation/document
llm_config: LLM configuration to use
agent_name: Optional agent name (memory owner)
extract_opinions: If True, extract ONLY opinions. If False, extract world and agent facts (no opinions)
Returns:
Tuple of (facts list, token usage) extracted from the chunk (possibly from sub-chunks)
@@ -1052,6 +1064,7 @@ async def _extract_facts_with_auto_split(
context=context,
llm_config=llm_config,
agent_name=agent_name,
extract_opinions=extract_opinions,
)
except OutputTooLongError:
# Output exceeded token limits - split the chunk in half and retry
@@ -1096,6 +1109,7 @@ async def _extract_facts_with_auto_split(
context=context,
llm_config=llm_config,
agent_name=agent_name,
extract_opinions=extract_opinions,
),
_extract_facts_with_auto_split(
chunk=second_half,
@@ -1105,6 +1119,7 @@ async def _extract_facts_with_auto_split(
context=context,
llm_config=llm_config,
agent_name=agent_name,
extract_opinions=extract_opinions,
),
]
@@ -1128,6 +1143,7 @@ async def extract_facts_from_text(
llm_config: LLMConfig,
agent_name: str,
context: str = "",
extract_opinions: bool = False,
) -> tuple[list[Fact], list[tuple[str, int]], TokenUsage]:
"""
Extract semantic facts from conversational or narrative text using LLM.
@@ -1144,6 +1160,7 @@ async def extract_facts_from_text(
context: Context about the conversation/document
llm_config: LLM configuration to use
agent_name: Agent name (memory owner)
extract_opinions: If True, extract ONLY opinions. If False, extract world and bank facts (no opinions)
Returns:
Tuple of (facts, chunks, usage) where:
@@ -1171,6 +1188,7 @@ async def extract_facts_from_text(
context=context,
llm_config=llm_config,
agent_name=agent_name,
extract_opinions=extract_opinions,
)
for i, chunk in enumerate(chunks)
]
@@ -1202,7 +1220,7 @@ SECONDS_PER_FACT = 10
async def extract_facts_from_contents(
contents: list[RetainContent], llm_config, agent_name: str
contents: list[RetainContent], llm_config, agent_name: str, extract_opinions: bool = False
) -> tuple[list[ExtractedFactType], list[ChunkMetadata], TokenUsage]:
"""
Extract facts from multiple content items in parallel.
@@ -1217,6 +1235,7 @@ async def extract_facts_from_contents(
contents: List of RetainContent objects to process
llm_config: LLM configuration for fact extraction
agent_name: Name of the agent (for agent-related fact detection)
extract_opinions: If True, extract only opinions; otherwise world/bank facts
Returns:
Tuple of (extracted_facts, chunks_metadata, usage)
@@ -1235,6 +1254,7 @@ async def extract_facts_from_contents(
context=item.context,
llm_config=llm_config,
agent_name=agent_name,
extract_opinions=extract_opinions,
)
fact_extraction_tasks.append(task)
@@ -101,8 +101,11 @@ async def retain_batch(
# Step 1: Extract facts from all contents
step_start = time.time()
extract_opinions = fact_type_override == "opinion"
extracted_facts, chunks, usage = await fact_extraction.extract_facts_from_contents(contents, llm_config, agent_name)
extracted_facts, chunks, usage = await fact_extraction.extract_facts_from_contents(
contents, llm_config, agent_name, extract_opinions
)
log_buffer.append(
f"[1] Extract facts: {len(extracted_facts)} facts, {len(chunks)} chunks from {len(contents)} contents in {time.time() - step_start:.3f}s"
)
@@ -19,6 +19,7 @@ async def extract_facts(
context: str = "",
llm_config: "LLMConfig" = None,
agent_name: str = None,
extract_opinions: bool = False,
) -> tuple[list["Fact"], list[tuple[str, int]]]:
"""
Extract semantic facts from text using LLM.
@@ -35,6 +36,7 @@ async def extract_facts(
context: Context about the conversation/document
llm_config: LLM configuration to use
agent_name: Optional agent name to help identify agent-related facts
extract_opinions: If True, extract ONLY opinions. If False, extract world and agent facts (no opinions)
Returns:
Tuple of (facts, chunks) where:
@@ -53,6 +55,7 @@ async def extract_facts(
context=context,
llm_config=llm_config,
agent_name=agent_name,
extract_opinions=extract_opinions,
)
if not facts:
@@ -0,0 +1,120 @@
"""Vertex AI token refresher with background refresh and caching."""
import asyncio
import logging
import threading
from datetime import datetime
from typing import Any
logger = logging.getLogger(__name__)
class VertexAITokenRefresher:
"""
Background token refresher for Vertex AI.
Refreshes Google Cloud access tokens every 50 minutes to ensure they don't expire (60-min default).
Thread-safe token caching for concurrent access from multiple async tasks.
"""
def __init__(self, credentials: Any, project_id: str, region: str):
"""
Initialize the token refresher.
Args:
credentials: Google Cloud credentials object (from google.auth.default or service_account)
project_id: GCP project ID
region: GCP region (e.g., "us-central1")
"""
self._credentials = credentials
self._project_id = project_id
self._region = region
# Thread-safe token cache
self._token: str | None = None
self._token_expiry: datetime | None = None
self._lock = threading.Lock()
# Background refresh task
self._refresh_task: asyncio.Task | None = None
self._stop_event = asyncio.Event()
# Initial token fetch (synchronous, must complete before returning)
self._refresh_token_sync()
def _refresh_token_sync(self) -> None:
"""Synchronously refresh the token (thread-safe)."""
try:
import google.auth.transport.requests
request = google.auth.transport.requests.Request()
self._credentials.refresh(request)
with self._lock:
self._token = self._credentials.token
self._token_expiry = self._credentials.expiry
logger.debug(f"Vertex AI token refreshed, expires at {self._token_expiry}")
except Exception as e:
logger.error(f"Failed to refresh Vertex AI token: {e}")
raise
async def _refresh_loop(self) -> None:
"""Background refresh loop (runs every 50 minutes)."""
while not self._stop_event.is_set():
try:
# Wait 50 minutes or until stop event
await asyncio.wait_for(self._stop_event.wait(), timeout=50 * 60)
# If we get here, stop was signaled
break
except asyncio.TimeoutError:
# 50 minutes passed, refresh token
try:
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, self._refresh_token_sync)
except Exception as e:
logger.error(f"Background token refresh failed: {e}")
# Continue loop - next API call will fail with auth error
def start_refresh_task(self) -> None:
"""Start the background refresh task."""
if self._refresh_task is None or self._refresh_task.done():
self._refresh_task = asyncio.create_task(self._refresh_loop())
logger.info("Vertex AI token refresh task started (refreshes every 50 minutes)")
async def stop(self) -> None:
"""Stop the background refresh task."""
if self._refresh_task is not None and not self._refresh_task.done():
self._stop_event.set()
try:
await asyncio.wait_for(self._refresh_task, timeout=5.0)
except asyncio.TimeoutError:
logger.warning("Vertex AI token refresh task did not stop within 5 seconds")
logger.info("Vertex AI token refresh task stopped")
def get_token(self) -> str:
"""
Get current access token (thread-safe).
Returns:
Current Google Cloud access token
Raises:
RuntimeError: If token is not available
"""
with self._lock:
if self._token is None:
raise RuntimeError("Vertex AI token not available")
return self._token
def get_base_url(self) -> str:
"""
Get the Vertex AI OpenAI-compatible endpoint URL.
Returns:
Base URL for Vertex AI OpenAI API
"""
return (
f"https://{self._region}-aiplatform.googleapis.com/v1beta1/"
f"projects/{self._project_id}/locations/{self._region}/endpoints/openapi"
)
+2 -2
View File
@@ -239,6 +239,7 @@ def main():
retain_extract_causal_links=config.retain_extract_causal_links,
retain_extraction_mode=config.retain_extraction_mode,
retain_custom_instructions=config.retain_custom_instructions,
retain_observations_async=config.retain_observations_async,
enable_observations=config.enable_observations,
consolidation_batch_size=config.consolidation_batch_size,
consolidation_max_tokens=config.consolidation_max_tokens,
@@ -253,9 +254,8 @@ def main():
worker_id=config.worker_id,
worker_poll_interval_ms=config.worker_poll_interval_ms,
worker_max_retries=config.worker_max_retries,
worker_batch_size=config.worker_batch_size,
worker_http_port=config.worker_http_port,
worker_max_slots=config.worker_max_slots,
worker_consolidation_max_slots=config.worker_consolidation_max_slots,
reflect_max_iterations=config.reflect_max_iterations,
mental_model_refresh_concurrency=config.mental_model_refresh_concurrency,
)
+12 -31
View File
@@ -32,9 +32,6 @@ class MCPToolsConfig:
# How to resolve bank_id for operations
bank_id_resolver: Callable[[], str | None]
# How to resolve API key for tenant auth (optional)
api_key_resolver: Callable[[], str | None] | None = None
# Whether to include bank_id as a parameter on tools (for multi-bank support)
include_bank_id_param: bool = False
@@ -49,16 +46,6 @@ class MCPToolsConfig:
retain_fire_and_forget: bool = False # If True, use asyncio.create_task pattern
def _get_request_context(config: MCPToolsConfig) -> RequestContext:
"""Create RequestContext with API key from resolver if available.
This enables tenant auth to work with MCP tools by propagating
the Bearer token from the MCP middleware to the memory engine.
"""
api_key = config.api_key_resolver() if config.api_key_resolver else None
return RequestContext(api_key=api_key)
def parse_timestamp(timestamp: str) -> datetime | None:
"""Parse an ISO format timestamp string.
@@ -168,14 +155,12 @@ def _register_retain(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsConfig)
if error:
return {"status": "error", "message": error}
request_context = _get_request_context(config)
async def _retain():
try:
await memory.retain_batch_async(
bank_id=target_bank,
contents=[content_dict],
request_context=request_context,
request_context=RequestContext(),
)
except Exception as e:
logger.error(f"Error storing memory: {e}", exc_info=True)
@@ -211,17 +196,16 @@ def _register_retain(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsConfig)
return f"Error: {error}"
contents = [content_dict]
request_context = _get_request_context(config)
if async_processing:
result = await memory.submit_async_retain(
bank_id=target_bank, contents=contents, request_context=request_context
bank_id=target_bank, contents=contents, request_context=RequestContext()
)
return f"Memory queued for background processing (operation_id: {result.get('operation_id', 'N/A')})"
else:
await memory.retain_batch_async(
bank_id=target_bank,
contents=contents,
request_context=request_context,
request_context=RequestContext(),
)
return f"Memory stored successfully in bank '{target_bank}'"
except Exception as e:
@@ -253,14 +237,12 @@ def _register_retain(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsConfig)
if error:
return {"status": "error", "message": error}
request_context = _get_request_context(config)
async def _retain():
try:
await memory.retain_batch_async(
bank_id=target_bank,
contents=[content_dict],
request_context=request_context,
request_context=RequestContext(),
)
except Exception as e:
logger.error(f"Error storing memory: {e}", exc_info=True)
@@ -298,7 +280,7 @@ def _register_recall(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsConfig)
fact_type=list(VALID_RECALL_FACT_TYPES),
budget=Budget.HIGH,
max_tokens=max_tokens,
request_context=_get_request_context(config),
request_context=RequestContext(),
)
return recall_result.model_dump_json(indent=2)
@@ -329,7 +311,7 @@ def _register_recall(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsConfig)
fact_type=list(VALID_RECALL_FACT_TYPES),
budget=Budget.HIGH,
max_tokens=max_tokens,
request_context=_get_request_context(config),
request_context=RequestContext(),
)
return recall_result.model_dump()
@@ -388,7 +370,7 @@ def _register_reflect(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsConfig
query=query,
budget=budget_enum,
context=context,
request_context=_get_request_context(config),
request_context=RequestContext(),
)
return reflect_result.model_dump_json(indent=2)
@@ -441,7 +423,7 @@ def _register_reflect(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsConfig
query=query,
budget=budget_enum,
context=context,
request_context=_get_request_context(config),
request_context=RequestContext(),
)
return reflect_result.model_dump()
@@ -465,7 +447,7 @@ def _register_list_banks(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsCon
JSON list of banks with their IDs, names, dispositions, and missions.
"""
try:
banks = await memory.list_banks(request_context=_get_request_context(config))
banks = await memory.list_banks(request_context=RequestContext())
return json.dumps({"banks": banks}, indent=2)
except Exception as e:
logger.error(f"Error listing banks: {e}", exc_info=True)
@@ -489,9 +471,8 @@ def _register_create_bank(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsCo
mission: Optional mission describing who the agent is and what they're trying to accomplish
"""
try:
request_context = _get_request_context(config)
# get_bank_profile auto-creates bank if it doesn't exist
profile = await memory.get_bank_profile(bank_id, request_context=request_context)
profile = await memory.get_bank_profile(bank_id, request_context=RequestContext())
# Update name/mission if provided
if name is not None or mission is not None:
@@ -499,10 +480,10 @@ def _register_create_bank(mcp: FastMCP, memory: MemoryEngine, config: MCPToolsCo
bank_id,
name=name,
mission=mission,
request_context=request_context,
request_context=RequestContext(),
)
# Fetch updated profile
profile = await memory.get_bank_profile(bank_id, request_context=request_context)
profile = await memory.get_bank_profile(bank_id, request_context=RequestContext())
# Serialize disposition if it's a Pydantic model
if "disposition" in profile and hasattr(profile["disposition"], "model_dump"):
+3 -3
View File
@@ -189,7 +189,7 @@ class MetricsCollectorBase:
Args:
provider: LLM provider name (openai, anthropic, gemini, groq, ollama, lmstudio)
model: Model name
scope: Scope identifier (e.g., "memory", "reflect", "consolidation")
scope: Scope identifier (e.g., "memory", "reflect", "entity_observation")
duration: Call duration in seconds
input_tokens: Number of input/prompt tokens
output_tokens: Number of output/completion tokens
@@ -321,7 +321,7 @@ class MetricsCollector(MetricsCollectorBase):
pass
Args:
operation: Operation name (retain, recall, reflect, consolidation)
operation: Operation name (retain, recall, reflect, entity_observation)
bank_id: Memory bank ID
source: Source of the operation (api, reflect, internal)
budget: Optional budget level (low, mid, high)
@@ -371,7 +371,7 @@ class MetricsCollector(MetricsCollectorBase):
Args:
provider: LLM provider name (openai, anthropic, gemini, groq, ollama, lmstudio)
model: Model name
scope: Scope identifier (e.g., "memory", "reflect", "consolidation")
scope: Scope identifier (e.g., "memory", "reflect", "entity_observation")
duration: Call duration in seconds
input_tokens: Number of input/prompt tokens
output_tokens: Number of output/completion tokens
+8 -4
View File
@@ -124,6 +124,12 @@ def main():
default=config.worker_poll_interval_ms,
help=f"Poll interval in milliseconds (default: {config.worker_poll_interval_ms}, env: HINDSIGHT_API_WORKER_POLL_INTERVAL_MS)",
)
parser.add_argument(
"--batch-size",
type=int,
default=config.worker_batch_size,
help=f"Tasks to claim per poll (default: {config.worker_batch_size}, env: HINDSIGHT_API_WORKER_BATCH_SIZE)",
)
parser.add_argument(
"--max-retries",
type=int,
@@ -162,9 +168,8 @@ def main():
print(f"Starting Hindsight Worker: {args.worker_id}")
print(f" Poll interval: {args.poll_interval}ms")
print(f" Batch size: {args.batch_size}")
print(f" Max retries: {args.max_retries}")
print(f" Max slots: {config.worker_max_slots}")
print(f" Consolidation max slots: {config.worker_consolidation_max_slots}")
print(f" HTTP server: {args.http_host}:{args.http_port}")
print()
@@ -208,10 +213,9 @@ def main():
worker_id=args.worker_id,
executor=memory.execute_task,
poll_interval_ms=args.poll_interval,
batch_size=args.batch_size,
max_retries=args.max_retries,
tenant_extension=tenant_extension,
max_slots=config.worker_max_slots,
consolidation_max_slots=config.worker_consolidation_max_slots,
)
# Create the HTTP app for metrics/health
+96 -225
View File
@@ -57,11 +57,10 @@ class WorkerPoller:
worker_id: str,
executor: Callable[[dict[str, Any]], Awaitable[None]],
poll_interval_ms: int = 500,
batch_size: int = 10,
max_retries: int = 3,
schema: str | None = None,
tenant_extension: "TenantExtension | None" = None,
max_slots: int = 10,
consolidation_max_slots: int = 2,
):
"""
Initialize the worker poller.
@@ -71,32 +70,28 @@ class WorkerPoller:
worker_id: Unique identifier for this worker
executor: Async function to execute tasks (typically MemoryEngine.execute_task)
poll_interval_ms: Interval between polls when no tasks found (milliseconds)
batch_size: Maximum number of tasks to claim per poll cycle
max_retries: Maximum retry attempts before marking task as failed
schema: Database schema for single-tenant support (ignored if tenant_extension is set)
tenant_extension: Extension for dynamic multi-tenant discovery. If set, list_tenants()
is called on each poll cycle to discover schemas dynamically.
max_slots: Maximum concurrent tasks per worker
consolidation_max_slots: Maximum concurrent consolidation tasks per worker
"""
self._pool = pool
self._worker_id = worker_id
self._executor = executor
self._poll_interval_ms = poll_interval_ms
self._batch_size = batch_size
self._max_retries = max_retries
self._schema = schema
self._tenant_extension = tenant_extension
self._max_slots = max_slots
self._consolidation_max_slots = consolidation_max_slots
self._shutdown = asyncio.Event()
self._current_tasks: set[asyncio.Task] = set()
self._in_flight_count = 0
self._in_flight_lock = asyncio.Lock()
self._last_progress_log = 0.0
self._tasks_completed_since_log = 0
# Track active tasks locally: operation_id -> (op_type, bank_id, schema, asyncio.Task)
self._active_tasks: dict[str, tuple[str, str, str | None, asyncio.Task]] = {}
# Track in-flight tasks by operation type
self._in_flight_by_type: dict[str, int] = {}
# Track active tasks locally: operation_id -> (op_type, bank_id, schema)
self._active_tasks: dict[str, tuple[str, str, str | None]] = {}
async def _get_schemas(self) -> list[str | None]:
"""Get list of schemas to poll. Returns [None] for public schema."""
@@ -107,114 +102,59 @@ class WorkerPoller:
# Single schema mode
return [self._schema]
async def _get_available_slots(self) -> tuple[int, int]:
"""
Calculate available slots for claiming tasks.
Returns:
(total_available, consolidation_available) tuple
"""
async with self._in_flight_lock:
total_in_flight = self._in_flight_count
consolidation_in_flight = self._in_flight_by_type.get("consolidation", 0)
total_available = max(0, self._max_slots - total_in_flight)
consolidation_available = max(0, self._consolidation_max_slots - consolidation_in_flight)
return total_available, consolidation_available
async def wait_for_active_tasks(self, timeout: float = 10.0) -> bool:
"""
Wait for all active background tasks to complete (test helper).
This is a test-only utility that allows tests to synchronize with
fire-and-forget background tasks without using sleep().
Args:
timeout: Maximum time to wait in seconds
Returns:
True if all tasks completed, False if timeout was reached
"""
start_time = asyncio.get_event_loop().time()
while True:
async with self._in_flight_lock:
if self._in_flight_count == 0:
return True
elapsed = asyncio.get_event_loop().time() - start_time
if elapsed >= timeout:
return False
# Short sleep to avoid busy-waiting
await asyncio.sleep(0.01)
async def claim_batch(self) -> list[ClaimedTask]:
"""
Claim pending tasks atomically across all tenant schemas,
respecting slot limits (total and consolidation).
Claim up to batch_size pending tasks atomically across all tenant schemas.
Uses FOR UPDATE SKIP LOCKED to ensure no conflicts with other workers.
For consolidation tasks specifically, skips pending tasks if there's already
a processing consolidation for the same bank (to avoid duplicate work).
If tenant_extension is configured, dynamically discovers schemas on each call.
Returns:
List of ClaimedTask objects containing operation_id, task_dict, and schema
"""
# Calculate available slots
total_available, consolidation_available = await self._get_available_slots()
if total_available <= 0:
return []
schemas = await self._get_schemas()
all_tasks: list[ClaimedTask] = []
remaining_total = total_available
remaining_consolidation = consolidation_available
remaining_batch = self._batch_size
for schema in schemas:
if remaining_total <= 0:
if remaining_batch <= 0:
break
tasks = await self._claim_batch_for_schema(schema, remaining_total, remaining_consolidation)
# Update remaining slots based on what was claimed
for task in tasks:
op_type = task.task_dict.get("operation_type", "unknown")
if op_type == "consolidation":
remaining_consolidation -= 1
tasks = await self._claim_batch_for_schema(schema, remaining_batch)
all_tasks.extend(tasks)
remaining_total -= len(tasks)
remaining_batch -= len(tasks)
return all_tasks
async def _claim_batch_for_schema(
self, schema: str | None, limit: int, consolidation_limit: int
) -> list[ClaimedTask]:
"""Claim tasks from a specific schema respecting slot limits."""
try:
return await self._claim_batch_for_schema_inner(schema, limit, consolidation_limit)
except Exception as e:
logger.warning(f"Worker {self._worker_id} failed to claim tasks for schema {schema or 'public'}: {e}")
return []
async def _claim_batch_for_schema_inner(
self, schema: str | None, limit: int, consolidation_limit: int
) -> list[ClaimedTask]:
"""Inner implementation for claiming tasks from a specific schema with slot limits."""
async def _claim_batch_for_schema(self, schema: str | None, limit: int) -> list[ClaimedTask]:
"""Claim tasks from a specific schema."""
table = fq_table("async_operations", schema)
async with self._pool.acquire() as conn:
async with conn.transaction():
# Strategy: Claim non-consolidation tasks first, then consolidation up to limit
# 1. Claim non-consolidation tasks (up to limit)
non_consolidation_rows = await conn.fetch(
# Select and lock pending tasks
# For consolidation: skip if same bank already has one processing
rows = await conn.fetch(
f"""
SELECT operation_id, task_payload
FROM {table}
WHERE status = 'pending'
AND task_payload IS NOT NULL
AND operation_type != 'consolidation'
FROM {table} AS pending
WHERE status = 'pending' AND task_payload IS NOT NULL
AND (
-- Non-consolidation tasks: always claimable
operation_type != 'consolidation'
OR
-- Consolidation: only if no other consolidation processing for same bank
NOT EXISTS (
SELECT 1 FROM {table} AS processing
WHERE processing.bank_id = pending.bank_id
AND processing.operation_type = 'consolidation'
AND processing.status = 'processing'
)
)
ORDER BY created_at
LIMIT $1
FOR UPDATE SKIP LOCKED
@@ -222,39 +162,11 @@ class WorkerPoller:
limit,
)
claimed_count = len(non_consolidation_rows)
remaining_limit = limit - claimed_count
# 2. Claim consolidation tasks (up to consolidation_limit and remaining_limit)
consolidation_rows = []
if consolidation_limit > 0 and remaining_limit > 0:
consolidation_rows = await conn.fetch(
f"""
SELECT operation_id, task_payload
FROM {table} AS pending
WHERE status = 'pending'
AND task_payload IS NOT NULL
AND operation_type = 'consolidation'
AND NOT EXISTS (
SELECT 1 FROM {table} AS processing
WHERE processing.bank_id = pending.bank_id
AND processing.operation_type = 'consolidation'
AND processing.status = 'processing'
)
ORDER BY created_at
LIMIT $1
FOR UPDATE SKIP LOCKED
""",
min(consolidation_limit, remaining_limit),
)
all_rows = non_consolidation_rows + consolidation_rows
if not all_rows:
if not rows:
return []
# Claim the tasks by updating status and worker_id
operation_ids = [row["operation_id"] for row in all_rows]
operation_ids = [row["operation_id"] for row in rows]
await conn.execute(
f"""
UPDATE {table}
@@ -272,7 +184,7 @@ class WorkerPoller:
task_dict=json.loads(row["task_payload"]),
schema=schema,
)
for row in all_rows
for row in rows
]
async def _mark_completed(self, operation_id: str, schema: str | None):
@@ -338,43 +250,18 @@ class WorkerPoller:
logger.warning(f"Task {operation_id} failed, will retry (attempt {retry_count + 1}/{self._max_retries})")
async def execute_task(self, task: ClaimedTask):
"""Execute a single task as a background job (fire-and-forget)."""
"""Execute a single task and update its status."""
task_type = task.task_dict.get("type", "unknown")
operation_type = task.task_dict.get("operation_type", "unknown")
bank_id = task.task_dict.get("bank_id", "unknown")
# Create background task
bg_task = asyncio.create_task(self._execute_task_inner(task))
# Track this task as active
async with self._in_flight_lock:
self._active_tasks[task.operation_id] = (task_type, bank_id, task.schema, bg_task)
self._in_flight_count += 1
self._in_flight_by_type[operation_type] = self._in_flight_by_type.get(operation_type, 0) + 1
# Add cleanup callback
bg_task.add_done_callback(lambda _: asyncio.create_task(self._cleanup_task(task.operation_id, operation_type)))
async def _cleanup_task(self, operation_id: str, operation_type: str):
"""Remove task from tracking after completion."""
async with self._in_flight_lock:
if operation_id in self._active_tasks:
self._active_tasks.pop(operation_id, None)
self._in_flight_count -= 1
count = self._in_flight_by_type.get(operation_type, 0)
if count > 0:
self._in_flight_by_type[operation_type] = count - 1
if self._in_flight_by_type[operation_type] == 0:
del self._in_flight_by_type[operation_type]
async def _execute_task_inner(self, task: ClaimedTask):
"""Inner task execution with error handling."""
task_type = task.task_dict.get("type", "unknown")
bank_id = task.task_dict.get("bank_id", "unknown")
self._active_tasks[task.operation_id] = (task_type, bank_id, task.schema)
try:
schema_info = f", schema={task.schema}" if task.schema else ""
logger.debug(f"Executing task {task.operation_id} (type={task_type}, bank={bank_id}{schema_info})")
# Pass schema to executor so it can set the correct context
if task.schema:
task.task_dict["_schema"] = task.schema
await self._executor(task.task_dict)
@@ -384,6 +271,10 @@ class WorkerPoller:
error_msg = f"{type(e).__name__}: {e}\n{traceback.format_exc()}"
logger.error(f"Task {task.operation_id} failed: {e}")
await self._retry_or_fail(task.operation_id, error_msg, task.schema)
finally:
# Remove from active tasks
async with self._in_flight_lock:
self._active_tasks.pop(task.operation_id, None)
async def recover_own_tasks(self) -> int:
"""
@@ -402,23 +293,20 @@ class WorkerPoller:
total_count = 0
for schema in schemas:
try:
table = fq_table("async_operations", schema)
table = fq_table("async_operations", schema)
result = await self._pool.execute(
f"""
UPDATE {table}
SET status = 'pending', worker_id = NULL, claimed_at = NULL, updated_at = now()
WHERE status = 'processing' AND worker_id = $1
""",
self._worker_id,
)
result = await self._pool.execute(
f"""
UPDATE {table}
SET status = 'pending', worker_id = NULL, claimed_at = NULL, updated_at = now()
WHERE status = 'processing' AND worker_id = $1
""",
self._worker_id,
)
# Parse "UPDATE N" to get count
count = int(result.split()[-1]) if result else 0
total_count += count
except Exception as e:
logger.warning(f"Worker {self._worker_id} failed to recover tasks for schema {schema or 'public'}: {e}")
# Parse "UPDATE N" to get count
count = int(result.split()[-1]) if result else 0
total_count += count
if total_count > 0:
logger.info(f"Worker {self._worker_id} recovered {total_count} stale tasks from previous run")
@@ -426,59 +314,59 @@ class WorkerPoller:
async def run(self):
"""
Main polling loop with fire-and-forget task execution.
Main polling loop.
Continuously polls for pending tasks, spawns them as background tasks,
and immediately continues polling (up to slot limits).
Continuously polls for pending tasks, claims them, and executes them
until shutdown is signaled.
If tenant_extension is configured, dynamically discovers schemas on each poll.
"""
# Recover any tasks from a previous crash before starting
await self.recover_own_tasks()
logger.info(
f"Worker {self._worker_id} starting polling loop "
f"(max_slots={self._max_slots}, consolidation_max_slots={self._consolidation_max_slots})"
)
logger.info(f"Worker {self._worker_id} starting polling loop")
while not self._shutdown.is_set():
try:
# Claim a batch of tasks (respecting slot limits)
# Claim a batch of tasks (across all tenant schemas if configured)
tasks = await self.claim_batch()
if tasks:
# Log batch info
task_types: dict[str, int] = {}
schemas_seen: set[str | None] = set()
consolidation_count = 0
for task in tasks:
t = task.task_dict.get("type", "unknown")
op_type = task.task_dict.get("operation_type", "unknown")
task_types[t] = task_types.get(t, 0) + 1
schemas_seen.add(task.schema)
if op_type == "consolidation":
consolidation_count += 1
types_str = ", ".join(f"{k}:{v}" for k, v in task_types.items())
schemas_str = ", ".join(s or "public" for s in schemas_seen)
logger.info(
f"Worker {self._worker_id} claimed {len(tasks)} tasks "
f"({consolidation_count} consolidation): {types_str} (schemas: {schemas_str})"
f"Worker {self._worker_id} claimed {len(tasks)} tasks: {types_str} (schemas: {schemas_str})"
)
# Spawn tasks as background jobs (fire-and-forget)
for task in tasks:
await self.execute_task(task)
# Track in-flight tasks
async with self._in_flight_lock:
self._in_flight_count += len(tasks)
# Continue immediately to claim more tasks (if slots available)
continue
# No tasks claimed (either no pending tasks or slots full)
# Wait before polling again
try:
await asyncio.wait_for(
self._shutdown.wait(),
timeout=self._poll_interval_ms / 1000,
)
except asyncio.TimeoutError:
pass # Normal timeout, continue polling
# Execute tasks concurrently
try:
await asyncio.gather(
*[self.execute_task(task) for task in tasks],
return_exceptions=True,
)
finally:
async with self._in_flight_lock:
self._in_flight_count -= len(tasks)
else:
# No tasks found, wait before polling again
try:
await asyncio.wait_for(
self._shutdown.wait(),
timeout=self._poll_interval_ms / 1000,
)
except asyncio.TimeoutError:
pass # Normal timeout, continue polling
# Log progress stats periodically
await self._log_progress_if_due()
@@ -509,27 +397,15 @@ class WorkerPoller:
while asyncio.get_event_loop().time() - start_time < timeout:
async with self._in_flight_lock:
in_flight = self._in_flight_count
active_task_objects = [task_info[3] for task_info in self._active_tasks.values()]
if in_flight == 0:
logger.info(f"Worker {self._worker_id} graceful shutdown complete")
return
logger.info(f"Worker {self._worker_id} waiting for {in_flight} in-flight tasks")
await asyncio.sleep(0.5)
# Wait for at least one task to complete
if active_task_objects:
done, _ = await asyncio.wait(active_task_objects, timeout=0.5, return_when=asyncio.FIRST_COMPLETED)
else:
await asyncio.sleep(0.5)
logger.warning(f"Worker {self._worker_id} shutdown timeout after {timeout}s, cancelling remaining tasks")
# Cancel remaining tasks
async with self._in_flight_lock:
for operation_id, (_, _, _, bg_task) in list(self._active_tasks.items()):
if not bg_task.done():
bg_task.cancel()
logger.warning(f"Worker {self._worker_id} shutdown timeout after {timeout}s")
async def _log_progress_if_due(self):
"""Log progress stats every PROGRESS_LOG_INTERVAL seconds."""
@@ -540,19 +416,14 @@ class WorkerPoller:
self._last_progress_log = now
try:
# Get local active tasks
# Get local active tasks (this worker only)
async with self._in_flight_lock:
in_flight = self._in_flight_count
in_flight_by_type = dict(self._in_flight_by_type)
active_tasks = dict(self._active_tasks)
active_tasks = dict(self._active_tasks) # Copy to avoid holding lock
consolidation_count = in_flight_by_type.get("consolidation", 0)
available_slots = self._max_slots - in_flight
available_consolidation_slots = self._consolidation_max_slots - consolidation_count
# Build local processing breakdown
# Build local processing breakdown grouped by (op_type, bank_id)
task_groups: dict[tuple[str, str], int] = {}
for op_type, bank_id, _, _ in active_tasks.values():
for op_type, bank_id, _ in active_tasks.values():
key = (op_type, bank_id)
task_groups[key] = task_groups.get(key, 0) + 1
@@ -561,7 +432,7 @@ class WorkerPoller:
if len(processing_info) > 10:
processing_str += f" +{len(processing_info) - 10} more"
# Get global stats from DB
# Get global stats from DB across all schemas
schemas = await self._get_schemas()
global_pending = 0
all_worker_counts: dict[str, int] = {}
@@ -573,6 +444,7 @@ class WorkerPoller:
row = await conn.fetchrow(f"SELECT COUNT(*) as count FROM {table} WHERE status = 'pending'")
global_pending += row["count"] if row else 0
# Get processing breakdown by worker
worker_rows = await conn.fetch(
f"""
SELECT worker_id, COUNT(*) as count
@@ -585,6 +457,7 @@ class WorkerPoller:
wid = wr["worker_id"] or "unknown"
all_worker_counts[wid] = all_worker_counts.get(wid, 0) + wr["count"]
# Format other workers' processing counts
other_workers = []
for wid, cnt in all_worker_counts.items():
if wid != self._worker_id:
@@ -593,9 +466,7 @@ class WorkerPoller:
schemas_str = ", ".join(s or "public" for s in schemas)
logger.info(
f"[WORKER_STATS] worker={self._worker_id} "
f"slots={in_flight}/{self._max_slots} (consolidation={consolidation_count}/{self._consolidation_max_slots}) | "
f"available={available_slots} (consolidation={available_consolidation_slots}) | "
f"[WORKER_STATS] worker={self._worker_id} in_flight={in_flight} | "
f"global: pending={global_pending} (schemas: {schemas_str}) | "
f"others: {others_str} | "
f"my_active: {processing_str}"
+11 -11
View File
@@ -346,11 +346,11 @@ class TestConsolidationIntegration:
or when one directly updates another (e.g., location change).
Given:
- "Alex lives in Italy"
- "Alex moved to the US recently" (updates the living location)
- "Nicolò lives in Italy"
- "Nicolò moved to the US recently" (updates the living location)
The second fact should UPDATE the first, not create a separate observation.
But unrelated facts like "Alex works at Vectorize" should stay separate.
But unrelated facts like "Nicolò works at Vectorize" should stay separate.
"""
bank_id = f"test-consolidation-merge-{uuid.uuid4().hex[:8]}"
@@ -360,14 +360,14 @@ class TestConsolidationIntegration:
# Retain a memory about living location
await memory.retain_async(
bank_id=bank_id,
content="Alex lives in Italy.",
content="Nicolò lives in Italy.",
request_context=request_context,
)
# Retain an unrelated memory (different topic - should NOT merge)
await memory.retain_async(
bank_id=bank_id,
content="Alex works at Vectorize as an engineer.",
content="Nicolò works at Vectorize as an engineer.",
request_context=request_context,
)
@@ -384,7 +384,7 @@ class TestConsolidationIntegration:
# Add a memory that UPDATES the living location (should merge with first)
await memory.retain_async(
bank_id=bank_id,
content="Alex recently moved to the United States.",
content="Nicolò recently moved to the United States.",
request_context=request_context,
)
@@ -485,9 +485,9 @@ class TestConsolidationIntegration:
they should be merged into ONE observation that captures the change.
Example:
- "Alex loves pizza"
- "Alex hates pizza"
→ Should become: "Alex used to love pizza but now hates it" (or similar)
- "Nicolò loves pizza"
- "Nicolò hates pizza"
→ Should become: "Nicolò used to love pizza but now hates it" (or similar)
"""
bank_id = f"test-consolidation-contradict-{uuid.uuid4().hex[:8]}"
@@ -497,7 +497,7 @@ class TestConsolidationIntegration:
# Add initial fact
await memory.retain_async(
bank_id=bank_id,
content="Alex loves pizza.",
content="Nicolò loves pizza.",
request_context=request_context,
)
@@ -515,7 +515,7 @@ class TestConsolidationIntegration:
# Add contradicting fact (same person, same topic, opposite sentiment)
await memory.retain_async(
bank_id=bank_id,
content="Alex hates pizza.",
content="Nicolò hates pizza.",
request_context=request_context,
)
@@ -58,6 +58,7 @@ async def test_fact_extraction_basic_analysis(llm_config):
llm_config=llm_config,
agent_name="test-agent",
context="Friday Standup meeting",
extract_opinions=False,
)
duration = time.time() - start_time
-44
View File
@@ -97,47 +97,3 @@ def test_path_parsing_logic():
bank_id, remaining = parse_path("/my-bank/some/path")
assert bank_id == "my-bank"
assert remaining == "/some/path"
@pytest.mark.asyncio
async def test_api_key_context_variable():
"""Test that API key context variable works correctly."""
from hindsight_api.api.mcp import get_current_api_key, _current_api_key
# Initially None
assert get_current_api_key() is None
# Set and verify
token = _current_api_key.set("test-api-key-123")
try:
assert get_current_api_key() == "test-api-key-123"
finally:
_current_api_key.reset(token)
# Back to None after reset
assert get_current_api_key() is None
@pytest.mark.asyncio
async def test_mcp_tools_propagate_api_key(mock_memory):
"""Test that MCP tools propagate API key to RequestContext."""
from hindsight_api.api.mcp import create_mcp_server, _current_bank_id, _current_api_key
mcp_server = create_mcp_server(mock_memory)
tools = mcp_server._tool_manager._tools
# Set both bank_id and api_key context
bank_token = _current_bank_id.set("test-bank")
api_key_token = _current_api_key.set("test-bearer-token")
try:
retain_tool = tools["retain"]
result = await retain_tool.fn(content="test content", context="test_context", async_processing=False)
assert "successfully" in result.lower()
# Verify the memory was called with request_context containing api_key
mock_memory.retain_batch_async.assert_called_once()
call_kwargs = mock_memory.retain_batch_async.call_args.kwargs
assert call_kwargs["request_context"].api_key == "test-bearer-token"
finally:
_current_bank_id.reset(bank_token)
_current_api_key.reset(api_key_token)
+3 -3
View File
@@ -358,7 +358,7 @@ class TestLLMMetrics:
collector.record_llm_call(
provider="gemini",
model="gemini-pro",
scope="memory",
scope="entity_observation",
duration=2.0,
success=True,
)
@@ -369,11 +369,11 @@ class TestLLMMetrics:
assert call_args[0][0] == 1
assert call_args[0][1]["provider"] == "gemini"
assert call_args[0][1]["model"] == "gemini-pro"
assert call_args[0][1]["scope"] == "memory"
assert call_args[0][1]["scope"] == "entity_observation"
def test_record_llm_call_different_scopes(self, collector):
"""Test recording LLM calls with different scopes."""
scopes = ["memory", "reflect", "consolidation", "answer"]
scopes = ["memory", "reflect", "entity_observation", "answer"]
for scope in scopes:
collector.llm_duration.record.reset_mock()
+1
View File
@@ -469,6 +469,7 @@ async def test_mixed_language_entities(memory, request_context):
budget=Budget.MID,
max_tokens=1000,
fact_type=["world"],
include_entities=True,
request_context=request_context,
)
+242 -6
View File
@@ -91,13 +91,156 @@ async def test_entity_extraction_on_retain(memory, request_context):
await conn.execute("DELETE FROM entities WHERE bank_id = $1", bank_id)
@pytest.mark.asyncio
async def test_regenerate_entity_observations(memory, request_context):
"""
Test explicit regeneration of summary for an entity.
"""
bank_id = f"test_regen_obs_{datetime.now(timezone.utc).timestamp()}"
try:
# Store facts about an entity
await memory.retain_async(
bank_id=bank_id,
content="Sarah is a product manager who loves user research and data analysis.",
context="work info",
event_date=datetime(2024, 1, 15, tzinfo=timezone.utc),
request_context=request_context,
)
await memory.wait_for_background_tasks()
# Find the Sarah entity
pool = await memory._get_pool()
async with pool.acquire() as conn:
entity_row = await conn.fetchrow(
"""
SELECT id, canonical_name
FROM entities
WHERE bank_id = $1 AND LOWER(canonical_name) LIKE '%sarah%'
LIMIT 1
""",
bank_id
)
if entity_row:
entity_id = str(entity_row['id'])
entity_name = entity_row['canonical_name']
# Manually regenerate summary (via observations API for backwards compat)
created_ids = await memory.regenerate_entity_observations(
bank_id=bank_id,
entity_id=entity_id,
entity_name=entity_name,
request_context=request_context,
)
print(f"\n=== Regenerated Summary ===")
print(f"Created {len(created_ids)} summary for {entity_name}")
# Get entity state
state = await memory.get_entity_state(
bank_id, entity_id, entity_name, request_context=request_context
)
for obs in state.observations:
print(f" - {obs.text}")
# Verify summary was created
if len(created_ids) > 0:
assert len(state.observations) == 1, "Should have exactly 1 observation (the summary)"
print(f"Summary regenerated successfully")
else:
print(f"Note: No summary was regenerated")
else:
print(f"Note: No 'Sarah' entity was extracted")
finally:
# Cleanup
pool = await memory._get_pool()
async with pool.acquire() as conn:
await conn.execute("DELETE FROM memory_units WHERE bank_id = $1", bank_id)
await conn.execute("DELETE FROM entities WHERE bank_id = $1", bank_id)
@pytest.mark.asyncio
async def test_entity_state_retrieval(memory, request_context):
"""
Test retrieving entity state with facts.
"""
bank_id = f"test_entity_state_{datetime.now(timezone.utc).timestamp()}"
try:
# Store facts
await memory.retain_async(
bank_id=bank_id,
content="Alice works at Google as a senior software engineer.",
context="work info",
event_date=datetime(2024, 1, 15, tzinfo=timezone.utc),
request_context=request_context,
)
await memory.retain_async(
bank_id=bank_id,
content="Alice loves hiking and outdoor photography.",
context="hobbies",
event_date=datetime(2024, 1, 16, tzinfo=timezone.utc),
request_context=request_context,
)
# Find the Alice entity
pool = await memory._get_pool()
async with pool.acquire() as conn:
entity_row = await conn.fetchrow(
"""
SELECT id, canonical_name
FROM entities
WHERE bank_id = $1 AND LOWER(canonical_name) LIKE '%alice%'
LIMIT 1
""",
bank_id
)
assert entity_row is not None, "Alice entity should have been extracted"
entity_id = str(entity_row['id'])
entity_name = entity_row['canonical_name']
# Check fact count
async with pool.acquire() as conn:
fact_count = await conn.fetchval(
"SELECT COUNT(*) FROM unit_entities WHERE entity_id = $1",
entity_row['id']
)
print(f"\n=== Entity State Test ===")
print(f"Entity: {entity_name} (id: {entity_id})")
print(f"Linked facts: {fact_count}")
# Get entity state
state = await memory.get_entity_state(
bank_id, entity_id, entity_name, request_context=request_context
)
assert state.entity_id == entity_id
assert state.canonical_name == entity_name
print(f"Entity state retrieved successfully")
finally:
# Cleanup
pool = await memory._get_pool()
async with pool.acquire() as conn:
await conn.execute("DELETE FROM memory_units WHERE bank_id = $1", bank_id)
await conn.execute("DELETE FROM entities WHERE bank_id = $1", bank_id)
@pytest.mark.asyncio
async def test_search_with_include_entities(memory, request_context):
"""
Test that recall accepts include_entities parameter for backwards compatibility.
Test that search with include_entities=True returns entity information.
Note: Entity observations have been deprecated. This test verifies the parameter
is still accepted without errors.
This test verifies that:
1. Entities are extracted after retain
2. Entity info is returned in recall results with include_entities=True
"""
bank_id = f"test_search_ent_{datetime.now(timezone.utc).timestamp()}"
@@ -106,6 +249,10 @@ async def test_search_with_include_entities(memory, request_context):
contents = [
"Alice is a data scientist who works on recommendation systems at Netflix.",
"Alice presented her research at the ML conference last month.",
"Alice is an expert in deep learning and neural networks.",
"Alice graduated from Stanford with a PhD in Computer Science.",
"Alice leads a team of 5 data scientists at Netflix.",
"Alice published a paper on collaborative filtering algorithms.",
]
for i, content in enumerate(contents):
@@ -120,7 +267,7 @@ async def test_search_with_include_entities(memory, request_context):
# Wait for background tasks
await memory.wait_for_background_tasks()
# Search with include_entities=True (should be accepted for backwards compatibility)
# Search with include_entities=True
result = await memory.recall_async(
bank_id=bank_id,
query="What does Alice do?",
@@ -132,9 +279,98 @@ async def test_search_with_include_entities(memory, request_context):
request_context=request_context,
)
# Verify recall works
assert len(result.results) > 0, "Should find some facts"
print(f"\n=== Search Results ===")
print(f"Found {len(result.results)} facts")
for fact in result.results:
print(f" - {fact.text}")
if fact.entities:
print(f" Entities: {', '.join(fact.entities)}")
# Verify results
assert len(result.results) > 0, "Should find some facts"
# Check if entities are included in facts
facts_with_entities = [f for f in result.results if f.entities]
assert len(facts_with_entities) > 0, "Some facts should have entity information"
print(f"{len(facts_with_entities)} facts have entity information")
# Check if entity info is returned
if result.entities:
print(f"Entity info included for {len(result.entities)} entities")
# Verify Alice entity is in results
alice_found = False
for name, state in result.entities.items():
assert state.canonical_name == name, "Entity canonical_name should match key"
assert state.entity_id, "Entity should have an ID"
if "alice" in name.lower():
alice_found = True
print(f"Alice entity found: {name}")
assert alice_found, "Alice entity should be in recall results"
finally:
# Cleanup
pool = await memory._get_pool()
async with pool.acquire() as conn:
await conn.execute("DELETE FROM memory_units WHERE bank_id = $1", bank_id)
await conn.execute("DELETE FROM entities WHERE bank_id = $1", bank_id)
@pytest.mark.asyncio
async def test_get_entity_state(memory, request_context):
"""
Test getting the full state of an entity.
"""
bank_id = f"test_entity_state_{datetime.now(timezone.utc).timestamp()}"
try:
# Store facts
await memory.retain_async(
bank_id=bank_id,
content="Bob is a frontend developer who specializes in React and TypeScript.",
context="work info",
event_date=datetime(2024, 1, 15, tzinfo=timezone.utc),
request_context=request_context,
)
await memory.wait_for_background_tasks()
# Find entity
pool = await memory._get_pool()
async with pool.acquire() as conn:
entity_row = await conn.fetchrow(
"""
SELECT id, canonical_name
FROM entities
WHERE bank_id = $1 AND LOWER(canonical_name) LIKE '%bob%'
LIMIT 1
""",
bank_id
)
if entity_row:
entity_id = str(entity_row['id'])
entity_name = entity_row['canonical_name']
# Get entity state
state = await memory.get_entity_state(
bank_id=bank_id,
entity_id=entity_id,
entity_name=entity_name,
limit=10,
request_context=request_context,
)
print(f"\n=== Entity State for {entity_name} ===")
print(f"Entity ID: {state.entity_id}")
print(f"Canonical Name: {state.canonical_name}")
print(f"Observations: {len(state.observations)}")
for obs in state.observations:
print(f" - {obs.text}")
assert state.entity_id == entity_id, "Entity ID should match"
assert state.canonical_name == entity_name, "Canonical name should match"
finally:
# Cleanup
+3
View File
@@ -16,6 +16,7 @@ async def test_retain_with_chunks(memory, request_context):
Test that retain function:
1. Stores facts with associated chunks
2. Recall returns chunk_id for each fact
3. Recall with include_entities=True also works (for compatibility)
"""
bank_id = f"test_chunks_{datetime.now(timezone.utc).timestamp()}"
document_id = "test_doc_123"
@@ -55,6 +56,7 @@ async def test_retain_with_chunks(memory, request_context):
budget=Budget.LOW,
max_tokens=500,
fact_type=["world"], # Search for world facts
include_entities=False, # Disable entities for simpler test
include_chunks=True, # Enable chunks
max_chunk_tokens=8192,
request_context=request_context,
@@ -144,6 +146,7 @@ async def test_chunks_and_entities_follow_fact_order(memory, request_context):
budget=Budget.MID,
max_tokens=1000,
fact_type=["world"],
include_entities=True,
include_chunks=True,
max_chunk_tokens=8192,
request_context=request_context,
+126 -1
View File
@@ -1,5 +1,5 @@
"""
Test reflect (think) function.
Test think function for opinion generation and consistency.
"""
import pytest
from datetime import datetime, timezone
@@ -7,6 +7,131 @@ from hindsight_api.engine.memory_engine import Budget
from hindsight_api import RequestContext
@pytest.mark.asyncio
async def test_think_opinion_consistency(memory, request_context):
"""
Test that think function:
1. Generates an opinion
2. Stores the opinion in the database
3. Returns consistent response on subsequent calls with the same query
"""
bank_id = f"test_think_{datetime.now(timezone.utc).timestamp()}"
try:
# Store some initial facts to give context for opinion formation
await memory.retain_async(
bank_id=bank_id,
content="Alice is a software engineer who has worked on 5 major projects. She always delivers on time and writes clean, well-documented code.",
context="performance review",
event_date=datetime(2024, 1, 15, tzinfo=timezone.utc),
request_context=request_context,
)
await memory.retain_async(
bank_id=bank_id,
content="Bob recently joined the team. He missed his first deadline and his code had many bugs.",
context="performance review",
event_date=datetime(2024, 2, 1, tzinfo=timezone.utc),
request_context=request_context,
)
# First think call - should generate opinions
query = "Who is a more reliable engineer?"
result1 = await memory.reflect_async(
bank_id=bank_id,
query=query,
budget=Budget.LOW,
request_context=request_context,
)
print(f"\n=== First Think Call ===")
print(f"Answer: {result1.text}")
# Verify we got an answer
assert result1.text, "First think call should return an answer"
assert result1.based_on, "Should return based_on facts"
# Wait for background opinion processing tasks to complete
await memory.wait_for_background_tasks()
# Search for stored opinions to verify they were actually saved
pool = await memory._get_pool()
async with pool.acquire() as conn:
stored_opinions = await conn.fetch(
"""
SELECT id, text, confidence_score, fact_type
FROM memory_units
WHERE bank_id = $1 AND fact_type = 'opinion'
ORDER BY created_at DESC
""",
bank_id
)
print(f"\n=== Stored Opinions in Database ===")
print(f"Total opinions stored: {len(stored_opinions)}")
for op in stored_opinions:
print(f" - {op['text']} (confidence: {op['confidence_score']:.2f})")
# Verify opinions were actually written to database
# NOTE: Opinion extraction may not always detect opinions depending on the LLM response format
if len(stored_opinions) > 0:
assert all(op['fact_type'] == 'opinion' for op in stored_opinions), "All stored items should have fact_type='opinion'"
print(f"✓ Opinions were successfully stored in database")
else:
print(f"⚠ Note: No opinions were extracted/stored (this can happen if the LLM response format doesn't trigger opinion extraction)")
# Second think call - should use the stored opinions
result2 = await memory.reflect_async(
bank_id=bank_id,
query=query,
budget=Budget.LOW,
request_context=request_context,
)
print(f"\n=== Second Think Call ===")
print(f"Answer: {result2.text}")
print(f"Existing opinions used: {len(result2.based_on.get('opinion', []))}")
for opinion in result2.based_on.get('opinion', []):
print(f" - {opinion.text}")
# Verify second call also got an answer
assert result2.text, "Second think call should return an answer"
# Verify second call used the stored opinions (if any were stored)
if len(stored_opinions) > 0:
assert len(result2.based_on.get('opinion', [])) > 0, "Second call should retrieve stored opinions"
# The responses should be consistent (both should mention the same person as more reliable)
# We'll do a basic check that they're not contradictory
text1_lower = result1.text.lower()
text2_lower = result2.text.lower()
print(f"\n=== Consistency Check ===")
# Check if Alice is mentioned as more reliable in first response
if 'alice' in text1_lower and ('reliable' in text1_lower or 'better' in text1_lower):
print("First response favors Alice")
# Second response should also favor Alice (consistency)
assert 'alice' in text2_lower, "Second response should also mention Alice"
print("Second response also mentions Alice - CONSISTENT ✓")
# Check if Bob is mentioned
if 'bob' in text1_lower:
print("First response mentions Bob")
if 'bob' in text2_lower:
print("Second response also mentions Bob - CONSISTENT ✓")
print(f"\n✅ Test passed - opinions were formed, stored, and used consistently")
finally:
# Clean up agent data
try:
await memory.delete_bank(bank_id, request_context=request_context)
except Exception as e:
print(f"Warning: Error during cleanup: {e}")
@pytest.mark.asyncio
async def test_think_without_prior_context(memory, request_context):
"""
+180 -121
View File
@@ -1,9 +1,10 @@
"""
Test Vertex AI provider integration using native genai SDK.
Test Vertex AI provider integration including token refresh and API calls.
"""
import asyncio
import os
from unittest.mock import MagicMock, patch
from unittest.mock import MagicMock, Mock, patch
import pytest
@@ -11,38 +12,110 @@ import pytest
pytest.importorskip("google.auth")
@pytest.mark.asyncio
async def test_token_refresher_initialization():
"""Test token refresher initialization with mocked credentials."""
from hindsight_api.engine.vertexai_token_refresher import VertexAITokenRefresher
# Mock credentials
mock_credentials = MagicMock()
mock_credentials.token = "test-token-123"
mock_credentials.expiry = None
with patch("google.auth.transport.requests.Request"):
refresher = VertexAITokenRefresher(mock_credentials, "test-project", "us-central1")
# Verify token was fetched
assert refresher.get_token() == "test-token-123"
# Verify base URL is correctly formatted
expected_url = (
"https://us-central1-aiplatform.googleapis.com/v1beta1/"
"projects/test-project/locations/us-central1/endpoints/openapi"
)
assert refresher.get_base_url() == expected_url
@pytest.mark.asyncio
async def test_token_refresher_background_refresh():
"""Test that background refresh task starts and stops correctly."""
from hindsight_api.engine.vertexai_token_refresher import VertexAITokenRefresher
# Mock credentials
mock_credentials = MagicMock()
mock_credentials.token = "test-token-123"
mock_credentials.expiry = None
with patch("google.auth.transport.requests.Request"):
refresher = VertexAITokenRefresher(mock_credentials, "test-project", "us-central1")
# Start refresh task
refresher.start_refresh_task()
assert refresher._refresh_task is not None
assert not refresher._refresh_task.done()
# Stop refresh task
await refresher.stop()
assert refresher._refresh_task.done()
@pytest.mark.asyncio
async def test_token_refresher_thread_safety():
"""Test that token access is thread-safe."""
from hindsight_api.engine.vertexai_token_refresher import VertexAITokenRefresher
# Mock credentials
mock_credentials = MagicMock()
mock_credentials.token = "test-token-123"
mock_credentials.expiry = None
with patch("google.auth.transport.requests.Request"):
refresher = VertexAITokenRefresher(mock_credentials, "test-project", "us-central1")
# Access token from multiple tasks concurrently
async def get_token_task():
return refresher.get_token()
results = await asyncio.gather(*[get_token_task() for _ in range(10)])
# All should return the same token
assert all(token == "test-token-123" for token in results)
@pytest.mark.asyncio
async def test_token_refresher_no_token_error():
"""Test that getting token without refresh raises error."""
from hindsight_api.engine.vertexai_token_refresher import VertexAITokenRefresher
# Mock credentials that fail to refresh
mock_credentials = MagicMock()
mock_credentials.token = None
with patch("google.auth.transport.requests.Request") as mock_request:
mock_request.side_effect = Exception("Refresh failed")
with pytest.raises(Exception, match="Refresh failed"):
VertexAITokenRefresher(mock_credentials, "test-project", "us-central1")
def test_llm_wrapper_vertexai_missing_dependency():
"""Test error when google-auth is not available and service account key is set."""
"""Test error when google-auth is not available."""
from hindsight_api.engine import llm_wrapper
# VERTEXAI_AVAILABLE only matters when a service account key is provided
# Temporarily disable Vertex AI availability
original_available = llm_wrapper.VERTEXAI_AVAILABLE
try:
llm_wrapper.VERTEXAI_AVAILABLE = False
with patch.dict(
os.environ,
{
"HINDSIGHT_API_LLM_VERTEXAI_PROJECT_ID": "test-project",
"HINDSIGHT_API_LLM_VERTEXAI_SERVICE_ACCOUNT_KEY": "/path/to/key.json",
},
clear=False,
):
from hindsight_api.config import clear_config_cache
with pytest.raises(ValueError, match="google-auth"):
from hindsight_api.engine.llm_wrapper import LLMProvider
clear_config_cache()
with pytest.raises(ValueError, match="google-auth"):
from hindsight_api.engine.llm_wrapper import LLMProvider
LLMProvider(
provider="vertexai",
api_key="",
base_url="",
model="google/gemini-2.0-flash-001",
)
clear_config_cache()
LLMProvider(
provider="vertexai",
api_key="",
base_url="",
model="google/gemini-2.0-flash-001",
)
finally:
llm_wrapper.VERTEXAI_AVAILABLE = original_available
@@ -50,6 +123,7 @@ def test_llm_wrapper_vertexai_missing_dependency():
def test_llm_wrapper_vertexai_missing_project_id():
"""Test error when project ID is not configured."""
with patch.dict(os.environ, {"HINDSIGHT_API_LLM_VERTEXAI_PROJECT_ID": ""}, clear=False):
# Clear config cache to reload from env
from hindsight_api.config import clear_config_cache
clear_config_cache()
@@ -64,52 +138,58 @@ def test_llm_wrapper_vertexai_missing_project_id():
model="google/gemini-2.0-flash-001",
)
# Restore config cache
clear_config_cache()
def test_llm_wrapper_vertexai_adc_auth():
"""Test Vertex AI with ADC authentication creates native genai client."""
@pytest.mark.asyncio
async def test_llm_wrapper_vertexai_adc_auth():
"""Test Vertex AI with ADC authentication (mocked)."""
from hindsight_api.engine.llm_wrapper import LLMProvider
mock_credentials = MagicMock()
mock_credentials.token = "test-token-adc"
mock_credentials.expiry = None
with patch.dict(
os.environ,
{"HINDSIGHT_API_LLM_VERTEXAI_PROJECT_ID": "test-project"},
clear=False,
):
# Clear config cache to reload from env
from hindsight_api.config import clear_config_cache
clear_config_cache()
# genai.Client handles ADC internally — just verify it creates the client
with patch("google.genai.Client") as mock_client_cls:
mock_client_cls.return_value = MagicMock()
with patch("google.auth.default", return_value=(mock_credentials, "test-project")):
with patch("google.auth.transport.requests.Request"):
provider = LLMProvider(
provider="vertexai",
api_key="",
base_url="",
model="google/gemini-2.0-flash-001",
)
provider = LLMProvider(
provider="vertexai",
api_key="",
base_url="",
model="google/gemini-2.0-flash-001",
)
assert provider.provider == "vertexai"
assert provider._vertexai_refresher is not None
assert "aiplatform.googleapis.com" in provider.base_url
assert provider.provider == "vertexai"
assert provider.model == "gemini-2.0-flash-001" # google/ prefix stripped
assert provider._gemini_client is not None
# Verify genai.Client was called with vertexai=True
mock_client_cls.assert_called_once_with(
vertexai=True,
project="test-project",
location="us-central1",
)
# Cleanup
await provider.cleanup()
# Restore config cache
clear_config_cache()
def test_llm_wrapper_vertexai_sa_auth():
"""Test Vertex AI with service account authentication passes credentials to genai client."""
@pytest.mark.asyncio
async def test_llm_wrapper_vertexai_sa_auth():
"""Test Vertex AI with service account authentication (mocked)."""
from hindsight_api.engine.llm_wrapper import LLMProvider
import google.auth.exceptions
mock_credentials = MagicMock()
mock_credentials.token = "test-token-sa"
mock_credentials.expiry = None
with patch.dict(
os.environ,
@@ -119,91 +199,69 @@ def test_llm_wrapper_vertexai_sa_auth():
},
clear=False,
):
# Clear config cache to reload from env
from hindsight_api.config import clear_config_cache
clear_config_cache()
# Mock ADC failure, SA success
with patch(
"google.oauth2.service_account.Credentials.from_service_account_file",
return_value=mock_credentials,
"google.auth.default",
side_effect=google.auth.exceptions.DefaultCredentialsError("ADC not available"),
):
with patch("google.genai.Client") as mock_client_cls:
mock_client_cls.return_value = MagicMock()
with patch(
"google.oauth2.service_account.Credentials.from_service_account_file",
return_value=mock_credentials,
):
with patch("google.auth.transport.requests.Request"):
provider = LLMProvider(
provider="vertexai",
api_key="",
base_url="",
model="google/gemini-2.0-flash-001",
)
provider = LLMProvider(
assert provider.provider == "vertexai"
assert provider._vertexai_refresher is not None
# Cleanup
await provider.cleanup()
# Restore config cache
clear_config_cache()
@pytest.mark.asyncio
async def test_llm_wrapper_vertexai_auth_failure():
"""Test Vertex AI with both ADC and SA auth failing."""
import google.auth.exceptions
with patch.dict(
os.environ,
{"HINDSIGHT_API_LLM_VERTEXAI_PROJECT_ID": "test-project"},
clear=False,
):
# Clear config cache to reload from env
from hindsight_api.config import clear_config_cache
clear_config_cache()
# Mock both ADC and SA failures
with patch(
"google.auth.default",
side_effect=google.auth.exceptions.DefaultCredentialsError("ADC failed"),
):
with pytest.raises(ValueError, match="authentication failed"):
from hindsight_api.engine.llm_wrapper import LLMProvider
LLMProvider(
provider="vertexai",
api_key="",
base_url="",
model="google/gemini-2.0-flash-001",
)
assert provider.provider == "vertexai"
assert provider._gemini_client is not None
# Verify credentials were passed to genai.Client
mock_client_cls.assert_called_once_with(
vertexai=True,
project="test-project",
location="us-central1",
credentials=mock_credentials,
)
clear_config_cache()
def test_llm_wrapper_vertexai_strips_google_prefix():
"""Test that google/ prefix is stripped from model name for native SDK."""
from hindsight_api.engine.llm_wrapper import LLMProvider
with patch.dict(
os.environ,
{"HINDSIGHT_API_LLM_VERTEXAI_PROJECT_ID": "test-project"},
clear=False,
):
from hindsight_api.config import clear_config_cache
clear_config_cache()
with patch("google.genai.Client") as mock_client_cls:
mock_client_cls.return_value = MagicMock()
provider = LLMProvider(
provider="vertexai",
api_key="",
base_url="",
model="google/gemini-2.0-flash-lite-001",
)
assert provider.model == "gemini-2.0-flash-lite-001"
clear_config_cache()
def test_llm_wrapper_vertexai_no_prefix_model():
"""Test that model without google/ prefix is unchanged."""
from hindsight_api.engine.llm_wrapper import LLMProvider
with patch.dict(
os.environ,
{"HINDSIGHT_API_LLM_VERTEXAI_PROJECT_ID": "test-project"},
clear=False,
):
from hindsight_api.config import clear_config_cache
clear_config_cache()
with patch("google.genai.Client") as mock_client_cls:
mock_client_cls.return_value = MagicMock()
provider = LLMProvider(
provider="vertexai",
api_key="",
base_url="",
model="gemini-2.0-flash-001",
)
assert provider.model == "gemini-2.0-flash-001"
# Restore config cache
clear_config_cache()
@@ -241,4 +299,5 @@ async def test_vertexai_integration_actual_api():
assert len(response) > 0
finally:
# Cleanup
await provider.cleanup()
+15 -212
View File
@@ -156,6 +156,7 @@ class TestWorkerPoller:
pool=pool,
worker_id="test-worker-1",
executor=mock_executor,
batch_size=10,
)
claimed = await poller.claim_batch()
@@ -176,8 +177,8 @@ class TestWorkerPoller:
assert row["worker_id"] == "test-worker-1"
@pytest.mark.asyncio
async def test_claim_batch_respects_max_slots(self, pool, clean_operations):
"""Test that claim_batch respects the max_slots limit."""
async def test_claim_batch_respects_batch_size(self, pool, clean_operations):
"""Test that claim_batch respects the batch_size limit."""
from hindsight_api.worker import WorkerPoller
# Create 10 pending tasks
@@ -195,11 +196,12 @@ class TestWorkerPoller:
payload,
)
# Claim with batch_size=3
poller = WorkerPoller(
pool=pool,
worker_id="test-worker-1",
executor=lambda x: None,
max_slots=3, # Limit to 3 concurrent tasks
batch_size=3,
)
claimed = await poller.claim_batch()
@@ -236,14 +238,11 @@ class TestWorkerPoller:
executor=mock_executor,
)
# Execute the task (fire-and-forget)
# Execute the task
task_dict = json.loads(payload)
claimed_task = ClaimedTask(operation_id=str(op_id), task_dict=task_dict, schema=None)
await poller.execute_task(claimed_task)
# Wait for background task to complete
completed = await poller.wait_for_active_tasks(timeout=5.0)
assert completed, "Task did not complete within timeout"
assert len(executed) == 1
# Verify task is marked as completed
@@ -284,15 +283,11 @@ class TestWorkerPoller:
max_retries=3,
)
# Execute (should fail and retry) - fire-and-forget
# Execute (should fail and retry)
task_dict = json.loads(payload)
claimed_task = ClaimedTask(operation_id=str(op_id), task_dict=task_dict, schema=None)
await poller.execute_task(claimed_task)
# Wait for background task to complete
completed = await poller.wait_for_active_tasks(timeout=5.0)
assert completed, "Task did not complete within timeout"
# Verify task is back to pending with incremented retry_count
row = await pool.fetchrow(
"SELECT status, retry_count, worker_id FROM async_operations WHERE operation_id = $1",
@@ -332,15 +327,11 @@ class TestWorkerPoller:
max_retries=3,
)
# Execute (should fail permanently) - fire-and-forget
# Execute (should fail permanently)
task_dict = json.loads(payload)
claimed_task = ClaimedTask(operation_id=str(op_id), task_dict=task_dict, schema=None)
await poller.execute_task(claimed_task)
# Wait for background task to complete
completed = await poller.wait_for_active_tasks(timeout=5.0)
assert completed, "Task did not complete within timeout"
# Verify task is marked as failed
row = await pool.fetchrow(
"SELECT status, error_message FROM async_operations WHERE operation_id = $1",
@@ -397,6 +388,7 @@ class TestWorkerPoller:
pool=pool,
worker_id="test-worker-1",
executor=lambda x: None,
batch_size=10,
)
claimed = await poller.claim_batch()
@@ -448,6 +440,7 @@ class TestWorkerPoller:
pool=pool,
worker_id="test-worker-1",
executor=lambda x: None,
batch_size=10,
)
claimed = await poller.claim_batch()
@@ -614,6 +607,7 @@ class TestConcurrentWorkers:
pool=pool,
worker_id=worker_id,
executor=lambda x: None,
batch_size=5, # Each worker tries to claim 5
)
claimed = await poller.claim_batch()
workers_claimed[worker_id] = [task.operation_id for task in claimed]
@@ -686,6 +680,7 @@ class TestConcurrentWorkers:
pool=pool,
worker_id="new-worker",
executor=lambda x: None,
batch_size=10,
)
claimed = await poller.claim_batch()
@@ -884,6 +879,7 @@ class TestDynamicTenantDiscovery:
pool=pool,
worker_id="test-worker-1",
executor=lambda x: None,
batch_size=10,
tenant_extension=mock_extension,
)
@@ -950,6 +946,7 @@ class TestDynamicTenantDiscovery:
pool=pool,
worker_id="test-worker-1",
executor=lambda x: None,
batch_size=10,
tenant_extension=dynamic_extension,
)
@@ -1011,6 +1008,7 @@ class TestDynamicTenantDiscovery:
pool=pool,
worker_id="test-worker-1",
executor=lambda x: None,
batch_size=10,
)
claimed = await poller.claim_batch()
@@ -1019,198 +1017,3 @@ class TestDynamicTenantDiscovery:
# All tasks should have schema=None (public)
for task in claimed:
assert task.schema is None
async def test_worker_fire_and_forget_nonblocking(pool, clean_operations):
"""
Test that worker continues polling while tasks run (fire-and-forget pattern).
This test verifies the FIX: With the old blocking behavior, the worker would
wait for all tasks in a batch to complete before claiming more. This test
would FAIL with the old code because tasks 3-4 wouldn't be claimed until
tasks 1-2 complete. With fire-and-forget, tasks 3-4 are claimed immediately.
"""
from hindsight_api.worker.poller import WorkerPoller
task_started = {} # operation_id -> Event (set when task starts)
task_canfinish = {} # operation_id -> Event (wait before finishing)
async def blocking_executor(task_dict: dict):
op_id = task_dict["operation_id"]
# Signal that this task has started
started = asyncio.Event()
task_started[op_id] = started
started.set()
# Block until we're told to finish
finish = asyncio.Event()
task_canfinish[op_id] = finish
await finish.wait()
poller = WorkerPoller(
pool=pool,
worker_id="test-worker",
executor=blocking_executor,
poll_interval_ms=50, # Fast polling
max_slots=10,
consolidation_max_slots=2,
)
bank_id = f"test-worker-{uuid.uuid4().hex[:8]}"
# Submit initial 2 tasks
task_ids = []
for i in range(2):
op_id = uuid.uuid4()
task_ids.append(str(op_id))
payload = json.dumps({"type": "test", "operation_type": "retain", "operation_id": str(op_id), "bank_id": bank_id})
await pool.execute(
"""
INSERT INTO async_operations (operation_id, bank_id, operation_type, status, task_payload)
VALUES ($1, $2, 'retain', 'pending', $3::jsonb)
""",
op_id,
bank_id,
payload,
)
poll_task = asyncio.create_task(poller.run())
try:
# Wait for first 2 tasks to start executing (but not finish)
for i in range(100): # Try for up to 1 second
if len(task_started) >= 2:
break
await asyncio.sleep(0.01)
assert len(task_started) == 2, f"Expected 2 tasks started, got {len(task_started)}"
# Verify tasks are in_flight
async with poller._in_flight_lock:
assert poller._in_flight_count == 2
# NOW submit 2 more tasks WHILE the first 2 are still running
for i in range(2):
op_id = uuid.uuid4()
task_ids.append(str(op_id))
payload = json.dumps({"type": "test", "operation_type": "retain", "operation_id": str(op_id), "bank_id": bank_id})
await pool.execute(
"""
INSERT INTO async_operations (operation_id, bank_id, operation_type, status, task_payload)
VALUES ($1, $2, 'retain', 'pending', $3::jsonb)
""",
op_id,
bank_id,
payload,
)
# KEY ASSERTION: Worker should claim tasks 3-4 WITHOUT waiting for 1-2 to finish
# This would FAIL with the old blocking behavior
for i in range(100): # Try for up to 1 second
if len(task_started) >= 4:
break
await asyncio.sleep(0.01)
assert len(task_started) == 4, (
f"Fire-and-forget FAILED: Expected 4 tasks started, got {len(task_started)}. "
"This means the worker blocked waiting for the first batch to complete."
)
# Verify all 4 tasks are in-flight
async with poller._in_flight_lock:
assert poller._in_flight_count == 4
# Clean up: allow all tasks to finish
for event in task_canfinish.values():
event.set()
finally:
# Ensure cleanup
for event in task_canfinish.values():
event.set()
await poller.shutdown_graceful(timeout=2.0)
try:
await asyncio.wait_for(poll_task, timeout=1.0)
except asyncio.CancelledError:
pass
async def test_worker_slot_limits_enforced(pool, clean_operations):
"""Test that worker respects max_slots and won't exceed the limit."""
from hindsight_api.worker.poller import WorkerPoller
tasks_started = set()
task_events = {}
async def controlled_executor(task_dict: dict):
op_id = task_dict["operation_id"]
tasks_started.add(op_id)
event = asyncio.Event()
task_events[op_id] = event
await event.wait()
poller = WorkerPoller(
pool=pool,
worker_id="test-worker",
executor=controlled_executor,
poll_interval_ms=50,
max_slots=3, # Only allow 3 concurrent tasks
consolidation_max_slots=1,
)
# Submit 10 tasks
bank_id = f"test-worker-{uuid.uuid4().hex[:8]}"
for i in range(10):
op_id = uuid.uuid4()
payload = json.dumps({"type": "test", "operation_type": "retain", "operation_id": str(op_id), "bank_id": bank_id})
await pool.execute(
"""
INSERT INTO async_operations (operation_id, bank_id, operation_type, status, task_payload)
VALUES ($1, $2, 'retain', 'pending', $3::jsonb)
""",
op_id,
bank_id,
payload,
)
poll_task = asyncio.create_task(poller.run())
try:
# Wait for slots to fill
for i in range(100):
if len(tasks_started) >= 3:
break
await asyncio.sleep(0.01)
# Should have claimed exactly 3 tasks (slot limit)
assert len(tasks_started) == 3
# Wait to ensure no additional tasks are claimed
for i in range(30):
await asyncio.sleep(0.01)
assert len(tasks_started) == 3, "Worker exceeded slot limit!"
# Release tasks one by one and verify remaining are claimed
completed = 0
while completed < 10 and len(tasks_started) < 10:
# Release the next batch
events_to_release = list(task_events.values())[completed:completed+3]
for event in events_to_release:
event.set()
completed += len(events_to_release)
# Wait for new tasks to be claimed
for i in range(100):
if len(tasks_started) >= min(completed + 3, 10):
break
await asyncio.sleep(0.01)
assert len(tasks_started) == 10
finally:
for event in task_events.values():
event.set()
await poller.shutdown_graceful(timeout=2.0)
try:
await asyncio.wait_for(poll_task, timeout=1.0)
except asyncio.CancelledError:
pass
+3 -70
View File
@@ -500,8 +500,6 @@ pub fn delete(
pub fn consolidate(
client: &ApiClient,
bank_id: &str,
wait: bool,
poll_interval: u64,
verbose: bool,
output_format: OutputFormat,
) -> Result<()> {
@@ -519,82 +517,17 @@ pub fn consolidate(
match response {
Ok(result) => {
let operation_id = result.operation_id.clone();
if output_format == OutputFormat::Pretty {
ui::print_success("Consolidation triggered");
println!(" {} {}", ui::dim("Operation ID:"), operation_id);
println!(" {} {}", ui::dim("Operation ID:"), result.operation_id);
if result.deduplicated {
println!(" {} {}", ui::dim("Note:"), "Reusing existing pending consolidation task");
}
println!();
println!("{}", ui::dim("Use 'hindsight operation get' to check the operation status."));
} else {
output::print_output(&result, output_format)?;
}
if !wait {
if output_format == OutputFormat::Pretty {
println!();
println!("{}", ui::dim("Use --wait to poll for completion, or 'hindsight operation get' to check status."));
}
return Ok(());
}
// Poll for completion
if output_format == OutputFormat::Pretty {
println!();
println!("{}", ui::dim(&format!("Polling every {}s for completion...", poll_interval)));
}
let start = std::time::Instant::now();
loop {
std::thread::sleep(std::time::Duration::from_secs(poll_interval));
let elapsed = start.elapsed().as_secs();
let ops_result = client.list_operations(bank_id, verbose);
match ops_result {
Ok(ops) => {
// Find the operation by ID
let op = ops.operations.iter().find(|o| o.id == operation_id);
match op.map(|o| o.status.as_str()) {
Some("completed") => {
if output_format == OutputFormat::Pretty {
ui::print_success(&format!("Consolidation completed ({}s)", elapsed));
}
break;
}
Some("failed") => {
let error_msg = op
.and_then(|o| o.error_message.as_ref())
.map(|s| s.as_str())
.unwrap_or("Unknown error");
if output_format == OutputFormat::Pretty {
ui::print_error(&format!("Consolidation failed: {}", error_msg));
}
std::process::exit(1);
}
Some(status) => {
if output_format == OutputFormat::Pretty {
println!("{} ({}s elapsed)", status, elapsed);
}
}
None => {
if output_format == OutputFormat::Pretty {
ui::print_warning(&format!("Operation {} not found in list", operation_id));
}
break;
}
}
}
Err(e) => {
if output_format == OutputFormat::Pretty {
ui::print_error(&format!("Failed to check operation status: {}", e));
}
return Err(e);
}
}
}
Ok(())
}
Err(e) => Err(e),
-141
View File
@@ -1,6 +1,4 @@
use anyhow::Result;
use chrono::{Duration as ChronoDuration, NaiveDate, Utc};
use std::collections::BTreeMap;
use crate::api::ApiClient;
use crate::output::{self, OutputFormat};
use crate::ui;
@@ -9,17 +7,11 @@ pub fn list(
client: &ApiClient,
agent_id: &str,
query: Option<String>,
date: Option<String>,
limit: i32,
offset: i32,
verbose: bool,
output_format: OutputFormat,
) -> Result<()> {
// If date filter is provided, use the date-aware listing
if date.is_some() {
return list_with_date(client, agent_id, date.as_deref(), verbose, output_format);
}
let spinner = if output_format == OutputFormat::Pretty {
Some(ui::create_spinner("Fetching documents..."))
} else {
@@ -58,139 +50,6 @@ pub fn list(
}
}
/// List documents with date filtering
fn list_with_date(
client: &ApiClient,
bank_id: &str,
date_filter: Option<&str>,
verbose: bool,
output_format: OutputFormat,
) -> Result<()> {
let spinner = if output_format == OutputFormat::Pretty {
Some(ui::create_spinner("Fetching all documents..."))
} else {
None
};
// Fetch all documents with pagination
let all_docs = fetch_all_documents(client, bank_id, verbose)?;
if let Some(mut sp) = spinner {
sp.finish();
}
// Parse the date filter
let target_date = parse_date_filter(date_filter)?;
// Filter and group documents by date
let mut by_date: BTreeMap<String, Vec<serde_json::Value>> = BTreeMap::new();
let mut filtered_count = 0;
for doc in all_docs {
let created_at = doc.get("created_at")
.and_then(|v| v.as_str())
.unwrap_or("");
// Parse the date part (YYYY-MM-DD) from created_at
let doc_date = created_at.split('T').next().unwrap_or("");
// Apply date filter if specified
if let Some(ref target) = target_date {
let target_str = target.format("%Y-%m-%d").to_string();
if doc_date != target_str {
continue;
}
}
filtered_count += 1;
by_date.entry(doc_date.to_string()).or_default().push(doc);
}
// Output
if output_format == OutputFormat::Pretty {
let filter_desc = match date_filter {
None | Some("yesterday") => "yesterday".to_string(),
Some("today") => "today".to_string(),
Some("all") => "all dates".to_string(),
Some(d) => d.to_string(),
};
ui::print_info(&format!(
"Documents for bank '{}' (filter: {}, showing: {})",
bank_id, filter_desc, filtered_count
));
println!();
// Show documents grouped by date (reverse order - newest first)
for (date_str, docs) in by_date.iter().rev() {
println!(" {} ({} documents)", date_str, docs.len());
for doc in docs {
let id = doc.get("id").and_then(|v| v.as_str()).unwrap_or("unknown");
let mem_count = doc.get("memory_unit_count").and_then(|v| v.as_i64()).unwrap_or(0);
println!(" - {} ({} memories)", id, mem_count);
}
println!();
}
} else {
// JSON/YAML output - convert to a list structure
let output: Vec<serde_json::Value> = by_date.values().flatten().cloned().collect();
output::print_output(&output, output_format)?;
}
Ok(())
}
/// Fetch all documents with pagination
fn fetch_all_documents(
client: &ApiClient,
bank_id: &str,
verbose: bool,
) -> Result<Vec<serde_json::Value>> {
let mut all_docs = Vec::new();
let mut offset = 0;
let limit = 500;
loop {
let response = client.list_documents(bank_id, None, Some(limit), Some(offset), verbose)?;
if response.items.is_empty() {
break;
}
// Convert Map<String, Value> to Value for each item
for item in response.items {
all_docs.push(serde_json::Value::Object(item));
}
offset += limit;
// Check if we've fetched everything
if all_docs.len() >= response.total as usize {
break;
}
}
Ok(all_docs)
}
/// Parse date filter string into a NaiveDate
fn parse_date_filter(filter: Option<&str>) -> Result<Option<NaiveDate>> {
match filter {
None | Some("yesterday") => {
// Default to yesterday
Ok(Some(Utc::now().date_naive() - ChronoDuration::days(1)))
}
Some("today") => Ok(Some(Utc::now().date_naive())),
Some("all") => Ok(None), // No filtering
Some(date_str) => {
// Try to parse as YYYY-MM-DD
NaiveDate::parse_from_str(date_str, "%Y-%m-%d")
.map(Some)
.map_err(|e| anyhow::anyhow!("Invalid date format '{}': {}. Use YYYY-MM-DD, 'yesterday', 'today', or 'all'", date_str, e))
}
}
}
pub fn get(
client: &ApiClient,
agent_id: &str,
+4 -16
View File
@@ -260,14 +260,6 @@ enum BankCommands {
Consolidate {
/// Bank ID
bank_id: String,
/// Wait for consolidation to complete (poll for status)
#[arg(long)]
wait: bool,
/// Poll interval in seconds (only used with --wait)
#[arg(long, default_value = "10")]
poll_interval: u64,
},
/// Clear all observations for a bank
@@ -449,10 +441,6 @@ enum DocumentCommands {
#[arg(short = 'q', long)]
query: Option<String>,
/// Filter by date (yesterday, today, YYYY-MM-DD, or all)
#[arg(short = 'd', long)]
date: Option<String>,
/// Maximum number of results
#[arg(short = 'l', long, default_value = "100")]
limit: i32,
@@ -766,8 +754,8 @@ fn run() -> Result<()> {
BankCommands::Delete { bank_id, yes } => {
commands::bank::delete(&client, &bank_id, yes, verbose, output_format)
}
BankCommands::Consolidate { bank_id, wait, poll_interval } => {
commands::bank::consolidate(&client, &bank_id, wait, poll_interval, verbose, output_format)
BankCommands::Consolidate { bank_id } => {
commands::bank::consolidate(&client, &bank_id, verbose, output_format)
}
BankCommands::ClearObservations { bank_id, yes } => {
commands::bank::clear_observations(&client, &bank_id, yes, verbose, output_format)
@@ -804,8 +792,8 @@ fn run() -> Result<()> {
// Document commands
Commands::Document(doc_cmd) => match doc_cmd {
DocumentCommands::List { bank_id, query, date, limit, offset } => {
commands::document::list(&client, &bank_id, query, date, limit, offset, verbose, output_format)
DocumentCommands::List { bank_id, query, limit, offset } => {
commands::document::list(&client, &bank_id, query, limit, offset, verbose, output_format)
}
DocumentCommands::Get { bank_id, document_id } => {
commands::document::get(&client, &bank_id, &document_id, verbose, output_format)
@@ -1644,7 +1644,7 @@ class MemoryApi:
) -> RecallResponse:
"""Recall memory
Recall memory using semantic similarity and spreading activation. The type parameter is optional and must be one of: - `world`: General knowledge about people, places, events, and things that happen - `experience`: Memories about experience, conversations, actions taken, and tasks performed
Recall memory using semantic similarity and spreading activation. The type parameter is optional and must be one of: - `world`: General knowledge about people, places, events, and things that happen - `experience`: Memories about experience, conversations, actions taken, and tasks performed - `opinion`: The bank's formed beliefs, perspectives, and viewpoints Set `include_entities=true` to get entity observations alongside recall results.
:param bank_id: (required)
:type bank_id: str
@@ -1720,7 +1720,7 @@ class MemoryApi:
) -> ApiResponse[RecallResponse]:
"""Recall memory
Recall memory using semantic similarity and spreading activation. The type parameter is optional and must be one of: - `world`: General knowledge about people, places, events, and things that happen - `experience`: Memories about experience, conversations, actions taken, and tasks performed
Recall memory using semantic similarity and spreading activation. The type parameter is optional and must be one of: - `world`: General knowledge about people, places, events, and things that happen - `experience`: Memories about experience, conversations, actions taken, and tasks performed - `opinion`: The bank's formed beliefs, perspectives, and viewpoints Set `include_entities=true` to get entity observations alongside recall results.
:param bank_id: (required)
:type bank_id: str
@@ -1796,7 +1796,7 @@ class MemoryApi:
) -> RESTResponseType:
"""Recall memory
Recall memory using semantic similarity and spreading activation. The type parameter is optional and must be one of: - `world`: General knowledge about people, places, events, and things that happen - `experience`: Memories about experience, conversations, actions taken, and tasks performed
Recall memory using semantic similarity and spreading activation. The type parameter is optional and must be one of: - `world`: General knowledge about people, places, events, and things that happen - `experience`: Memories about experience, conversations, actions taken, and tasks performed - `opinion`: The bank's formed beliefs, perspectives, and viewpoints Set `include_entities=true` to get entity observations alongside recall results.
:param bank_id: (required)
:type bank_id: str
@@ -1950,7 +1950,7 @@ class MemoryApi:
) -> ReflectResponse:
"""Reflect and generate answer
Reflect and formulate an answer using bank identity, world facts, and opinions. This endpoint: 1. Retrieves experience (conversations and events) 2. Retrieves world facts relevant to the query 3. Retrieves existing opinions (bank's perspectives) 4. Uses LLM to formulate a contextual answer 5. Returns plain text answer and the facts used
Reflect and formulate an answer using bank identity, world facts, and opinions. This endpoint: 1. Retrieves experience (conversations and events) 2. Retrieves world facts relevant to the query 3. Retrieves existing opinions (bank's perspectives) 4. Uses LLM to formulate a contextual answer 5. Extracts and stores any new opinions formed 6. Returns plain text answer, the facts used, and new opinions
:param bank_id: (required)
:type bank_id: str
@@ -2026,7 +2026,7 @@ class MemoryApi:
) -> ApiResponse[ReflectResponse]:
"""Reflect and generate answer
Reflect and formulate an answer using bank identity, world facts, and opinions. This endpoint: 1. Retrieves experience (conversations and events) 2. Retrieves world facts relevant to the query 3. Retrieves existing opinions (bank's perspectives) 4. Uses LLM to formulate a contextual answer 5. Returns plain text answer and the facts used
Reflect and formulate an answer using bank identity, world facts, and opinions. This endpoint: 1. Retrieves experience (conversations and events) 2. Retrieves world facts relevant to the query 3. Retrieves existing opinions (bank's perspectives) 4. Uses LLM to formulate a contextual answer 5. Extracts and stores any new opinions formed 6. Returns plain text answer, the facts used, and new opinions
:param bank_id: (required)
:type bank_id: str
@@ -2102,7 +2102,7 @@ class MemoryApi:
) -> RESTResponseType:
"""Reflect and generate answer
Reflect and formulate an answer using bank identity, world facts, and opinions. This endpoint: 1. Retrieves experience (conversations and events) 2. Retrieves world facts relevant to the query 3. Retrieves existing opinions (bank's perspectives) 4. Uses LLM to formulate a contextual answer 5. Returns plain text answer and the facts used
Reflect and formulate an answer using bank identity, world facts, and opinions. This endpoint: 1. Retrieves experience (conversations and events) 2. Retrieves world facts relevant to the query 3. Retrieves existing opinions (bank's perspectives) 4. Uses LLM to formulate a contextual answer 5. Extracts and stores any new opinions formed 6. Returns plain text answer, the facts used, and new opinions
:param bank_id: (required)
:type bank_id: str
@@ -236,6 +236,9 @@ export const getMemory = <ThrowOnError extends boolean = false>(
* The type parameter is optional and must be one of:
* - `world`: General knowledge about people, places, events, and things that happen
* - `experience`: Memories about experience, conversations, actions taken, and tasks performed
* - `opinion`: The bank's formed beliefs, perspectives, and viewpoints
*
* Set `include_entities=true` to get entity observations alongside recall results.
*/
export const recallMemories = <ThrowOnError extends boolean = false>(
options: Options<RecallMemoriesData, ThrowOnError>,
@@ -263,7 +266,8 @@ export const recallMemories = <ThrowOnError extends boolean = false>(
* 2. Retrieves world facts relevant to the query
* 3. Retrieves existing opinions (bank's perspectives)
* 4. Uses LLM to formulate a contextual answer
* 5. Returns plain text answer and the facts used
* 5. Extracts and stores any new opinions formed
* 6. Returns plain text answer, the facts used, and new opinions
*/
export const reflect = <ThrowOnError extends boolean = false>(
options: Options<ReflectData, ThrowOnError>,
@@ -1178,7 +1178,7 @@ export type RecallRequest = {
/**
* Types
*
* List of fact types to recall: 'world', 'experience', 'observation'. Defaults to world and experience if not specified.
* List of fact types to recall: 'world', 'experience', 'observation'. Defaults to world and experience if not specified. Note: 'opinion' is accepted but ignored (opinions are excluded from recall).
*/
types?: Array<string> | null;
budget?: Budget;
@@ -1,5 +1,6 @@
import { NextResponse } from "next/server";
import { DATAPLANE_URL, getDataplaneHeaders } from "@/lib/hindsight-client";
const DATAPLANE_URL = process.env.HINDSIGHT_CP_DATAPLANE_API_URL || "http://localhost:8888";
export async function GET(
request: Request,
@@ -14,7 +15,7 @@ export async function GET(
const response = await fetch(
`${DATAPLANE_URL}/v1/default/banks/${bankId}/directives/${directiveId}`,
{ method: "GET", headers: getDataplaneHeaders() }
{ method: "GET" }
);
if (!response.ok) {
@@ -48,7 +49,7 @@ export async function PATCH(
`${DATAPLANE_URL}/v1/default/banks/${bankId}/directives/${directiveId}`,
{
method: "PATCH",
headers: getDataplaneHeaders({ "Content-Type": "application/json" }),
headers: { "Content-Type": "application/json" },
body: JSON.stringify(body),
}
);
@@ -83,7 +84,7 @@ export async function DELETE(
const response = await fetch(
`${DATAPLANE_URL}/v1/default/banks/${bankId}/directives/${directiveId}`,
{ method: "DELETE", headers: getDataplaneHeaders() }
{ method: "DELETE" }
);
if (!response.ok) {
@@ -1,5 +1,6 @@
import { NextResponse } from "next/server";
import { DATAPLANE_URL, getDataplaneHeaders } from "@/lib/hindsight-client";
const DATAPLANE_URL = process.env.HINDSIGHT_CP_DATAPLANE_API_URL || "http://localhost:8888";
export async function GET(request: Request, { params }: { params: Promise<{ bankId: string }> }) {
try {
@@ -21,7 +22,7 @@ export async function GET(request: Request, { params }: { params: Promise<{ bank
}
const url = `${DATAPLANE_URL}/v1/default/banks/${bankId}/directives${queryParams.toString() ? `?${queryParams}` : ""}`;
const response = await fetch(url, { method: "GET", headers: getDataplaneHeaders() });
const response = await fetch(url, { method: "GET" });
if (!response.ok) {
const errorText = await response.text();
@@ -49,7 +50,7 @@ export async function POST(request: Request, { params }: { params: Promise<{ ban
const response = await fetch(`${DATAPLANE_URL}/v1/default/banks/${bankId}/directives`, {
method: "POST",
headers: getDataplaneHeaders({ "Content-Type": "application/json" }),
headers: { "Content-Type": "application/json" },
body: JSON.stringify(body),
});
@@ -1,5 +1,6 @@
import { NextResponse } from "next/server";
import { DATAPLANE_URL, getDataplaneHeaders } from "@/lib/hindsight-client";
const DATAPLANE_URL = process.env.HINDSIGHT_CP_DATAPLANE_API_URL || "http://localhost:8888";
export async function POST(
request: Request,
@@ -17,7 +18,7 @@ export async function POST(
const response = await fetch(
`${DATAPLANE_URL}/v1/default/banks/${bankId}/mental-models/${mentalModelId}/refresh`,
{ method: "POST", headers: getDataplaneHeaders() }
{ method: "POST" }
);
if (!response.ok) {
@@ -1,5 +1,6 @@
import { NextResponse } from "next/server";
import { DATAPLANE_URL, getDataplaneHeaders } from "@/lib/hindsight-client";
const DATAPLANE_URL = process.env.HINDSIGHT_CP_DATAPLANE_API_URL || "http://localhost:8888";
export async function GET(
request: Request,
@@ -17,7 +18,7 @@ export async function GET(
const response = await fetch(
`${DATAPLANE_URL}/v1/default/banks/${bankId}/mental-models/${mentalModelId}`,
{ method: "GET", headers: getDataplaneHeaders() }
{ method: "GET" }
);
if (!response.ok) {
@@ -57,7 +58,7 @@ export async function PATCH(
`${DATAPLANE_URL}/v1/default/banks/${bankId}/mental-models/${mentalModelId}`,
{
method: "PATCH",
headers: getDataplaneHeaders({ "Content-Type": "application/json" }),
headers: { "Content-Type": "application/json" },
body: JSON.stringify(body),
}
);
@@ -95,7 +96,7 @@ export async function DELETE(
const response = await fetch(
`${DATAPLANE_URL}/v1/default/banks/${bankId}/mental-models/${mentalModelId}`,
{ method: "DELETE", headers: getDataplaneHeaders() }
{ method: "DELETE" }
);
if (!response.ok) {
@@ -1,5 +1,6 @@
import { NextResponse } from "next/server";
import { DATAPLANE_URL, getDataplaneHeaders } from "@/lib/hindsight-client";
const DATAPLANE_URL = process.env.HINDSIGHT_CP_DATAPLANE_API_URL || "http://localhost:8888";
export async function GET(request: Request, { params }: { params: Promise<{ bankId: string }> }) {
try {
@@ -21,7 +22,7 @@ export async function GET(request: Request, { params }: { params: Promise<{ bank
}
const url = `${DATAPLANE_URL}/v1/default/banks/${bankId}/mental-models${queryParams.toString() ? `?${queryParams}` : ""}`;
const response = await fetch(url, { method: "GET", headers: getDataplaneHeaders() });
const response = await fetch(url, { method: "GET" });
if (!response.ok) {
const errorText = await response.text();
@@ -52,7 +53,7 @@ export async function POST(request: Request, { params }: { params: Promise<{ ban
const response = await fetch(`${DATAPLANE_URL}/v1/default/banks/${bankId}/mental-models`, {
method: "POST",
headers: getDataplaneHeaders({ "Content-Type": "application/json" }),
headers: { "Content-Type": "application/json" },
body: JSON.stringify(body),
});
@@ -1,5 +1,6 @@
import { NextResponse } from "next/server";
import { DATAPLANE_URL, getDataplaneHeaders } from "@/lib/hindsight-client";
const DATAPLANE_URL = process.env.HINDSIGHT_CP_DATAPLANE_API_URL || "http://localhost:8888";
export async function GET(
request: Request,
@@ -14,7 +15,7 @@ export async function GET(
const response = await fetch(
`${DATAPLANE_URL}/v1/default/banks/${bankId}/mental-models/${modelId}`,
{ method: "GET", headers: getDataplaneHeaders() }
{ method: "GET" }
);
if (!response.ok) {
@@ -24,14 +24,6 @@ export async function GET(request: NextRequest) {
},
});
if (response.error || !response.data) {
console.error("Graph API error:", response.error);
return NextResponse.json(
{ error: response.error || "Failed to fetch graph data" },
{ status: 500 }
);
}
return NextResponse.json(response.data, { status: 200 });
} catch (error) {
console.error("Error fetching graph data:", error);
@@ -1,6 +1,5 @@
import { NextResponse } from "next/server";
import { createClient, createConfig, sdk } from "@vectorize-io/hindsight-client";
import { getDataplaneHeaders } from "@/lib/hindsight-client";
const HEALTH_CHECK_TIMEOUT_MS = 3000;
@@ -28,7 +27,6 @@ export async function GET() {
createConfig({
baseUrl: dataplaneUrl,
signal: controller.signal,
headers: getDataplaneHeaders(),
})
);
@@ -1,5 +1,6 @@
import { NextRequest, NextResponse } from "next/server";
import { DATAPLANE_URL, getDataplaneHeaders } from "@/lib/hindsight-client";
const DATAPLANE_URL = process.env.HINDSIGHT_CP_DATAPLANE_API_URL || "http://localhost:8888";
export async function GET(
request: NextRequest,
@@ -18,7 +19,9 @@ export async function GET(
`${DATAPLANE_URL}/v1/default/banks/${bankId}/memories/${memoryId}`,
{
method: "GET",
headers: getDataplaneHeaders({ "Content-Type": "application/json" }),
headers: {
"Content-Type": "application/json",
},
}
);
@@ -130,21 +130,21 @@ const TRAIT_LABELS: Record<
skepticism: {
label: "Skepticism",
shortLabel: "S",
description: "How skeptical vs trusting when forming observations",
description: "How skeptical vs trusting when forming opinions",
lowLabel: "Trusting",
highLabel: "Skeptical",
},
literalism: {
label: "Literalism",
shortLabel: "L",
description: "How literally to interpret information when forming observations",
description: "How literally to interpret information when forming opinions",
lowLabel: "Flexible",
highLabel: "Literal",
},
empathy: {
label: "Empathy",
shortLabel: "E",
description: "How much to consider emotional context when forming observations",
description: "How much to consider emotional context when forming opinions",
lowLabel: "Detached",
highLabel: "Empathetic",
},
@@ -718,9 +718,7 @@ export function BankProfileView() {
<Brain className="w-5 h-5 text-primary" />
Disposition Profile
</CardTitle>
<CardDescription>
Traits that shape how observations are formed via Reflect
</CardDescription>
<CardDescription>Traits that shape how opinions are formed via Reflect</CardDescription>
</CardHeader>
<CardContent>
{profile && (
@@ -368,6 +368,31 @@ export function ThinkView() {
</CardContent>
</Card>
{/* New Opinions Formed */}
{result.new_opinions && result.new_opinions.length > 0 && (
<Card className="border-green-200 dark:border-green-800">
<CardHeader className="bg-green-50 dark:bg-green-950">
<CardTitle className="flex items-center gap-2">
<Sparkles className="w-5 h-5" />
New Opinions Formed
</CardTitle>
<CardDescription>New beliefs generated from this interaction</CardDescription>
</CardHeader>
<CardContent className="pt-6">
<div className="space-y-3">
{result.new_opinions.map((opinion: any, i: number) => (
<div key={i} className="p-3 bg-muted rounded-lg border border-border">
<div className="font-semibold text-foreground">{opinion.text}</div>
<div className="text-sm text-muted-foreground mt-1">
Confidence: {opinion.confidence?.toFixed(2)}
</div>
</div>
))}
</div>
</CardContent>
</Card>
)}
{/* Directive */}
<Card className="border-blue-200 dark:border-blue-800">
<CardHeader className="py-4">
@@ -5,37 +5,17 @@
import { HindsightClient, createClient, createConfig, sdk } from "@vectorize-io/hindsight-client";
export const DATAPLANE_URL = process.env.HINDSIGHT_CP_DATAPLANE_API_URL || "http://localhost:8888";
const DATAPLANE_API_KEY = process.env.HINDSIGHT_CP_DATAPLANE_API_KEY || "";
/**
* Auth headers for direct fetch calls to the dataplane API.
*/
export function getDataplaneHeaders(extra?: Record<string, string>): Record<string, string> {
const headers: Record<string, string> = { ...extra };
if (DATAPLANE_API_KEY) {
headers["Authorization"] = `Bearer ${DATAPLANE_API_KEY}`;
}
return headers;
}
const DATAPLANE_URL = process.env.HINDSIGHT_CP_DATAPLANE_API_URL || "http://localhost:8888";
/**
* High-level client with convenience methods
*/
export const hindsightClient = new HindsightClient({
baseUrl: DATAPLANE_URL,
apiKey: DATAPLANE_API_KEY || undefined,
});
export const hindsightClient = new HindsightClient({ baseUrl: DATAPLANE_URL });
/**
* Low-level client for direct SDK access
*/
export const lowLevelClient = createClient(
createConfig({
baseUrl: DATAPLANE_URL,
headers: DATAPLANE_API_KEY ? { Authorization: `Bearer ${DATAPLANE_API_KEY}` } : undefined,
})
);
export const lowLevelClient = createClient(createConfig({ baseUrl: DATAPLANE_URL }));
/**
* Export SDK functions for direct API access
@@ -474,6 +474,7 @@ Observations are consolidated knowledge synthesized from facts.
| `HINDSIGHT_API_ENABLE_OBSERVATIONS` | Enable observation consolidation | `true` |
| `HINDSIGHT_API_CONSOLIDATION_BATCH_SIZE` | Memories to load per batch (internal optimization) | `50` |
| `HINDSIGHT_API_CONSOLIDATION_MAX_TOKENS` | Max tokens for recall when finding related observations during consolidation | `1024` |
| `HINDSIGHT_API_RETAIN_OBSERVATIONS_ASYNC` | Run observation generation asynchronously (after retain completes) | `false` |
### Reflect
@@ -504,10 +505,9 @@ Configuration for background task processing. By default, the API processes task
| `HINDSIGHT_API_WORKER_ENABLED` | Enable internal worker in API process | `true` |
| `HINDSIGHT_API_WORKER_ID` | Unique worker identifier | hostname |
| `HINDSIGHT_API_WORKER_POLL_INTERVAL_MS` | Database polling interval in milliseconds | `500` |
| `HINDSIGHT_API_WORKER_BATCH_SIZE` | Tasks to claim per poll cycle | `10` |
| `HINDSIGHT_API_WORKER_MAX_RETRIES` | Max retries before marking task failed | `3` |
| `HINDSIGHT_API_WORKER_HTTP_PORT` | HTTP port for worker metrics/health (worker CLI only) | `8889` |
| `HINDSIGHT_API_WORKER_MAX_SLOTS` | Maximum concurrent tasks per worker | `10` |
| `HINDSIGHT_API_WORKER_CONSOLIDATION_MAX_SLOTS` | Maximum concurrent consolidation tasks per worker | `2` |
### Performance Optimization
+3 -3
View File
@@ -343,7 +343,7 @@
"Memory"
],
"summary": "Recall memory",
"description": "Recall memory using semantic similarity and spreading activation.\n\nThe type parameter is optional and must be one of:\n- `world`: General knowledge about people, places, events, and things that happen\n- `experience`: Memories about experience, conversations, actions taken, and tasks performed",
"description": "Recall memory using semantic similarity and spreading activation.\n\nThe type parameter is optional and must be one of:\n- `world`: General knowledge about people, places, events, and things that happen\n- `experience`: Memories about experience, conversations, actions taken, and tasks performed\n- `opinion`: The bank's formed beliefs, perspectives, and viewpoints\n\nSet `include_entities=true` to get entity observations alongside recall results.",
"operationId": "recall_memories",
"parameters": [
{
@@ -412,7 +412,7 @@
"Memory"
],
"summary": "Reflect and generate answer",
"description": "Reflect and formulate an answer using bank identity, world facts, and opinions.\n\nThis endpoint:\n1. Retrieves experience (conversations and events)\n2. Retrieves world facts relevant to the query\n3. Retrieves existing opinions (bank's perspectives)\n4. Uses LLM to formulate a contextual answer\n5. Returns plain text answer and the facts used",
"description": "Reflect and formulate an answer using bank identity, world facts, and opinions.\n\nThis endpoint:\n1. Retrieves experience (conversations and events)\n2. Retrieves world facts relevant to the query\n3. Retrieves existing opinions (bank's perspectives)\n4. Uses LLM to formulate a contextual answer\n5. Extracts and stores any new opinions formed\n6. Returns plain text answer, the facts used, and new opinions",
"operationId": "reflect",
"parameters": [
{
@@ -4948,7 +4948,7 @@
}
],
"title": "Types",
"description": "List of fact types to recall: 'world', 'experience', 'observation'. Defaults to world and experience if not specified."
"description": "List of fact types to recall: 'world', 'experience', 'observation'. Defaults to world and experience if not specified. Note: 'opinion' is accepted but ignored (opinions are excluded from recall)."
},
"budget": {
"$ref": "#/components/schemas/Budget",
Generated
+1403 -1179
View File
File diff suppressed because it is too large Load Diff