fc941d5cae
* feat: add HINDSIGHT_API_DEFAULT_BANK_TEMPLATE env var Server-level default bank template applied automatically to every newly-created bank. Holds an inline JSON BankTemplateManifest with the same shape as the /import endpoint body. Fields set by the template become per-bank overrides so they take precedence over equivalent HINDSIGHT_API_* env defaults. The template is applied once on first creation and never reapplied, so user overrides via PATCH /config are never clobbered. Malformed manifests are logged and ignored so a broken server-level setting cannot wedge bank creation. * chore: regenerate docs skill * test: update async_retain test mock for renamed bank_profile helper
365 lines
11 KiB
Python
365 lines
11 KiB
Python
"""
|
|
bank profile utilities for disposition and mission management.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
import uuid
|
|
from typing import TypedDict
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
from ...config import get_config
|
|
from ..db_utils import acquire_with_retry
|
|
from ..memory_engine import fq_table, get_current_schema
|
|
from ..response_models import DispositionTraits
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Fact types that get per-bank partial vector indexes, mapped to their 4-char index suffix.
|
|
_BANK_INDEX_FACT_TYPES: dict[str, str] = {
|
|
"world": "worl",
|
|
"experience": "expr",
|
|
"observation": "obsv",
|
|
}
|
|
|
|
|
|
def _bank_index_name(ft: str, internal_id: str) -> str:
|
|
"""Deterministic, schema-safe vector index name for a (bank, fact_type) pair.
|
|
|
|
Uses the first 16 hex chars of internal_id (8 bytes of entropy) — unique
|
|
enough in practice, fits comfortably within PostgreSQL's 63-char identifier limit.
|
|
"""
|
|
uid = str(internal_id).replace("-", "")[:16]
|
|
return f"idx_mu_emb_{_BANK_INDEX_FACT_TYPES[ft]}_{uid}"
|
|
|
|
|
|
def _vector_index_clause() -> str:
|
|
"""Return the USING clause for vector index creation based on the configured extension."""
|
|
ext = get_config().vector_extension
|
|
if ext == "pgvectorscale":
|
|
return "USING diskann (embedding vector_cosine_ops) WITH (num_neighbors = 50)"
|
|
elif ext == "vchord":
|
|
return "USING vchordrq (embedding vector_l2_ops)"
|
|
else: # pgvector (default)
|
|
return "USING hnsw (embedding vector_cosine_ops)"
|
|
|
|
|
|
async def create_bank_vector_indexes(conn, bank_id: str, internal_id: str) -> None:
|
|
"""Create per-(bank, fact_type) partial vector indexes for a newly created bank.
|
|
|
|
Respects the HINDSIGHT_API_VECTOR_EXTENSION config to use the appropriate
|
|
index type (HNSW for pgvector, DiskANN for pgvectorscale, vchordrq for vchord).
|
|
|
|
Called immediately after the bank row is first inserted. Safe on empty banks
|
|
(index build is instant). Idempotent via CREATE INDEX IF NOT EXISTS.
|
|
bank_id is escaped for SQL literal safety (apostrophes doubled).
|
|
"""
|
|
table = fq_table("memory_units")
|
|
escaped = bank_id.replace("'", "''")
|
|
using_clause = _vector_index_clause()
|
|
for ft in _BANK_INDEX_FACT_TYPES:
|
|
idx = _bank_index_name(ft, internal_id)
|
|
await conn.execute(
|
|
f"CREATE INDEX IF NOT EXISTS {idx} "
|
|
f"ON {table} {using_clause} "
|
|
f"WHERE fact_type = '{ft}' AND bank_id = '{escaped}'"
|
|
)
|
|
|
|
|
|
async def drop_bank_vector_indexes(conn, internal_id: str) -> None:
|
|
"""Drop per-(bank, fact_type) partial vector indexes for a bank being deleted.
|
|
|
|
Called before the bank row is deleted so internal_id is still known.
|
|
Idempotent via DROP INDEX IF EXISTS.
|
|
"""
|
|
schema = get_current_schema()
|
|
for ft in _BANK_INDEX_FACT_TYPES:
|
|
idx = _bank_index_name(ft, internal_id)
|
|
await conn.execute(f"DROP INDEX IF EXISTS {schema}.{idx}")
|
|
|
|
|
|
DEFAULT_DISPOSITION = {
|
|
"skepticism": 3,
|
|
"literalism": 3,
|
|
"empathy": 3,
|
|
}
|
|
|
|
|
|
class BankProfile(TypedDict):
|
|
"""Type for bank profile data."""
|
|
|
|
name: str
|
|
disposition: DispositionTraits
|
|
mission: str
|
|
|
|
|
|
class MissionMergeResponse(BaseModel):
|
|
"""LLM response for mission merge."""
|
|
|
|
mission: str = Field(description="Merged mission in first person perspective")
|
|
|
|
|
|
async def get_bank_profile(pool, bank_id: str) -> BankProfile:
|
|
"""
|
|
Get bank profile (name, disposition + mission).
|
|
Auto-creates bank with default values if not exists.
|
|
|
|
Args:
|
|
pool: Database connection pool
|
|
bank_id: bank IDentifier
|
|
|
|
Returns:
|
|
BankProfile with name, typed DispositionTraits, and mission
|
|
"""
|
|
profile, _ = await get_or_create_bank_profile(pool, bank_id)
|
|
return profile
|
|
|
|
|
|
async def get_or_create_bank_profile(pool, bank_id: str) -> tuple[BankProfile, bool]:
|
|
"""
|
|
Get bank profile, auto-creating with defaults if it doesn't exist.
|
|
|
|
Same as get_bank_profile, but also returns a flag indicating whether the
|
|
bank was freshly created on this call. Used by the memory engine to apply
|
|
the HINDSIGHT_API_DEFAULT_BANK_TEMPLATE hook on first bank creation.
|
|
|
|
Returns:
|
|
Tuple of (BankProfile, created) where created is True if the bank
|
|
did not exist before this call.
|
|
"""
|
|
async with acquire_with_retry(pool) as conn:
|
|
# Try to get existing bank
|
|
row = await conn.fetchrow(
|
|
f"""
|
|
SELECT name, disposition, mission
|
|
FROM {fq_table("banks")} WHERE bank_id = $1
|
|
""",
|
|
bank_id,
|
|
)
|
|
|
|
if row:
|
|
# asyncpg returns JSONB as a string, so parse it
|
|
disposition_data = row["disposition"]
|
|
if isinstance(disposition_data, str):
|
|
disposition_data = json.loads(disposition_data)
|
|
|
|
return (
|
|
BankProfile(
|
|
name=row["name"],
|
|
disposition=DispositionTraits(**disposition_data),
|
|
mission=row["mission"] or "",
|
|
),
|
|
False,
|
|
)
|
|
|
|
# Bank doesn't exist, create with defaults.
|
|
# Generate internal_id here so we control the value and can use it
|
|
# immediately for vector index creation without a RETURNING round-trip.
|
|
internal_id = uuid.uuid4()
|
|
inserted = await conn.fetchval(
|
|
f"""
|
|
INSERT INTO {fq_table("banks")} (bank_id, name, disposition, mission, internal_id)
|
|
VALUES ($1, $2, $3::jsonb, $4, $5)
|
|
ON CONFLICT (bank_id) DO NOTHING
|
|
RETURNING bank_id
|
|
""",
|
|
bank_id,
|
|
bank_id, # Default name is the bank_id
|
|
json.dumps(DEFAULT_DISPOSITION),
|
|
"",
|
|
internal_id,
|
|
)
|
|
|
|
created = inserted is not None
|
|
if created:
|
|
# Fresh insert — create per-bank vector indexes (instant on empty bank)
|
|
await create_bank_vector_indexes(conn, bank_id, str(internal_id))
|
|
|
|
return (
|
|
BankProfile(name=bank_id, disposition=DispositionTraits(**DEFAULT_DISPOSITION), mission=""),
|
|
created,
|
|
)
|
|
|
|
|
|
async def update_bank_disposition(pool, bank_id: str, disposition: dict[str, int]) -> None:
|
|
"""
|
|
Update bank disposition traits.
|
|
|
|
Args:
|
|
pool: Database connection pool
|
|
bank_id: bank IDentifier
|
|
disposition: Dict with skepticism, literalism, empathy (all 1-5)
|
|
"""
|
|
# Ensure bank exists first
|
|
await get_bank_profile(pool, bank_id)
|
|
|
|
async with acquire_with_retry(pool) as conn:
|
|
await conn.execute(
|
|
f"""
|
|
UPDATE {fq_table("banks")}
|
|
SET disposition = $2::jsonb,
|
|
updated_at = NOW()
|
|
WHERE bank_id = $1
|
|
""",
|
|
bank_id,
|
|
json.dumps(disposition),
|
|
)
|
|
|
|
|
|
async def set_bank_mission(pool, bank_id: str, mission: str) -> None:
|
|
"""
|
|
Set bank mission (replacing any existing mission).
|
|
|
|
Args:
|
|
pool: Database connection pool
|
|
bank_id: bank IDentifier
|
|
mission: The mission text
|
|
"""
|
|
# Ensure bank exists first
|
|
await get_bank_profile(pool, bank_id)
|
|
|
|
async with acquire_with_retry(pool) as conn:
|
|
await conn.execute(
|
|
f"""
|
|
UPDATE {fq_table("banks")}
|
|
SET mission = $2,
|
|
updated_at = NOW()
|
|
WHERE bank_id = $1
|
|
""",
|
|
bank_id,
|
|
mission,
|
|
)
|
|
|
|
|
|
async def merge_bank_mission(pool, llm_config, bank_id: str, new_info: str) -> dict:
|
|
"""
|
|
Merge new mission information with existing mission using LLM.
|
|
Normalizes to first person ("I") and resolves conflicts.
|
|
|
|
Args:
|
|
pool: Database connection pool
|
|
llm_config: LLM configuration for mission merging
|
|
bank_id: bank IDentifier
|
|
new_info: New mission information to add/merge
|
|
|
|
Returns:
|
|
Dict with 'mission' (str) key
|
|
"""
|
|
# Get current profile
|
|
profile = await get_bank_profile(pool, bank_id)
|
|
current_mission = profile["mission"]
|
|
|
|
# Use LLM to merge missions
|
|
result = await _llm_merge_mission(llm_config, current_mission, new_info)
|
|
|
|
merged_mission = result["mission"]
|
|
|
|
# Update in database
|
|
async with acquire_with_retry(pool) as conn:
|
|
await conn.execute(
|
|
f"""
|
|
UPDATE {fq_table("banks")}
|
|
SET mission = $2,
|
|
updated_at = NOW()
|
|
WHERE bank_id = $1
|
|
""",
|
|
bank_id,
|
|
merged_mission,
|
|
)
|
|
|
|
return {"mission": merged_mission}
|
|
|
|
|
|
async def _llm_merge_mission(llm_config, current: str, new_info: str) -> dict:
|
|
"""
|
|
Use LLM to intelligently merge mission information.
|
|
|
|
Args:
|
|
llm_config: LLM configuration to use
|
|
current: Current mission text
|
|
new_info: New information to merge
|
|
|
|
Returns:
|
|
Dict with 'mission' (str) key
|
|
"""
|
|
prompt = f"""You are helping maintain an agent's mission statement.
|
|
|
|
Current mission: {current if current else "(empty)"}
|
|
|
|
New information to add: {new_info}
|
|
|
|
Instructions:
|
|
1. Merge the new information with the current mission
|
|
2. If there are conflicts, the NEW information overwrites the old
|
|
3. Keep additions that don't conflict
|
|
4. Output in FIRST PERSON ("I") perspective
|
|
5. Be concise - keep it under 500 characters
|
|
6. Return ONLY the merged mission text, no explanations
|
|
|
|
Merged mission:"""
|
|
|
|
try:
|
|
messages = [{"role": "user", "content": prompt}]
|
|
|
|
content = await llm_config.call(
|
|
messages=messages, scope="bank_mission", temperature=0.3, max_completion_tokens=8192
|
|
)
|
|
|
|
logger.info(f"LLM response for mission merge (first 500 chars): {content[:500]}")
|
|
|
|
merged = content.strip()
|
|
if not merged or merged.lower() in ["(empty)", "none", "n/a"]:
|
|
merged = new_info if new_info else ""
|
|
return {"mission": merged}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error merging mission with LLM: {e}")
|
|
# Fallback: just append new info
|
|
if current:
|
|
merged = f"{current} {new_info}".strip()
|
|
else:
|
|
merged = new_info
|
|
|
|
return {"mission": merged}
|
|
|
|
|
|
async def list_banks(pool) -> list:
|
|
"""
|
|
List all banks in the system.
|
|
|
|
Args:
|
|
pool: Database connection pool
|
|
|
|
Returns:
|
|
List of dicts with bank_id, name, disposition, mission, created_at, updated_at
|
|
"""
|
|
async with acquire_with_retry(pool) as conn:
|
|
rows = await conn.fetch(
|
|
f"""
|
|
SELECT bank_id, name, disposition, mission, created_at, updated_at
|
|
FROM {fq_table("banks")}
|
|
ORDER BY updated_at DESC
|
|
"""
|
|
)
|
|
|
|
result = []
|
|
for row in rows:
|
|
# asyncpg returns JSONB as a string, so parse it
|
|
disposition_data = row["disposition"]
|
|
if isinstance(disposition_data, str):
|
|
disposition_data = json.loads(disposition_data)
|
|
|
|
result.append(
|
|
{
|
|
"bank_id": row["bank_id"],
|
|
"name": row["name"],
|
|
"disposition": disposition_data,
|
|
"mission": row["mission"] or "",
|
|
"created_at": row["created_at"].isoformat() if row["created_at"] else None,
|
|
"updated_at": row["updated_at"].isoformat() if row["updated_at"] else None,
|
|
}
|
|
)
|
|
|
|
return result
|