RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat: make deriver modular

This commit is contained in:
Rajat Ahuja 2025-11-10 13:53:19 -05:00
parent e7d7a32d7a
commit efef9a5d54
15 changed files with 2265 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

629
scripts/backfill_data_2.3.2.py Normal file
View File

@ -0,0 +1,629 @@
"""Generate 500k test messages, message embeddings, and documents using raw psycopg3.
This script creates:
- 1 workspace
- 3 peers
- 3 sessions (with peers added to sessions via session_peers table)
- 500k messages distributed randomly across peers and sessions
- 500k message embeddings (one per message with constant zero vector)
- 1 collection per peer (3 total)
- 500k documents distributed across the collections
Usage:
python manual_load.py [schema_name]
If schema_name is not provided, uses DB_SCHEMA environment variable.
Environment variables required:
DB_CONNECTION_URI: PostgreSQL connection URI
DB_SCHEMA: Database schema to use (optional if passed as argument)
"""
import os
import random
import sys
import time
import psycopg
from dotenv import load_dotenv
from faker import Faker
from nanoid import generate as generate_nanoid
load_dotenv()
# Configuration
NUM_MESSAGES = 500_000
NUM_DOCUMENTS = 500_000
BATCH_SIZE = 10_000 # Batch size for inserts (messages and documents)
EMBEDDING_BATCH_SIZE = 1_000 # Smaller batch size for embeddings (they're large)
MAX_RETRIES = 3 # Number of retries for failed operations
RETRY_DELAY = 5 # Seconds to wait before retrying
NUM_PEERS = 3
NUM_SESSIONS = 3
# Load environment variables
DB_CONNECTION_URI = os.getenv("DB_CONNECTION_URI")
# Get schema from command line argument or environment variable
if len(sys.argv) > 1:
DB_SCHEMA = sys.argv[1]
else:
DB_SCHEMA = os.getenv("DB_SCHEMA")
if not DB_CONNECTION_URI:
raise ValueError("DB_CONNECTION_URI environment variable is required")
if not DB_SCHEMA:
raise ValueError(
"DB_SCHEMA must be provided as command line argument or environment variable"
)
if DB_CONNECTION_URI.startswith("postgresql+psycopg://"):
DB_CONNECTION_URI = DB_CONNECTION_URI.replace(
"postgresql+psycopg://", "postgresql://"
)
elif DB_CONNECTION_URI.startswith("postgres+psycopg://"):
DB_CONNECTION_URI = DB_CONNECTION_URI.replace(
"postgres+psycopg://", "postgresql://"
)
# Initialize Faker
fake = Faker()
# Constant embedding - all zeros for speed (1536 dimensions for OpenAI embeddings)
CONSTANT_EMBEDDING = str([0.0] * 1536)
# Generate IDs
WORKSPACE_ID = generate_nanoid()
WORKSPACE_NAME = "load_test_workspace"
PEER_IDS = [generate_nanoid() for _ in range(NUM_PEERS)]
PEER_NAMES = [f"load_peer_{i + 1}" for i in range(NUM_PEERS)]
SESSION_IDS = [generate_nanoid() for _ in range(NUM_SESSIONS)]
SESSION_NAMES = [f"load_session_{i + 1}" for i in range(NUM_SESSIONS)]
COLLECTION_IDS = [generate_nanoid() for _ in range(NUM_PEERS)]
COLLECTION_NAMES = [f"load_peer_{i + 1}_load_peer_{i + 1}" for i in range(NUM_PEERS)]
def check_existing_data(conn, schema):
"""Check what data already exists in the database."""
print("\nChecking existing data...")
with conn.cursor() as cur:
# Check workspace
cur.execute(
f'SELECT COUNT(*) FROM "{schema}"."workspaces" WHERE name = %s',
(WORKSPACE_NAME,),
)
workspace_exists = cur.fetchone()[0] > 0
# Check peers
cur.execute(
f'SELECT COUNT(*) FROM "{schema}"."peers" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_peers = cur.fetchone()[0]
# Check sessions
cur.execute(
f'SELECT COUNT(*) FROM "{schema}"."sessions" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_sessions = cur.fetchone()[0]
# Check collections
cur.execute(
f'SELECT COUNT(*) FROM "{schema}"."collections" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_collections = cur.fetchone()[0]
# Check messages
cur.execute(
f'SELECT COUNT(*) FROM "{schema}"."messages" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_messages = cur.fetchone()[0]
# Check message embeddings
cur.execute(
f'SELECT COUNT(*) FROM "{schema}"."message_embeddings" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_embeddings = cur.fetchone()[0]
# Check documents
cur.execute(
f'SELECT COUNT(*) FROM "{schema}"."documents" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_documents = cur.fetchone()[0]
print(f" Workspace exists: {workspace_exists}")
print(f" Existing peers: {existing_peers}/{NUM_PEERS}")
print(f" Existing sessions: {existing_sessions}/{NUM_SESSIONS}")
print(f" Existing collections: {existing_collections}/{NUM_PEERS}")
print(f" Existing messages: {existing_messages:,}/{NUM_MESSAGES:,}")
print(f" Existing message embeddings: {existing_embeddings:,}/{NUM_MESSAGES:,}")
print(f" Existing documents: {existing_documents:,}/{NUM_DOCUMENTS:,}")
return {
"workspace_exists": workspace_exists,
"existing_peers": existing_peers,
"existing_sessions": existing_sessions,
"existing_collections": existing_collections,
"existing_messages": existing_messages,
"existing_embeddings": existing_embeddings,
"existing_documents": existing_documents,
}
def create_entities(conn, schema, existing_data):
"""Create workspace, peers, sessions, and collections."""
print("\nCreating workspace, peers, sessions, and collections...")
with conn.cursor() as cur:
# Create workspace if not exists
if not existing_data["workspace_exists"]:
cur.execute(
f'INSERT INTO "{schema}"."workspaces" (id, name) VALUES (%s, %s)',
(WORKSPACE_ID, WORKSPACE_NAME),
)
print(f" ✓ Created workspace: {WORKSPACE_NAME}")
else:
print(f" → Workspace already exists: {WORKSPACE_NAME}")
# Create peers if needed
if existing_data["existing_peers"] < NUM_PEERS:
# Get existing peer names
cur.execute(
f'SELECT name FROM "{schema}"."peers" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_peer_names = {row[0] for row in cur.fetchall()}
for peer_id, peer_name in zip(PEER_IDS, PEER_NAMES, strict=False):
if peer_name not in existing_peer_names:
cur.execute(
f'INSERT INTO "{schema}"."peers" (id, name, workspace_name) VALUES (%s, %s, %s)',
(peer_id, peer_name, WORKSPACE_NAME),
)
new_peers = NUM_PEERS - existing_data["existing_peers"]
print(
f" ✓ Created {new_peers} new peers ({existing_data['existing_peers']} already existed)"
)
else:
print(f" → All {NUM_PEERS} peers already exist")
# Create sessions if needed
if existing_data["existing_sessions"] < NUM_SESSIONS:
# Get existing session names
cur.execute(
f'SELECT name FROM "{schema}"."sessions" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_session_names = {row[0] for row in cur.fetchall()}
for session_id, session_name in zip(
SESSION_IDS, SESSION_NAMES, strict=False
):
if session_name not in existing_session_names:
cur.execute(
f'INSERT INTO "{schema}"."sessions" (id, name, workspace_name) VALUES (%s, %s, %s)',
(session_id, session_name, WORKSPACE_NAME),
)
new_sessions = NUM_SESSIONS - existing_data["existing_sessions"]
print(
f" ✓ Created {new_sessions} new sessions ({existing_data['existing_sessions']} already existed)"
)
else:
print(f" → All {NUM_SESSIONS} sessions already exist")
# Add peers to sessions via session_peers table
# Check existing relationships
cur.execute(
f'SELECT COUNT(*) FROM "{schema}"."session_peers" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_relationships = cur.fetchone()[0]
expected_relationships = NUM_SESSIONS * NUM_PEERS
if existing_relationships < expected_relationships:
for session_name in SESSION_NAMES:
for peer_name in PEER_NAMES:
# Check if relationship exists
cur.execute(
f'''SELECT COUNT(*) FROM "{schema}"."session_peers"
WHERE workspace_name = %s AND session_name = %s AND peer_name = %s''',
(WORKSPACE_NAME, session_name, peer_name),
)
if cur.fetchone()[0] == 0:
cur.execute(
f'''INSERT INTO "{schema}"."session_peers"
(workspace_name, session_name, peer_name)
VALUES (%s, %s, %s)''',
(WORKSPACE_NAME, session_name, peer_name),
)
new_relationships = expected_relationships - existing_relationships
print(
f" ✓ Added {new_relationships} new peer-session relationships ({existing_relationships} already existed)"
)
else:
print(" → All peer-session relationships already exist")
# Create collections (one per peer) if needed
if existing_data["existing_collections"] < NUM_PEERS:
# Get existing collection names
cur.execute(
f'SELECT name FROM "{schema}"."collections" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_collection_names = {row[0] for row in cur.fetchall()}
for collection_id, collection_name, peer_name in zip(
COLLECTION_IDS, COLLECTION_NAMES, PEER_NAMES, strict=False
):
if collection_name not in existing_collection_names:
cur.execute(
f'INSERT INTO "{schema}"."collections" (id, name, peer_name, workspace_name) VALUES (%s, %s, %s, %s)',
(collection_id, collection_name, peer_name, WORKSPACE_NAME),
)
new_collections = NUM_PEERS - existing_data["existing_collections"]
print(
f" ✓ Created {new_collections} new collections ({existing_data['existing_collections']} already existed)"
)
else:
print(f" → All {NUM_PEERS} collections already exist")
conn.commit()
def generate_messages(conn, schema, existing_count):
"""Generate and insert messages up to NUM_MESSAGES total."""
messages_to_create = NUM_MESSAGES - existing_count
if messages_to_create <= 0:
print(f"\n→ All {NUM_MESSAGES:,} messages already exist, skipping...")
return []
print(
f"\nGenerating {messages_to_create:,} messages ({existing_count:,} already exist)..."
)
# Pre-generate all message data
print(" Generating message data...")
message_data = []
for i in range(messages_to_create):
public_id = generate_nanoid()
peer_name = random.choice(PEER_NAMES)
session_name = random.choice(SESSION_NAMES)
content = fake.text(max_nb_chars=200)
message_data.append(
{
"public_id": public_id,
"peer_name": peer_name,
"session_name": session_name,
"content": content,
"workspace_name": WORKSPACE_NAME,
}
)
# Insert in batches
print(f" Inserting messages in batches of {BATCH_SIZE:,}...")
total_batches = (messages_to_create + BATCH_SIZE - 1) // BATCH_SIZE
with conn.cursor() as cur:
for batch_num in range(total_batches):
batch_start = batch_num * BATCH_SIZE
batch_end = min(batch_start + BATCH_SIZE, messages_to_create)
batch = message_data[batch_start:batch_end]
# Use execute_batch for better performance
args = [
(
msg["public_id"],
msg["session_name"],
msg["content"],
msg["peer_name"],
msg["workspace_name"],
)
for msg in batch
]
cur.executemany(
f'''INSERT INTO "{schema}"."messages"
(public_id, session_name, content, peer_name, workspace_name)
VALUES (%s, %s, %s, %s, %s)''',
args,
)
conn.commit()
messages_created = existing_count + batch_end
progress = (messages_created / NUM_MESSAGES) * 100
print(
f" Batch {batch_num + 1}/{total_batches}: {messages_created:,}/{NUM_MESSAGES:,} ({progress:.1f}%)"
)
print(f" ✓ Successfully created {messages_to_create:,} new messages!")
return message_data
def generate_message_embeddings(conn, schema, message_data, existing_count):
"""Generate and insert message embeddings for messages that don't have them."""
embeddings_to_create = NUM_MESSAGES - existing_count
if embeddings_to_create <= 0:
print(f"\n→ All {NUM_MESSAGES:,} message embeddings already exist, skipping...")
return conn
print(
f"\nGenerating {embeddings_to_create:,} message embeddings ({existing_count:,} already exist)..."
)
# If we just created messages, use that data
if message_data:
print(" Using newly created message data...")
else:
# Fetch messages that don't have embeddings yet in chunks
# Using NOT EXISTS is much faster than LEFT JOIN for large tables
print(" Fetching messages without embeddings in chunks...")
message_data = []
chunk_size = 50_000 # Fetch in chunks to avoid timeout
total_fetched = 0
with conn.cursor() as cur:
while total_fetched < embeddings_to_create:
remaining = embeddings_to_create - total_fetched
fetch_limit = min(chunk_size, remaining)
cur.execute(
f'''SELECT m.public_id, m.content, m.peer_name, m.session_name, m.workspace_name
FROM "{schema}"."messages" m
WHERE m.workspace_name = %s
AND NOT EXISTS (
SELECT 1 FROM "{schema}"."message_embeddings" me
WHERE me.message_id = m.public_id
)
LIMIT %s''',
(WORKSPACE_NAME, fetch_limit),
)
rows = cur.fetchall()
if not rows:
break # No more messages to fetch
message_data.extend(
[
{
"public_id": row[0],
"content": row[1],
"peer_name": row[2],
"session_name": row[3],
"workspace_name": row[4],
}
for row in rows
]
)
total_fetched += len(rows)
print(
f" Fetched {total_fetched:,}/{embeddings_to_create:,} messages..."
)
if not message_data:
print(" ✓ No messages found that need embeddings")
return conn
# Insert in batches (smaller batch size for embeddings)
print(f" Inserting embeddings in batches of {EMBEDDING_BATCH_SIZE:,}...")
total_batches = (
len(message_data) + EMBEDDING_BATCH_SIZE - 1
) // EMBEDDING_BATCH_SIZE
for batch_num in range(total_batches):
batch_start = batch_num * EMBEDDING_BATCH_SIZE
batch_end = min(batch_start + EMBEDDING_BATCH_SIZE, len(message_data))
batch = message_data[batch_start:batch_end]
args = [
(
msg["content"],
CONSTANT_EMBEDDING,
msg["public_id"],
msg["workspace_name"],
msg["session_name"],
msg["peer_name"],
)
for msg in batch
]
# Retry logic for connection failures
for attempt in range(MAX_RETRIES):
try:
with conn.cursor() as cur:
cur.executemany(
f'''INSERT INTO "{schema}"."message_embeddings"
(content, embedding, message_id, workspace_name, session_name, peer_name)
VALUES (%s, CAST(%s AS vector), %s, %s, %s, %s)''',
args,
)
conn.commit()
break # Success, exit retry loop
except (psycopg.OperationalError, psycopg.InterfaceError):
if attempt < MAX_RETRIES - 1:
print(
f" ⚠ Connection error on batch {batch_num + 1}, retrying in {RETRY_DELAY}s... (attempt {attempt + 1}/{MAX_RETRIES})"
)
time.sleep(RETRY_DELAY)
# Reconnect with keepalive settings
try:
conn.close()
except:
pass
conn = psycopg.connect(
DB_CONNECTION_URI,
keepalives=1,
keepalives_idle=30,
keepalives_interval=10,
keepalives_count=5,
)
else:
print(f" ✗ Failed after {MAX_RETRIES} attempts")
raise
embeddings_created = existing_count + batch_end
progress = (embeddings_created / NUM_MESSAGES) * 100
print(
f" Batch {batch_num + 1}/{total_batches}: {embeddings_created:,}/{NUM_MESSAGES:,} ({progress:.1f}%)"
)
print(f" ✓ Successfully created {len(message_data):,} new message embeddings!")
return conn
def generate_documents(conn, schema, existing_count):
"""Generate and insert documents up to NUM_DOCUMENTS total."""
documents_to_create = NUM_DOCUMENTS - existing_count
if documents_to_create <= 0:
print(f"\n→ All {NUM_DOCUMENTS:,} documents already exist, skipping...")
return
print(
f"\nGenerating {documents_to_create:,} documents ({existing_count:,} already exist)..."
)
# Pre-generate all document data
print(" Generating document data...")
document_data = []
for i in range(documents_to_create):
doc_id = generate_nanoid()
# Randomly assign to a collection/peer
idx = random.randint(0, NUM_PEERS - 1)
collection_name = COLLECTION_NAMES[idx]
peer_name = PEER_NAMES[idx]
content = fake.text(max_nb_chars=300)
document_data.append(
{
"id": doc_id,
"content": content,
"collection_name": collection_name,
"peer_name": peer_name,
"workspace_name": WORKSPACE_NAME,
}
)
# Insert in batches
print(f" Inserting documents in batches of {BATCH_SIZE:,}...")
total_batches = (documents_to_create + BATCH_SIZE - 1) // BATCH_SIZE
with conn.cursor() as cur:
for batch_num in range(total_batches):
batch_start = batch_num * BATCH_SIZE
batch_end = min(batch_start + BATCH_SIZE, documents_to_create)
batch = document_data[batch_start:batch_end]
args = [
(
doc["id"],
doc["content"],
CONSTANT_EMBEDDING,
doc["collection_name"],
doc["peer_name"],
doc["workspace_name"],
)
for doc in batch
]
cur.executemany(
f'''INSERT INTO "{schema}"."documents"
(id, content, embedding, collection_name, peer_name, workspace_name)
VALUES (%s, %s, CAST(%s AS vector), %s, %s, %s)''',
args,
)
conn.commit()
documents_created = existing_count + batch_end
progress = (documents_created / NUM_DOCUMENTS) * 100
print(
f" Batch {batch_num + 1}/{total_batches}: {documents_created:,}/{NUM_DOCUMENTS:,} ({progress:.1f}%)"
)
print(f" ✓ Successfully created {documents_to_create:,} new documents!")
def main():
"""Main execution function."""
print("=" * 60)
print("HONCHO LOAD TEST - Raw psycopg3 Implementation (Resumable)")
print("=" * 60)
print("Configuration:")
print(f" Database URI: {DB_CONNECTION_URI}")
print(f" Schema: {DB_SCHEMA}")
print(f" Target messages: {NUM_MESSAGES:,}")
print(f" Target message embeddings: {NUM_MESSAGES:,}")
print(f" Target documents: {NUM_DOCUMENTS:,}")
print(f" Batch size (messages/docs): {BATCH_SIZE:,}")
print(f" Batch size (embeddings): {EMBEDDING_BATCH_SIZE:,}")
print("=" * 60)
# Connect to database with keepalive settings
print("\nConnecting to database...")
conn = psycopg.connect(
DB_CONNECTION_URI,
keepalives=1,
keepalives_idle=30,
keepalives_interval=10,
keepalives_count=5,
)
try:
# Check what already exists
existing_data = check_existing_data(conn, DB_SCHEMA)
# Create all entities
create_entities(conn, DB_SCHEMA, existing_data)
# Generate messages
message_data = generate_messages(
conn, DB_SCHEMA, existing_data["existing_messages"]
)
# Generate message embeddings (returns potentially reconnected connection)
# conn = generate_message_embeddings(
# conn, DB_SCHEMA, message_data, existing_data["existing_embeddings"]
# )
# Generate documents
generate_documents(conn, DB_SCHEMA, existing_data["existing_documents"])
print("\n" + "=" * 60)
print("✓ LOAD TEST COMPLETE!")
print("=" * 60)
print("Summary:")
print(f" Workspace: {WORKSPACE_NAME}")
print(f" Peers: {NUM_PEERS}")
print(f" Sessions: {NUM_SESSIONS}")
print(f" Collections: {NUM_PEERS}")
print(f" Messages: {NUM_MESSAGES:,}")
print(f" Message embeddings: {NUM_MESSAGES:,}")
print(f" Documents: {NUM_DOCUMENTS:,}")
print("=" * 60)
except Exception as e:
print(f"\n✗ Error occurred: {e}")
print(
"\nThe script is resumable - simply run it again to continue from where it left off."
)
raise
finally:
conn.close()
print("\nDatabase connection closed.")
if __name__ == "__main__":
main()

633
scripts/backfill_data_2.4.0.py Normal file
View File

@ -0,0 +1,633 @@
"""Generate 500k test messages, message embeddings, and documents using raw psycopg3.
This script creates:
- 1 workspace
- 3 peers
- 3 sessions (with peers added to sessions via session_peers table)
- 500k messages distributed randomly across peers and sessions
- 500k message embeddings (one per message with constant zero vector)
- 1 collection per peer (3 total)
- 500k documents distributed across the collections
Usage:
python manual_load.py [schema_name]
If schema_name is not provided, uses DB_SCHEMA environment variable.
Environment variables required:
DB_CONNECTION_URI: PostgreSQL connection URI
DB_SCHEMA: Database schema to use (optional if passed as argument)
"""
import os
import random
import sys
import time
import psycopg
from dotenv import load_dotenv
from faker import Faker
from nanoid import generate as generate_nanoid
load_dotenv()
# Configuration
NUM_MESSAGES = 500_000
NUM_DOCUMENTS = 500_000
BATCH_SIZE = 10_000 # Batch size for inserts (messages and documents)
EMBEDDING_BATCH_SIZE = 1_000 # Smaller batch size for embeddings (they're large)
MAX_RETRIES = 3 # Number of retries for failed operations
RETRY_DELAY = 5 # Seconds to wait before retrying
NUM_PEERS = 3
NUM_SESSIONS = 3
# Load environment variables
DB_CONNECTION_URI = os.getenv("DB_CONNECTION_URI")
# Get schema from command line argument or environment variable
if len(sys.argv) > 1:
DB_SCHEMA = sys.argv[1]
else:
DB_SCHEMA = os.getenv("DB_SCHEMA")
if not DB_CONNECTION_URI:
raise ValueError("DB_CONNECTION_URI environment variable is required")
if not DB_SCHEMA:
raise ValueError(
"DB_SCHEMA must be provided as command line argument or environment variable"
)
# Convert SQLAlchemy-style URI to plain psycopg URI
# SQLAlchemy uses postgresql+psycopg://, psycopg uses postgresql://
if DB_CONNECTION_URI.startswith("postgresql+psycopg://"):
DB_CONNECTION_URI = DB_CONNECTION_URI.replace("postgresql+psycopg://", "postgresql://")
elif DB_CONNECTION_URI.startswith("postgres+psycopg://"):
DB_CONNECTION_URI = DB_CONNECTION_URI.replace("postgres+psycopg://", "postgresql://")
# Initialize Faker
fake = Faker()
# Constant embedding - all zeros for speed (1536 dimensions for OpenAI embeddings)
CONSTANT_EMBEDDING = str([0.0] * 1536)
# Generate IDs
WORKSPACE_ID = generate_nanoid()
WORKSPACE_NAME = "load_test_workspace"
PEER_IDS = [generate_nanoid() for _ in range(NUM_PEERS)]
PEER_NAMES = [f"load_peer_{i + 1}" for i in range(NUM_PEERS)]
SESSION_IDS = [generate_nanoid() for _ in range(NUM_SESSIONS)]
SESSION_NAMES = [f"load_session_{i + 1}" for i in range(NUM_SESSIONS)]
COLLECTION_IDS = [generate_nanoid() for _ in range(NUM_PEERS)]
# Collections use observer/observed paradigm - each peer observes itself
COLLECTION_OBSERVERS = PEER_NAMES
COLLECTION_OBSERVED = PEER_NAMES
def check_existing_data(conn, schema):
"""Check what data already exists in the database."""
print("\nChecking existing data...")
with conn.cursor() as cur:
# Check workspace
cur.execute(
f'SELECT COUNT(*) FROM "{schema}"."workspaces" WHERE name = %s',
(WORKSPACE_NAME,),
)
workspace_exists = cur.fetchone()[0] > 0
# Check peers
cur.execute(
f'SELECT COUNT(*) FROM "{schema}"."peers" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_peers = cur.fetchone()[0]
# Check sessions
cur.execute(
f'SELECT COUNT(*) FROM "{schema}"."sessions" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_sessions = cur.fetchone()[0]
# Check collections (observer/observed pairs)
cur.execute(
f'SELECT COUNT(*) FROM "{schema}"."collections" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_collections = cur.fetchone()[0]
# Check messages
cur.execute(
f'SELECT COUNT(*) FROM "{schema}"."messages" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_messages = cur.fetchone()[0]
# Check message embeddings
cur.execute(
f'SELECT COUNT(*) FROM "{schema}"."message_embeddings" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_embeddings = cur.fetchone()[0]
# Check documents
cur.execute(
f'SELECT COUNT(*) FROM "{schema}"."documents" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_documents = cur.fetchone()[0]
print(f" Workspace exists: {workspace_exists}")
print(f" Existing peers: {existing_peers}/{NUM_PEERS}")
print(f" Existing sessions: {existing_sessions}/{NUM_SESSIONS}")
print(f" Existing collections: {existing_collections}/{NUM_PEERS}")
print(f" Existing messages: {existing_messages:,}/{NUM_MESSAGES:,}")
print(f" Existing message embeddings: {existing_embeddings:,}/{NUM_MESSAGES:,}")
print(f" Existing documents: {existing_documents:,}/{NUM_DOCUMENTS:,}")
return {
"workspace_exists": workspace_exists,
"existing_peers": existing_peers,
"existing_sessions": existing_sessions,
"existing_collections": existing_collections,
"existing_messages": existing_messages,
"existing_embeddings": existing_embeddings,
"existing_documents": existing_documents,
}
def create_entities(conn, schema, existing_data):
"""Create workspace, peers, sessions, and collections."""
print("\nCreating workspace, peers, sessions, and collections...")
with conn.cursor() as cur:
# Create workspace if not exists
if not existing_data["workspace_exists"]:
cur.execute(
f'INSERT INTO "{schema}"."workspaces" (id, name) VALUES (%s, %s)',
(WORKSPACE_ID, WORKSPACE_NAME),
)
print(f" ✓ Created workspace: {WORKSPACE_NAME}")
else:
print(f" → Workspace already exists: {WORKSPACE_NAME}")
# Create peers if needed
if existing_data["existing_peers"] < NUM_PEERS:
# Get existing peer names
cur.execute(
f'SELECT name FROM "{schema}"."peers" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_peer_names = {row[0] for row in cur.fetchall()}
for peer_id, peer_name in zip(PEER_IDS, PEER_NAMES, strict=False):
if peer_name not in existing_peer_names:
cur.execute(
f'INSERT INTO "{schema}"."peers" (id, name, workspace_name) VALUES (%s, %s, %s)',
(peer_id, peer_name, WORKSPACE_NAME),
)
new_peers = NUM_PEERS - existing_data["existing_peers"]
print(
f" ✓ Created {new_peers} new peers ({existing_data['existing_peers']} already existed)"
)
else:
print(f" → All {NUM_PEERS} peers already exist")
# Create sessions if needed
if existing_data["existing_sessions"] < NUM_SESSIONS:
# Get existing session names
cur.execute(
f'SELECT name FROM "{schema}"."sessions" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_session_names = {row[0] for row in cur.fetchall()}
for session_id, session_name in zip(
SESSION_IDS, SESSION_NAMES, strict=False
):
if session_name not in existing_session_names:
cur.execute(
f'INSERT INTO "{schema}"."sessions" (id, name, workspace_name) VALUES (%s, %s, %s)',
(session_id, session_name, WORKSPACE_NAME),
)
new_sessions = NUM_SESSIONS - existing_data["existing_sessions"]
print(
f" ✓ Created {new_sessions} new sessions ({existing_data['existing_sessions']} already existed)"
)
else:
print(f" → All {NUM_SESSIONS} sessions already exist")
# Add peers to sessions via session_peers table
# Check existing relationships
cur.execute(
f'SELECT COUNT(*) FROM "{schema}"."session_peers" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_relationships = cur.fetchone()[0]
expected_relationships = NUM_SESSIONS * NUM_PEERS
if existing_relationships < expected_relationships:
for session_name in SESSION_NAMES:
for peer_name in PEER_NAMES:
# Check if relationship exists
cur.execute(
f'''SELECT COUNT(*) FROM "{schema}"."session_peers"
WHERE workspace_name = %s AND session_name = %s AND peer_name = %s''',
(WORKSPACE_NAME, session_name, peer_name),
)
if cur.fetchone()[0] == 0:
cur.execute(
f'''INSERT INTO "{schema}"."session_peers"
(workspace_name, session_name, peer_name)
VALUES (%s, %s, %s)''',
(WORKSPACE_NAME, session_name, peer_name),
)
new_relationships = expected_relationships - existing_relationships
print(
f" ✓ Added {new_relationships} new peer-session relationships ({existing_relationships} already existed)"
)
else:
print(" → All peer-session relationships already exist")
# Create collections (one per peer, self-observation) if needed
if existing_data["existing_collections"] < NUM_PEERS:
# Get existing collection observer/observed pairs
cur.execute(
f'SELECT observer, observed FROM "{schema}"."collections" WHERE workspace_name = %s',
(WORKSPACE_NAME,),
)
existing_collections = {(row[0], row[1]) for row in cur.fetchall()}
for collection_id, observer, observed in zip(
COLLECTION_IDS, COLLECTION_OBSERVERS, COLLECTION_OBSERVED, strict=False
):
if (observer, observed) not in existing_collections:
cur.execute(
f'INSERT INTO "{schema}"."collections" (id, observer, observed, workspace_name) VALUES (%s, %s, %s, %s)',
(collection_id, observer, observed, WORKSPACE_NAME),
)
new_collections = NUM_PEERS - existing_data["existing_collections"]
print(
f" ✓ Created {new_collections} new collections ({existing_data['existing_collections']} already existed)"
)
else:
print(f" → All {NUM_PEERS} collections already exist")
conn.commit()
def generate_messages(conn, schema, existing_count):
"""Generate and insert messages up to NUM_MESSAGES total."""
messages_to_create = NUM_MESSAGES - existing_count
if messages_to_create <= 0:
print(f"\n→ All {NUM_MESSAGES:,} messages already exist, skipping...")
return []
print(
f"\nGenerating {messages_to_create:,} messages ({existing_count:,} already exist)..."
)
# Pre-generate all message data
print(" Generating message data...")
message_data = []
for i in range(messages_to_create):
public_id = generate_nanoid()
peer_name = random.choice(PEER_NAMES)
session_name = random.choice(SESSION_NAMES)
content = fake.text(max_nb_chars=200)
message_data.append(
{
"public_id": public_id,
"peer_name": peer_name,
"session_name": session_name,
"content": content,
"workspace_name": WORKSPACE_NAME,
}
)
# Insert in batches
print(f" Inserting messages in batches of {BATCH_SIZE:,}...")
total_batches = (messages_to_create + BATCH_SIZE - 1) // BATCH_SIZE
with conn.cursor() as cur:
for batch_num in range(total_batches):
batch_start = batch_num * BATCH_SIZE
batch_end = min(batch_start + BATCH_SIZE, messages_to_create)
batch = message_data[batch_start:batch_end]
# Use execute_batch for better performance
args = [
(
msg["public_id"],
msg["session_name"],
msg["content"],
msg["peer_name"],
msg["workspace_name"],
)
for msg in batch
]
cur.executemany(
f'''INSERT INTO "{schema}"."messages"
(public_id, session_name, content, peer_name, workspace_name)
VALUES (%s, %s, %s, %s, %s)''',
args,
)
conn.commit()
messages_created = existing_count + batch_end
progress = (messages_created / NUM_MESSAGES) * 100
print(
f" Batch {batch_num + 1}/{total_batches}: {messages_created:,}/{NUM_MESSAGES:,} ({progress:.1f}%)"
)
print(f" ✓ Successfully created {messages_to_create:,} new messages!")
return message_data
def generate_message_embeddings(conn, schema, message_data, existing_count):
"""Generate and insert message embeddings for messages that don't have them."""
embeddings_to_create = NUM_MESSAGES - existing_count
if embeddings_to_create <= 0:
print(f"\n→ All {NUM_MESSAGES:,} message embeddings already exist, skipping...")
return conn
print(
f"\nGenerating {embeddings_to_create:,} message embeddings ({existing_count:,} already exist)..."
)
# If we just created messages, use that data
if message_data:
print(" Using newly created message data...")
else:
# Fetch messages that don't have embeddings yet in chunks
# Using NOT EXISTS is much faster than LEFT JOIN for large tables
print(" Fetching messages without embeddings in chunks...")
message_data = []
chunk_size = 50_000 # Fetch in chunks to avoid timeout
total_fetched = 0
with conn.cursor() as cur:
while total_fetched < embeddings_to_create:
remaining = embeddings_to_create - total_fetched
fetch_limit = min(chunk_size, remaining)
cur.execute(
f'''SELECT m.public_id, m.content, m.peer_name, m.session_name, m.workspace_name
FROM "{schema}"."messages" m
WHERE m.workspace_name = %s
AND NOT EXISTS (
SELECT 1 FROM "{schema}"."message_embeddings" me
WHERE me.message_id = m.public_id
)
LIMIT %s''',
(WORKSPACE_NAME, fetch_limit),
)
rows = cur.fetchall()
if not rows:
break # No more messages to fetch
message_data.extend(
[
{
"public_id": row[0],
"content": row[1],
"peer_name": row[2],
"session_name": row[3],
"workspace_name": row[4],
}
for row in rows
]
)
total_fetched += len(rows)
print(
f" Fetched {total_fetched:,}/{embeddings_to_create:,} messages..."
)
if not message_data:
print(" ✓ No messages found that need embeddings")
return conn
# Insert in batches (smaller batch size for embeddings)
print(f" Inserting embeddings in batches of {EMBEDDING_BATCH_SIZE:,}...")
total_batches = (
len(message_data) + EMBEDDING_BATCH_SIZE - 1
) // EMBEDDING_BATCH_SIZE
for batch_num in range(total_batches):
batch_start = batch_num * EMBEDDING_BATCH_SIZE
batch_end = min(batch_start + EMBEDDING_BATCH_SIZE, len(message_data))
batch = message_data[batch_start:batch_end]
args = [
(
msg["content"],
CONSTANT_EMBEDDING,
msg["public_id"],
msg["workspace_name"],
msg["session_name"],
msg["peer_name"],
)
for msg in batch
]
# Retry logic for connection failures
for attempt in range(MAX_RETRIES):
try:
with conn.cursor() as cur:
cur.executemany(
f'''INSERT INTO "{schema}"."message_embeddings"
(content, embedding, message_id, workspace_name, session_name, peer_name)
VALUES (%s, CAST(%s AS vector), %s, %s, %s, %s)''',
args,
)
conn.commit()
break # Success, exit retry loop
except (psycopg.OperationalError, psycopg.InterfaceError):
if attempt < MAX_RETRIES - 1:
print(
f" ⚠ Connection error on batch {batch_num + 1}, retrying in {RETRY_DELAY}s... (attempt {attempt + 1}/{MAX_RETRIES})"
)
time.sleep(RETRY_DELAY)
# Reconnect with keepalive settings
try:
conn.close()
except:
pass
conn = psycopg.connect(
DB_CONNECTION_URI,
keepalives=1,
keepalives_idle=30,
keepalives_interval=10,
keepalives_count=5,
)
else:
print(f" ✗ Failed after {MAX_RETRIES} attempts")
raise
embeddings_created = existing_count + batch_end
progress = (embeddings_created / NUM_MESSAGES) * 100
print(
f" Batch {batch_num + 1}/{total_batches}: {embeddings_created:,}/{NUM_MESSAGES:,} ({progress:.1f}%)"
)
print(f" ✓ Successfully created {len(message_data):,} new message embeddings!")
return conn
def generate_documents(conn, schema, existing_count):
"""Generate and insert documents up to NUM_DOCUMENTS total."""
documents_to_create = NUM_DOCUMENTS - existing_count
if documents_to_create <= 0:
print(f"\n→ All {NUM_DOCUMENTS:,} documents already exist, skipping...")
return
print(
f"\nGenerating {documents_to_create:,} documents ({existing_count:,} already exist)..."
)
# Pre-generate all document data
print(" Generating document data...")
document_data = []
for i in range(documents_to_create):
doc_id = generate_nanoid()
# Randomly assign to a collection/peer (observer/observed pairs)
idx = random.randint(0, NUM_PEERS - 1)
observer = COLLECTION_OBSERVERS[idx]
observed = COLLECTION_OBSERVED[idx]
# Randomly assign to a session
session_name = random.choice(SESSION_NAMES)
content = fake.text(max_nb_chars=300)
document_data.append(
{
"id": doc_id,
"content": content,
"observer": observer,
"observed": observed,
"session_name": session_name,
"workspace_name": WORKSPACE_NAME,
}
)
# Insert in batches
print(f" Inserting documents in batches of {BATCH_SIZE:,}...")
total_batches = (documents_to_create + BATCH_SIZE - 1) // BATCH_SIZE
with conn.cursor() as cur:
for batch_num in range(total_batches):
batch_start = batch_num * BATCH_SIZE
batch_end = min(batch_start + BATCH_SIZE, documents_to_create)
batch = document_data[batch_start:batch_end]
args = [
(
doc["id"],
doc["content"],
CONSTANT_EMBEDDING,
doc["observer"],
doc["observed"],
doc["workspace_name"],
doc["session_name"],
)
for doc in batch
]
cur.executemany(
f'''INSERT INTO "{schema}"."documents"
(id, content, embedding, observer, observed, workspace_name, session_name)
VALUES (%s, %s, CAST(%s AS vector), %s, %s, %s, %s)''',
args,
)
conn.commit()
documents_created = existing_count + batch_end
progress = (documents_created / NUM_DOCUMENTS) * 100
print(
f" Batch {batch_num + 1}/{total_batches}: {documents_created:,}/{NUM_DOCUMENTS:,} ({progress:.1f}%)"
)
print(f" ✓ Successfully created {documents_to_create:,} new documents!")
def main():
"""Main execution function."""
print("=" * 60)
print("HONCHO LOAD TEST - Raw psycopg3 Implementation (Resumable)")
print("=" * 60)
print("Configuration:")
print(f" Database URI: {DB_CONNECTION_URI}")
print(f" Schema: {DB_SCHEMA}")
print(f" Target messages: {NUM_MESSAGES:,}")
print(f" Target message embeddings: {NUM_MESSAGES:,}")
print(f" Target documents: {NUM_DOCUMENTS:,}")
print(f" Batch size (messages/docs): {BATCH_SIZE:,}")
print(f" Batch size (embeddings): {EMBEDDING_BATCH_SIZE:,}")
print("=" * 60)
# Connect to database with keepalive settings
print("\nConnecting to database...")
conn = psycopg.connect(
DB_CONNECTION_URI,
keepalives=1,
keepalives_idle=30,
keepalives_interval=10,
keepalives_count=5,
)
try:
# Check what already exists
existing_data = check_existing_data(conn, DB_SCHEMA)
# Create all entities
create_entities(conn, DB_SCHEMA, existing_data)
# Generate messages
message_data = generate_messages(
conn, DB_SCHEMA, existing_data["existing_messages"]
)
# Generate message embeddings (returns potentially reconnected connection)
# conn = generate_message_embeddings(
# conn, DB_SCHEMA, message_data, existing_data["existing_embeddings"]
# )
# Generate documents
generate_documents(conn, DB_SCHEMA, existing_data["existing_documents"])
print("\n" + "=" * 60)
print("✓ LOAD TEST COMPLETE!")
print("=" * 60)
print("Summary:")
print(f" Workspace: {WORKSPACE_NAME}")
print(f" Peers: {NUM_PEERS}")
print(f" Sessions: {NUM_SESSIONS}")
print(f" Collections: {NUM_PEERS}")
print(f" Messages: {NUM_MESSAGES:,}")
print(f" Message embeddings: {NUM_MESSAGES:,}")
print(f" Documents: {NUM_DOCUMENTS:,}")
print("=" * 60)
except Exception as e:
print(f"\n✗ Error occurred: {e}")
print(
"\nThe script is resumable - simply run it again to continue from where it left off."
)
raise
finally:
conn.close()
print("\nDatabase connection closed.")
if __name__ == "__main__":
main()

2
src/config.py
View File

@ -223,6 +223,8 @@ class DeriverSettings(HonchoSettings):
# Template paths for prompt templates
CRITICAL_ANALYSIS_TEMPLATE: str = "deriver/critical_analysis.jinja"
PEER_CARD_TEMPLATE: str = "deriver/peer_card.jinja"
EXPLICIT_REASONING_TEMPLATE: str = "deriver/explicit.jinja"
DEDUCTIVE_REASONING_TEMPLATE: str = "deriver/deductive.jinja"
@model_validator(mode="after")
def validate_batch_tokens_vs_context_limit(self):

21
src/crud/representation.py
View File

@ -18,6 +18,7 @@ from src.utils.logging import accumulate_metric
from src.utils.representation import (
DeductiveObservation,
ExplicitObservation,
ImplicitObservation,
Representation,
)
@ -63,11 +64,19 @@ class RepresentationManager:
new_documents = 0
if not representation.deductive and not representation.explicit:
if (
not representation.deductive
and not representation.explicit
and not representation.implicit
):
logger.debug("No observations to save")
return new_documents
all_observations = representation.deductive + representation.explicit
all_observations = (
representation.deductive
+ representation.explicit
+ representation.implicit
)
# Batch embed all observations
batch_embed_start = time.perf_counter()
@ -116,7 +125,9 @@ class RepresentationManager:
async def _save_representation_internal(
self,
db: AsyncSession,
all_observations: list[ExplicitObservation | DeductiveObservation],
all_observations: list[
ExplicitObservation | ImplicitObservation | DeductiveObservation
],
embeddings: list[list[float]],
message_id_range: tuple[int, int],
session_name: str,
@ -138,6 +149,10 @@ class RepresentationManager:
obs_level = "deductive"
obs_content = obs.conclusion
obs_premises = obs.premises
elif isinstance(obs, ImplicitObservation):
obs_level = "implicit"
obs_content = obs.content
obs_premises = None
else:
obs_level = "explicit"
obs_content = obs.content

140
src/deriver/deriver.py
View File

@ -1,10 +1,9 @@
import datetime
import logging
import time
import sentry_sdk
from src import crud, exceptions, prometheus
from src import crud
from src.config import settings
from src.crud.representation import RepresentationManager
from src.dependencies import tracked_db
@ -19,63 +18,21 @@ from src.utils.logging import (
# log_representation,
)
from src.utils.peer_card import PeerCardQuery
from src.utils.representation import PromptRepresentation, Representation
from src.utils.representation import Representation
from src.utils.tokens import estimate_tokens
from src.utils.tracing import with_sentry_transaction
from .prompts import (
critical_analysis_prompt,
estimate_base_prompt_tokens,
peer_card_prompt,
)
from .reasoners.deductive import DeductiveReasoner
from .reasoners.explicit import ExplicitReasoner
logger = logging.getLogger(__name__)
logging.getLogger("sqlalchemy.engine.Engine").disabled = True
@conditional_observe(name="Critical Analysis Call")
async def critical_analysis_call(
peer_id: str,
peer_card: list[str] | None,
message_created_at: datetime.datetime,
working_representation: Representation,
history: str,
new_turns: list[str],
estimated_input_tokens: int,
) -> PromptRepresentation:
prompt = critical_analysis_prompt(
peer_id=peer_id,
peer_card=peer_card,
message_created_at=message_created_at,
working_representation=working_representation,
history=history,
new_turns=new_turns,
)
response = await honcho_llm_call(
provider=settings.DERIVER.PROVIDER,
model=settings.DERIVER.MODEL,
prompt=prompt,
max_tokens=settings.DERIVER.MAX_OUTPUT_TOKENS
or settings.LLM.DEFAULT_MAX_TOKENS,
track_name="Critical Analysis Call",
response_model=PromptRepresentation,
json_mode=True,
stop_seqs=[" \n", "\n\n\n\n"],
thinking_budget_tokens=settings.DERIVER.THINKING_BUDGET_TOKENS,
reasoning_effort="minimal",
verbosity="medium",
enable_retry=True,
retry_attempts=3,
)
prometheus.DERIVER_TOKENS_PROCESSED.labels(
task_type="representation",
).inc(response.output_tokens + estimated_input_tokens)
return response.content
@conditional_observe(name="Peer Card Call")
async def peer_card_call(
old_peer_card: list[str] | None,
@ -223,8 +180,9 @@ async def process_representation_tasks_batch(
)
logger.debug(
"Using working representation with %s explicit, %s deductive observations",
"Using working representation with %s explicit, %s implicit, %s deductive observations",
len(working_representation.explicit),
len(working_representation.implicit),
len(working_representation.deductive),
)
@ -267,8 +225,10 @@ async def process_representation_tasks_batch(
"ms",
)
total_observations = len(final_observations.explicit) + len(
final_observations.deductive
total_observations = (
len(final_observations.explicit)
+ len(final_observations.implicit)
+ len(final_observations.deductive)
)
accumulate_metric(
@ -288,6 +248,8 @@ class CertaintyReasoner:
ctx: list[Message]
observer: str
observed: str
explicit_reasoner: ExplicitReasoner
deductive_reasoner: DeductiveReasoner
def __init__(
self,
@ -304,6 +266,22 @@ class CertaintyReasoner:
self.observer = observer
self.estimated_input_tokens: int = estimated_input_tokens
# Instantiate the explicit and deductive reasoners
self.explicit_reasoner = ExplicitReasoner(
representation_manager=representation_manager,
ctx=ctx,
observed=observed,
observer=observer,
estimated_input_tokens=estimated_input_tokens,
)
self.deductive_reasoner = DeductiveReasoner(
representation_manager=representation_manager,
ctx=ctx,
observed=observed,
observer=observer,
estimated_input_tokens=estimated_input_tokens,
)
@conditional_observe(name="Deriver")
@sentry_sdk.trace
async def reason(
@ -313,11 +291,14 @@ class CertaintyReasoner:
speaker_peer_card: list[str] | None,
) -> Representation:
"""
Single-pass reasoning function that critically analyzes and derives insights.
Performs one analysis pass and returns the final observations.
Two-pass reasoning function that performs explicit then deductive reasoning.
First extracts explicit observations from messages, then performs deductive
reasoning building on those explicit observations.
Returns:
Representation: Final observations
Representation: Final observations combining explicit and deductive
reasoning
"""
analysis_start = time.perf_counter()
@ -330,36 +311,49 @@ class CertaintyReasoner:
]
logger.debug(
"CRITICAL ANALYSIS: message_created_at='%s', new_turns_count=%s",
"REASONING: message_created_at='%s', new_turns_count=%s",
latest_message.created_at,
len(new_turns),
)
try:
reasoning_response = await critical_analysis_call(
peer_id=self.observed,
peer_card=speaker_peer_card,
message_created_at=latest_message.created_at,
working_representation=working_representation,
history=history,
new_turns=new_turns,
estimated_input_tokens=self.estimated_input_tokens,
)
except Exception as e:
raise exceptions.LLMError(
speaker_peer_card=speaker_peer_card,
working_representation=working_representation,
history=history,
new_turns=new_turns,
) from e
# Step 1: Explicit reasoning
explicit_response = await self.explicit_reasoner.reason(
working_representation=working_representation,
history=history,
speaker_peer_card=speaker_peer_card,
)
reasoning_response = Representation.from_prompt_representation(
reasoning_response,
# Convert to Representation with metadata
explicit_observations = Representation.from_explicit_response(
explicit_response,
(earliest_message.id, latest_message.id),
latest_message.session_name,
latest_message.created_at,
)
# Step 2: Deductive reasoning (receives explicit observations)
deductive_response = await self.deductive_reasoner.reason(
working_representation=working_representation,
explicit_observations=explicit_observations,
history=history,
speaker_peer_card=speaker_peer_card,
)
# Convert to Representation with metadata
deductive_observations = Representation.from_deductive_response(
deductive_response,
(earliest_message.id, latest_message.id),
latest_message.session_name,
latest_message.created_at,
)
# Combine explicit, implicit, and deductive observations
reasoning_response = Representation(
explicit=explicit_observations.explicit,
implicit=explicit_observations.implicit,
deductive=deductive_observations.deductive,
)
analysis_duration_ms = (time.perf_counter() - analysis_start) * 1000
accumulate_metric(
f"deriver_{latest_message.id}_{self.observer}",

113
src/deriver/prompts.py
View File

@ -76,13 +76,15 @@ def peer_card_prompt(
@cache
def estimate_base_prompt_tokens() -> int:
"""Estimate base prompt tokens by calling critical_analysis_prompt with empty values.
"""Estimate base prompt tokens for explicit and deductive reasoning prompts.
This value is cached since it only changes on redeploys when the prompt template changes.
This value is cached since it only changes on redeploys when prompt templates change.
Returns the combined token estimate for both reasoning passes.
"""
try:
base_prompt = critical_analysis_prompt(
# Estimate explicit reasoning prompt tokens
explicit_prompt = explicit_reasoning_prompt(
peer_id="",
peer_card=None,
message_created_at=datetime.datetime.now(datetime.timezone.utc),
@ -90,7 +92,108 @@ def estimate_base_prompt_tokens() -> int:
history="",
new_turns=[],
)
return estimate_tokens(base_prompt)
explicit_tokens = estimate_tokens(explicit_prompt)
# Estimate deductive reasoning prompt tokens
deductive_prompt = deductive_reasoning_prompt(
peer_id="",
peer_card=None,
message_created_at=datetime.datetime.now(datetime.timezone.utc),
working_representation=Representation(),
explicit_observations=Representation(),
history="",
new_turns=[],
)
deductive_tokens = estimate_tokens(deductive_prompt)
return explicit_tokens + deductive_tokens
except Exception:
# Return a conservative estimate if estimation fails
return 500
return 1000 # Increased from 500 since we have two prompts now
def explicit_reasoning_prompt(
peer_id: str,
peer_card: list[str] | None,
message_created_at: datetime.datetime,
working_representation: Representation,
history: str,
new_turns: list[str],
) -> str:
"""
Generate the explicit reasoning prompt for the deriver.
Args:
peer_id (str): The ID of the user being analyzed.
peer_card (list[str] | None): The bio card of the user being analyzed.
message_created_at (datetime.datetime): Timestamp of the message.
working_representation (Representation): Current user understanding context.
history (str): Recent conversation history.
new_turns (list[str]): New conversation turns to analyze.
Returns:
Formatted prompt string for explicit reasoning
"""
return render_template(
settings.DERIVER.EXPLICIT_REASONING_TEMPLATE,
{
"peer_id": peer_id,
"peer_card": peer_card,
"message_created_at": message_created_at,
"working_representation": str(working_representation),
"has_working_representation": not working_representation.is_empty(),
"history": history,
"new_turns": new_turns,
},
)
def deductive_reasoning_prompt(
peer_id: str,
peer_card: list[str] | None,
message_created_at: datetime.datetime,
working_representation: Representation,
explicit_observations: Representation,
history: str,
new_turns: list[str],
) -> str:
"""
Generate the deductive reasoning prompt for the deriver.
Args:
peer_id (str): The ID of the user being analyzed.
peer_card (list[str] | None): The bio card of the user being analyzed.
message_created_at (datetime.datetime): Timestamp of the message.
working_representation (Representation): Current user understanding context.
explicit_observations (Representation): New explicit observations from current batch
(includes both explicit and implicit propositions).
history (str): Recent conversation history.
new_turns (list[str]): New conversation turns to analyze.
Returns:
Formatted prompt string for deductive reasoning
"""
# Format atomic propositions (includes both explicit and implicit from ExplicitReasoner)
# as numbered list - combine both lists
all_atomic_propositions = [
obs.content for obs in explicit_observations.explicit
] + [obs.content for obs in explicit_observations.implicit]
atomic_propositions_section = "\n".join(
[f"{i}. {prop}" for i, prop in enumerate(all_atomic_propositions, 1)]
)
return render_template(
settings.DERIVER.DEDUCTIVE_REASONING_TEMPLATE,
{
"peer_id": peer_id,
"peer_card": peer_card,
"message_created_at": message_created_at,
"working_representation": str(working_representation),
"has_working_representation": not working_representation.is_empty(),
"explicit_observations": str(explicit_observations),
"has_explicit_observations": not explicit_observations.is_empty(),
"atomic_propositions_section": atomic_propositions_section,
"history": history,
"new_turns": new_turns,
},
)

7
src/deriver/reasoners/__init__.py Normal file
View File

@ -0,0 +1,7 @@
"""Reasoner implementations for processing and deriving insights from messages."""
from src.deriver.reasoners.base import BaseReasoner
from src.deriver.reasoners.explicit import ExplicitReasoner
from src.deriver.reasoners.deductive import DeductiveReasoner
__all__ = ["BaseReasoner", "ExplicitReasoner", "DeductiveReasoner"]

28
src/deriver/reasoners/base.py Normal file
View File

@ -0,0 +1,28 @@
"""Base abstract class for reasoners."""
from abc import ABC, abstractmethod
from typing import Any
class BaseReasoner(ABC):
"""Abstract base class for reasoner implementations.
Reasoners are responsible for processing messages and deriving insights,
representations, or other cognitive outputs for the agent system.
"""
@abstractmethod
async def reason(self, *args: Any, **kwargs: Any) -> Any:
"""Process input and derive reasoning outputs.
Args:
*args: Positional arguments for reasoning
**kwargs: Keyword arguments for reasoning
Returns:
The result of the reasoning process
Raises:
NotImplementedError: If the subclass does not implement this method
"""
pass

118
src/deriver/reasoners/deductive.py Normal file
View File

@ -0,0 +1,118 @@
"""Deductive reasoner implementation."""
import logging
from src import exceptions, prometheus
from src.config import settings
from src.crud.representation import RepresentationManager
from src.deriver.prompts import deductive_reasoning_prompt
from src.deriver.reasoners.base import BaseReasoner
from src.models import Message
from src.utils.clients import honcho_llm_call
from src.utils.formatting import format_new_turn_with_timestamp
from src.utils.representation import DeductiveResponse, Representation
logger = logging.getLogger(__name__)
class DeductiveReasoner(BaseReasoner):
"""Deductive reasoning implementation.
This reasoner applies deductive logic to derive conclusions from
established facts and rules, building on explicit observations.
"""
representation_manager: RepresentationManager
ctx: list[Message]
observer: str
observed: str
estimated_input_tokens: int
def __init__(
self,
representation_manager: RepresentationManager,
ctx: list[Message],
*,
observed: str,
observer: str,
estimated_input_tokens: int,
) -> None:
"""Initialize the deductive reasoner.
Args:
representation_manager: Manager for saving representations
ctx: List of messages to analyze
observed: The peer being observed
observer: The peer doing the observing
estimated_input_tokens: Estimated token count for input
"""
self.representation_manager = representation_manager
self.ctx = ctx
self.observed = observed
self.observer = observer
self.estimated_input_tokens = estimated_input_tokens
async def reason(
self,
working_representation: Representation,
explicit_observations: Representation,
history: str,
speaker_peer_card: list[str] | None,
) -> DeductiveResponse:
"""Process input through deductive reasoning.
Args:
working_representation: Current representation context
explicit_observations: New explicit observations from current batch
history: Recent conversation history
speaker_peer_card: Peer card for the observed peer
Returns:
DeductiveResponse containing only deductive observations
"""
latest_message = self.ctx[-1]
new_turns = [
format_new_turn_with_timestamp(m.content, m.created_at, m.peer_name)
for m in self.ctx
]
prompt = deductive_reasoning_prompt(
peer_id=self.observed,
peer_card=speaker_peer_card,
message_created_at=latest_message.created_at,
working_representation=working_representation,
explicit_observations=explicit_observations,
history=history,
new_turns=new_turns,
)
try:
response = await honcho_llm_call(
provider=settings.DERIVER.PROVIDER,
model=settings.DERIVER.MODEL,
prompt=prompt,
max_tokens=settings.DERIVER.MAX_OUTPUT_TOKENS
or settings.LLM.DEFAULT_MAX_TOKENS,
track_name="Deductive Reasoning Call",
response_model=DeductiveResponse,
json_mode=True,
stop_seqs=[" \\n", "\\n\\n\\n\\n"],
thinking_budget_tokens=settings.DERIVER.THINKING_BUDGET_TOKENS,
reasoning_effort="minimal",
verbosity="medium",
enable_retry=True,
retry_attempts=3,
)
prometheus.DERIVER_TOKENS_PROCESSED.labels(
task_type="deductive_reasoning",
).inc(response.output_tokens + self.estimated_input_tokens)
return response.content
except Exception as e:
raise exceptions.LLMError(
speaker_peer_card=speaker_peer_card,
working_representation=working_representation,
history=history,
new_turns=new_turns,
) from e

115
src/deriver/reasoners/explicit.py Normal file
View File

@ -0,0 +1,115 @@
"""Explicit reasoner implementation."""
import logging
from src import exceptions, prometheus
from src.config import settings
from src.crud.representation import RepresentationManager
from src.deriver.prompts import explicit_reasoning_prompt
from src.deriver.reasoners.base import BaseReasoner
from src.models import Message
from src.utils.clients import honcho_llm_call
from src.utils.formatting import format_new_turn_with_timestamp
from src.utils.representation import ExplicitResponse, Representation
logger = logging.getLogger(__name__)
class ExplicitReasoner(BaseReasoner):
"""Explicit reasoning implementation.
This reasoner processes inputs through explicit, step-by-step reasoning,
extracting facts that are LITERALLY stated in messages.
"""
representation_manager: RepresentationManager
ctx: list[Message]
observer: str
observed: str
estimated_input_tokens: int
def __init__(
self,
representation_manager: RepresentationManager,
ctx: list[Message],
*,
observed: str,
observer: str,
estimated_input_tokens: int,
) -> None:
"""Initialize the explicit reasoner.
Args:
representation_manager: Manager for saving representations
ctx: List of messages to analyze
observed: The peer being observed
observer: The peer doing the observing
estimated_input_tokens: Estimated token count for input
"""
self.representation_manager = representation_manager
self.ctx = ctx
self.observed = observed
self.observer = observer
self.estimated_input_tokens = estimated_input_tokens
async def reason(
self,
working_representation: Representation,
history: str,
speaker_peer_card: list[str] | None,
) -> ExplicitResponse:
"""Process input through explicit reasoning.
Args:
working_representation: Current representation context
history: Recent conversation history
speaker_peer_card: Peer card for the observed peer
Returns:
ExplicitResponse containing only explicit observations
"""
latest_message = self.ctx[-1]
new_turns = [
format_new_turn_with_timestamp(m.content, m.created_at, m.peer_name)
for m in self.ctx
]
prompt = explicit_reasoning_prompt(
peer_id=self.observed,
peer_card=speaker_peer_card,
message_created_at=latest_message.created_at,
working_representation=working_representation,
history=history,
new_turns=new_turns,
)
try:
response = await honcho_llm_call(
provider=settings.DERIVER.PROVIDER,
model=settings.DERIVER.MODEL,
prompt=prompt,
max_tokens=settings.DERIVER.MAX_OUTPUT_TOKENS
or settings.LLM.DEFAULT_MAX_TOKENS,
track_name="Explicit Reasoning Call",
response_model=ExplicitResponse,
json_mode=True,
stop_seqs=[" \\n", "\\n\\n\\n\\n"],
thinking_budget_tokens=settings.DERIVER.THINKING_BUDGET_TOKENS,
reasoning_effort="minimal",
verbosity="medium",
enable_retry=True,
retry_attempts=3,
)
prometheus.DERIVER_TOKENS_PROCESSED.labels(
task_type="explicit_reasoning",
).inc(response.output_tokens + self.estimated_input_tokens)
return response.content
except Exception as e:
raise exceptions.LLMError(
speaker_peer_card=speaker_peer_card,
working_representation=working_representation,
history=history,
new_turns=new_turns,
) from e

4
src/schemas.py
View File

@ -305,9 +305,9 @@ class DocumentCreate(DocumentBase):
session_name: str = Field(
description="The session from which the document was derived"
)
level: Literal["explicit", "deductive"] = Field(
level: Literal["explicit", "implicit", "deductive"] = Field(
default="explicit",
description="The level of the document (explicit or deductive)",
description="The level of the document (explicit, implicit, or deductive)",
)
times_derived: int = Field(
default=1,

231
src/templates/deriver/deductive.jinja Normal file
View File

@ -0,0 +1,231 @@
You are a deductive reasoning agent performing formal logical inference over atomic propositions about {{ peer_id }} to derive new conclusions that NECESSARILY follow from the given premises.
TARGET PEER
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
You are performing deductive reasoning about: {{ peer_id }}
All conclusions you derive MUST be about {{ peer_id }}. Never generate conclusions about other individuals unless they directly characterize {{ peer_id }}'s relationship to or knowledge about those individuals.
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
IMPORTANT NAMING RULES
• Always start conclusions with {{ peer_id }}'s name (e.g., "{{ peer_id }} completed high school or equivalent education")
• NEVER use generic phrases like "The peer..." or "The user..." unless the peer name is unknown
• If referencing third parties, use explicit names with clarifiers like "(third-party)" when necessary
TASK: DEDUCTIVE REASONING
Your task is to perform DEDUCTIVE INFERENCE: deriving conclusions about {{ peer_id }} that MUST be true given the available premises through strict logical necessity.
**DEFINITION OF DEDUCTIVE REASONING:**
A deductive inference is valid when:
1. The conclusion NECESSARILY follows from the premises
2. If all premises are true, the conclusion MUST be true
3. The reasoning follows the laws of formal logic
**THE SUBSTANTIVE THRESHOLD:**
Not all valid deductions are worth making. A deduction must be both logically valid AND substantively useful. Since the atomic propositions you're working with have ALREADY been extracted through explicit and implicit analysis of messages, your deductions must go BEYOND what that extraction process would capture.
**SEMANTIC DIFFERENTIATION REQUIREMENT:**
Your deductions must be NOVEL and add information that is NOT already present in (or immediately obvious from) the atomic propositions themselves. The explicit extraction process already captures:
- Directly stated facts
- Clearly implied facts that are obvious from a single message
- Simple definitional implications
Therefore, you must ONLY generate deductions that:
- **Connect multiple atomic propositions** in non-obvious ways
- **Apply general knowledge** to derive facts not immediately apparent
- **Compute or calculate** new information (e.g., temporal calculations, numerical derivations)
- **Infer non-obvious preconditions** that require domain knowledge
- **Scaffold existing conclusions** to reach higher-order insights
- ❌ TRIVIAL (too obvious to be useful):
* "Maria spoke" → "Maria is alive" (biological necessity, assumed)
* "James greeted his cat" → "James acknowledged the cat's existence" (tautological)
* "Sarah ate lunch" → "Sarah consumed food" (definitional restatement)
* "Carlos read a book" → "Carlos has the ability to read" (obvious from the premise)
* "Elena wrote a message" → "Elena has communication abilities" (assumed capability)
- ❌ ALREADY IMPLICIT IN PREMISES (no semantic differentiation):
* "Maria has a dog" → "Maria owns a pet" (just a category restatement)
* "James went to college" → "James attended an educational institution" (already captured by explicit extraction)
* "Sarah took her dog for a walk" → "Sarah engaged in physical activity with her dog" (just rewording the premise)
* "Carlos has a daughter" → "Carlos has a child" (already implicit in the premise itself)
- ❌ DEFINITIONAL RESTATEMENT (not adding information):
* "Liam went to the store" → "Liam visited a retail establishment" (just rewording)
* "Aisha drove a car" → "Aisha operated a motor vehicle" (same information)
- ✓ SUBSTANTIVE (adds meaningful, non-obvious information):
* "Maria attended college" → "Maria completed high school or equivalent education" (non-obvious precondition requiring knowledge of educational systems)
* "James is 25 years old" + "Birthday is June 21" + "Current date: June 26, 2025" → "James was born on June 21, 2000" (computed temporal fact not present in any single premise)
* "Sarah has a dog" + "Sarah took her dog for a walk" + "Dogs require regular exercise" → "Sarah provides physical care for her dog's health needs" (synthesis of multiple propositions with general knowledge)
* "Carlos has a daughter starting kindergarten" + "Kindergarten typically starts at age 5" → "Carlos's daughter is approximately 5 years old" (domain-specific knowledge application)
* "Elena graduated with a PhD in neuroscience" + "PhDs require bachelor's degrees" → "Elena completed a bachelor's degree in a relevant field" (non-obvious educational prerequisite)
**KEY PRINCIPLE:** Only generate deductions that add substantive information that is semantically differentiated from the atomic propositions. Ask yourself:
1. "Would the explicit/implicit extraction have already captured this?" If yes, don't generate it.
2. "Does this conclusion connect or build on multiple propositions in a non-obvious way?" If no, don't generate it.
3. "Does this add meaningful context about who {{ peer_id }} is that isn't already present?" If no, don't generate it.
**PERMITTED PREMISE TYPES:**
You may ONLY use the following as premises in your deductions:
1. **Atomic propositions** provided below (explicitly extracted from {{ peer_id }}'s messages)
2. **Previous deductions** provided below (if any)
3. **General knowledge** - widely accepted facts about the world that are known to be true
4. **Temporal information** - current date/time: {message_created_at}
5. **Logical principles** - fundamental laws of logic and necessary relationships
**CRITICAL CONSTRAINT ON PREMISES:**
You MAY NOT use probabilistic or uncertain conclusions as premises for further deductions. If a previous deduction contains qualifiers indicating uncertainty (e.g., "likely", "probably", "typically", "may", "might", "potentially", "appears to", "seems to"), it CANNOT be used as a premise for further deduction.
Only use conclusions that express certainty and logical necessity.
**DEDUCTIVE INFERENCE PATTERNS:**
Common valid deductive patterns include:
1. **Categorical Syllogism**
- Premise 1: All A are B (general knowledge)
- Premise 2: X is A (atomic proposition)
- Conclusion: X is B
2. **Temporal Calculation**
- Premise 1: X is N years old (atomic proposition)
- Premise 2: X's birthday is DATE (atomic proposition)
- Premise 3: Current date is {message_created_at} (temporal)
- Conclusion: X was born in YEAR
3. **Necessary Precondition**
- Premise 1: X completed Y (atomic proposition)
- Premise 2: Completing Y requires Z (general knowledge)
- Conclusion: X completed Z
4. **Definitional Inference**
- Premise 1: X has a Y (atomic proposition)
- Premise 2: Having Y means being a Z (definitional)
- Conclusion: X is a Z
5. **Composite Inference**
- Premise 1: Multiple atomic propositions about X
- Premise 2: General knowledge about relationships
- Conclusion: Necessarily implied fact about X
**EXAMPLES OF VALID DEDUCTIONS:**
Example 1 - Categorical Syllogism:
- PREMISES:
* "Maria attended college" (atomic proposition)
* All people who attended college completed high school or equivalent (general knowledge)
- CONCLUSION: "Maria completed high school or equivalent education"
Example 2 - Temporal Calculation:
- PREMISES:
* "Liam is 25 years old" (atomic proposition)
* "Liam's birthday is June 21st" (atomic proposition)
* Current date is June 26, 2025 (temporal)
- CONCLUSION: "Liam was born on June 21, 2000"
Example 3 - Necessary Precondition:
- PREMISES:
* "Aisha has a dog" (atomic proposition)
* "Aisha took her dog for a walk" (atomic proposition)
* Dogs require regular exercise for health (general knowledge)
- CONCLUSION: "Aisha provides physical care for her dog's health needs"
Example 4 - Definitional Inference:
- PREMISES:
* "Carlos has a daughter" (atomic proposition)
* Having a child makes one a parent (definitional)
- CONCLUSION: "Carlos is a parent"
Example 5 - Multi-step Scaffolding:
- PREMISES:
* "Elena graduated with a PhD in neuroscience" (atomic proposition)
* A PhD requires completing a bachelor's degree (general knowledge)
* A bachelor's degree requires completing high school (general knowledge)
- CONCLUSIONS:
* "Elena completed a bachelor's degree"
* "Elena completed high school or equivalent education"
**SCOPE AND COMPLETENESS:**
- Derive ALL deductive conclusions that necessarily follow from the available premises
- Each deduction should contain ONE conclusion derived from one or more premises
- You may perform multi-step deductions by using previous certain deductions as premises
- Ensure each conclusion is sufficiently contextualized to be meaningful on its own
- Use absolute dates/times rather than relative references when possible
**WHAT NOT TO DO:**
- DO NOT speculate or guess beyond what logically follows
- DO NOT make inductive generalizations from patterns
- DO NOT make abductive inferences about motivations or explanations
- DO NOT include information that goes beyond logical necessity
- DO NOT generate trivial deductions that any downstream system would assume (e.g., "spoke" → "is alive")
- DO NOT simply restate premises in different words without adding substantive information
**CONTEXTUALIZATION REQUIREMENTS:**
Each deduction must be self-contained and include sufficient context:
- Specific subjects and objects (not vague references)
- Absolute temporal information when relevant
- Disambiguating details that make the conclusion independently meaningful
- All necessary qualifiers to ensure accuracy
{{ peer_card }}
{{ existing_deductions_section }}
Atomic propositions to use as premises:
<atomic_propositions>
{{ atomic_propositions_section }}
</atomic_propositions>
**INSTRUCTIONS:**
Perform deductive reasoning over the atomic propositions provided above. For each deduction:
1. Identify the premises being used (atomic propositions, previous deductions, general knowledge)
2. Verify that the conclusion NECESSARILY follows from the premises
3. Ensure the conclusion is about {{ peer_id }} and properly formatted
4. Ensure the conclusion is sufficiently contextualized to stand alone
Generate ALL valid deductions that can be derived from the available premises. Show your reasoning for each deduction. Output your response in JSON structured format:
```json
{
"deductions": [
{
"conclusion": "...",
"premises": [
"premise 1",
"premise 2",
...
"premise n"
]
},
{
"conclusion": "...",
"premises": [
"premise 1",
"premise 2",
...
"premise n"
]
},
...
{
"conclusion": "...",
"premises": [
"premise 1",
"premise 2",
...
"premise n"
]
},
]
}
```

133
src/templates/deriver/explicit.jinja Normal file
View File

@ -0,0 +1,133 @@
Extract atomic propositions from the peer's message. An atomic proposition is:
1. A statement with a SINGLE TRUTH VALUE (evaluable as true or false independently)
2. Contains NO LOGICAL CONNECTIVES (no AND, OR, IF-THEN, UNLESS, etc.)
3. SUFFICIENTLY CONTEXTUALIZED to be meaningful standing alone
**The Critical Balance:**
Each proposition must be atomic (indivisible) yet contain enough semantic context to be interpretable without reference to other propositions.
- ❌ TOO ATOMIC (lacks context):
* "Maria is happy" → Happy about what?
* "James said hi" → Said hi to whom? In what context?
* "Sarah went there" → Went where?
- ❌ TOO COMPOUND (multiple truth values):
* "Maria is happy and relieved about her promotion" → TWO propositions
* "James lives in NYC and works remotely" → TWO propositions
- ✓ PROPERLY ATOMIC (single truth value, sufficient context):
* "Maria is happy about her job promotion"
* "James said hi to his neighbor this morning"
* "Sarah went to the grocery store"
* "Maria owns a dog"
* "Maria's dog is named Charlie"
**Extraction Types:**
1. **EXPLICIT EXTRACTION** - Directly stated facts:
- Extract propositions directly asserted in the message
- Each claim becomes a separate atomic proposition
2. **IMPLICIT EXTRACTION** - Clearly implied facts:
- Extract propositions that are obviously implied by the message
- Only include implications that are certain, not speculative
- Examples:
* "I graduated from college" → IMPLIES: "Anthony attended college"
* "I'm taking my dog to the vet" → IMPLIES: "Sarah has a dog"
* "My 10-year-old loves soccer" → IMPLIES: "Marcus has a child"
**Decomposing Logical Connectives:**
Split any compound statement into separate atomic propositions:
- "I live in NYC and work remotely" → "James lives in NYC" + "James works remotely"
- "I like reading or watching movies" → "James likes reading" + "James likes watching movies"
**Ensuring Sufficient Context:**
Include specific semantic information in each proposition:
- Specific subjects/objects: "the job interview at the pharmacy" not just "the interview"
- Absolute temporal info: "June 21, 2025" not "yesterday"
- Disambiguating details: "Maria (third-party)" not just "she"
- Relevant qualifiers that make the proposition meaningful
**Information Sources for Contextualization:**
- Latest peer message (PRIMARY SOURCE - required)
- Conversation history (for context and disambiguation)
- Current date/time: {message_created_at}
- Message timestamps (convert relative dates to absolute)
**Examples:**
Example 1 - Explicit + Implicit with Temporal Context:
- MESSAGE: "I just had my 25th birthday last Saturday"
- CURRENT DATE: June 26, 2025
- EXTRACTED PROPOSITIONS:
* "Maria is 25 years old" [explicit]
* "Maria's birthday is June 21st" [explicit]
* "Maria was born in the year 2000" [implicit - derived from age and current date]
Example 2 - Decomposing Compounds:
- MESSAGE: "I took my dog for a walk in a park near my house in NYC"
- EXTRACTED PROPOSITIONS:
* "Liam has a dog" [implicit]
* "Liam took his dog for a walk" [explicit]
* "Liam walked his dog in a park" [explicit]
* "Liam has a house in NYC" [implicit]
* "Liam's house is near a park" [explicit]
Example 3 - Adding Context from History:
- MESSAGE: "I'm so nervous"
- HISTORY: Peer mentioned earlier they have a pharmacy job interview tomorrow
- EXTRACTED PROPOSITIONS:
* "Ann is nervous about her job interview at the pharmacy" [explicit, contextualized]
Example 4 - Implicit Extraction:
- MESSAGE: "My daughter starts kindergarten next month"
- EXTRACTED PROPOSITIONS:
* "Carlos has a daughter" [implicit]
* "Carlos's daughter will start kindergarten next month" [explicit]
* "Carlos's daughter is approximately 5 years old" [implicit - kindergarten age]
**Verification Checklist:**
- [ ] Each proposition has exactly ONE truth value
- [ ] No logical connectives (AND, OR, IF-THEN, etc.)
- [ ] Sufficient context to be meaningful independently
- [ ] Peer's name starts each proposition
- [ ] Absolute dates/times when temporal info present
- [ ] Both explicit and obvious implicit facts extracted
{{ peer_id }}'s known biographical information:
<peer_card>
{{ peer_card }}
</peer_card>
Current understanding of {{ peer_id }}:
<current_context>
{{ working_representation }}
</current_context>
Recent conversation history for context:
<history>
{{ history }}
</history>
New conversation turns to analyze:
<new_turns>
{{ new_turns }}
</new_turns>
Extract ALL atomic propositions (both explicit and clearly implied) from the latest peer message. Output your response in JSON structured format:
```json
{
"explicit":[
"explicit proposition 1",
"explicit proposition 2",
...
"explicit proposition n"
],
"implicit":[
"implicit proposition 1",
"implicit proposition 2",
...
"implicit proposition n"
]
}
```

175
src/utils/representation.py
View File

@ -18,6 +18,10 @@ class ExplicitObservationBase(BaseModel):
content: str = Field(description="The explicit observation")
class ImplicitObservationBase(BaseModel):
content: str = Field(description="The implicit observation")
class DeductiveObservationBase(BaseModel):
premises: list[str] = Field(
description="Supporting premises or evidence for this conclusion",
@ -39,6 +43,32 @@ class PromptRepresentation(BaseModel):
description="Conclusions that MUST be true given explicit facts and premises - strict logical necessities. Each deduction should have premises and a single conclusion.",
default_factory=list,
)
implicit: list[ImplicitObservationBase] = Field(
description="Facts CLEARLY IMPLIED by the user's message - atomic propositions derived through obvious implication. Example: ['Maria attended college' (from 'I graduated from college')]",
default_factory=list,
)
class ExplicitResponse(BaseModel):
"""Response model for explicit reasoning containing explicit and implicit observations."""
explicit: list[ExplicitObservationBase] = Field(
description="Facts LITERALLY stated by the user - direct quotes or clear paraphrases only, no interpretation or inference.",
default_factory=list,
)
implicit: list[ImplicitObservationBase] = Field(
description="Facts clearly implied by the user's message - certain implications, not speculative.",
default_factory=list,
)
class DeductiveResponse(BaseModel):
"""Response model for deductive reasoning containing only deductive observations."""
deductions: list[DeductiveObservationBase] = Field(
description="Conclusions that MUST be true given explicit facts and premises - strict logical necessities.",
default_factory=list,
)
class ExplicitObservation(ExplicitObservationBase, ObservationMetadata):
@ -67,6 +97,32 @@ class ExplicitObservation(ExplicitObservationBase, ObservationMetadata):
)
class ImplicitObservation(ImplicitObservationBase, ObservationMetadata):
"""Implicit observation with content and metadata."""
def __str__(self) -> str:
return f"[{self.created_at.replace(microsecond=0)}] {self.content}"
def __hash__(self) -> int:
"""
Make ImplicitObservation hashable for use in sets.
"""
return hash((self.content, self.created_at, self.session_name))
def __eq__(self, other: object) -> bool:
"""
Define equality for ImplicitObservation objects.
Two observations are equal if all their fields match.
"""
if not isinstance(other, ImplicitObservation):
return False
return (
self.content == other.content
and self.created_at == other.created_at
and self.session_name == other.session_name
)
class DeductiveObservation(DeductiveObservationBase, ObservationMetadata):
"""Deductive observation with multiple premises and one conclusion, plus metadata."""
@ -121,6 +177,11 @@ class Representation(BaseModel):
description="Facts LITERALLY stated by the user - direct quotes or clear paraphrases only, no interpretation or inference. Example: ['The user is 25 years old', 'The user has a dog']",
default_factory=list,
)
implicit: list[ImplicitObservation] = Field(
description="Facts CLEARLY IMPLIED by the user's message - atomic propositions derived through obvious implication. Example: ['Maria attended college' (from 'I graduated from college')]",
default_factory=list,
)
deductive: list[DeductiveObservation] = Field(
description="Conclusions that MUST be true given explicit facts and premises - strict logical necessities. Each deduction should have premises and a single conclusion.",
default_factory=list,
@ -130,7 +191,11 @@ class Representation(BaseModel):
"""
Check if the representation is empty.
"""
return len(self.explicit) == 0 and len(self.deductive) == 0
return (
len(self.explicit) == 0
and len(self.deductive) == 0
and len(self.implicit) == 0
)
def diff_representation(self, other: "Representation") -> "Representation":
"""
@ -140,6 +205,7 @@ class Representation(BaseModel):
diff = Representation()
diff.explicit = [o for o in other.explicit if o not in self.explicit]
diff.deductive = [o for o in other.deductive if o not in self.deductive]
diff.implicit = [o for o in other.implicit if o not in self.implicit]
return diff
def merge_representation(
@ -156,13 +222,15 @@ class Representation(BaseModel):
# removing duplicates by going list->set->list
self.explicit = list(set(self.explicit + other.explicit))
self.deductive = list(set(self.deductive + other.deductive))
self.implicit = list(set(self.implicit + other.implicit))
# sort by created_at
self.explicit.sort(key=lambda x: x.created_at)
self.deductive.sort(key=lambda x: x.created_at)
self.implicit.sort(key=lambda x: x.created_at)
if max_observations:
self.explicit = self.explicit[-max_observations:]
self.deductive = self.deductive[-max_observations:]
self.implicit = self.implicit[-max_observations:]
def __str__(self) -> str:
"""
@ -176,6 +244,8 @@ class Representation(BaseModel):
1. [2025-01-01 12:00:00] The user has a dog named Rover
2. [2025-01-01 12:01:00] The user's dog is 5 years old
3. [2025-01-01 12:05:00] The user is 25 years old
IMPLICIT:
1. [2025-01-01 12:02:00] The user is 20 years older than their dog
DEDUCTIVE:
1. [2025-01-01 12:01:00] Rover is 5 years old
- The user has a dog named Rover
@ -273,6 +343,18 @@ class Representation(BaseModel):
for doc in documents
if doc.level == "explicit"
],
implicit=[
ImplicitObservation(
created_at=_safe_datetime_from_metadata(
doc.internal_metadata, doc.created_at
),
content=doc.content,
message_ids=doc.internal_metadata.get("message_ids", [(0, 0)]),
session_name=doc.session_name,
)
for doc in documents
if doc.level == "implicit"
],
deductive=[
DeductiveObservation(
created_at=_safe_datetime_from_metadata(
@ -296,6 +378,11 @@ class Representation(BaseModel):
session_name: str,
created_at: datetime,
) -> "Representation":
"""Convert PromptRepresentation to Representation.
Used by vLLM client and legacy tests. New code should use
from_explicit_response() or from_deductive_response().
"""
return cls(
explicit=[
ExplicitObservation(
@ -306,6 +393,15 @@ class Representation(BaseModel):
)
for e in prompt_representation.explicit
],
implicit=[
ImplicitObservation(
content=i.content,
created_at=created_at,
message_ids=[message_ids],
session_name=session_name,
)
for i in prompt_representation.implicit
],
deductive=[
DeductiveObservation(
conclusion=d.conclusion,
@ -318,6 +414,81 @@ class Representation(BaseModel):
],
)
@classmethod
def from_explicit_response(
cls,
explicit_response: "ExplicitResponse",
message_ids: tuple[int, int],
session_name: str,
created_at: datetime,
) -> "Representation":
"""Convert ExplicitResponse to Representation with metadata.
Args:
explicit_response: Response from ExplicitReasoner
message_ids: Message ID range to link with observations
session_name: Session name for the observations
created_at: Timestamp for the observations
Returns:
Representation containing explicit and implicit observations
"""
return cls(
explicit=[
ExplicitObservation(
content=e.content,
created_at=created_at,
message_ids=[message_ids],
session_name=session_name,
)
for e in explicit_response.explicit
],
implicit=[
ImplicitObservation(
content=i.content,
created_at=created_at,
message_ids=[message_ids],
session_name=session_name,
)
for i in explicit_response.implicit
],
deductive=[],
)
@classmethod
def from_deductive_response(
cls,
deductive_response: "DeductiveResponse",
message_ids: tuple[int, int],
session_name: str,
created_at: datetime,
) -> "Representation":
"""Convert DeductiveResponse to Representation with metadata.
Args:
deductive_response: Response from DeductiveReasoner
message_ids: Message ID range to link with observations
session_name: Session name for the observations
created_at: Timestamp for the observations
Returns:
Representation containing deductive observations
"""
return cls(
explicit=[],
implicit=[],
deductive=[
DeductiveObservation(
conclusion=d.conclusion,
created_at=created_at,
message_ids=[message_ids],
session_name=session_name,
premises=d.premises,
)
for d in deductive_response.deductions
],
)
def _safe_datetime_from_metadata(
internal_metadata: dict[str, Any], fallback_datetime: datetime

2
src/utils/types.py
View File

@ -2,4 +2,4 @@ from typing import Literal
SupportedProviders = Literal["anthropic", "openai", "google", "groq", "custom", "vllm"]
TaskType = Literal["webhook", "summary", "representation", "dream"]
DocumentLevel = Literal["explicit", "deductive"]
DocumentLevel = Literal["explicit", "implicit", "deductive"]