From efef9a5d542138c6e4403ace9fc1d7416edafba6 Mon Sep 17 00:00:00 2001 From: Rajat Ahuja Date: Mon, 10 Nov 2025 13:53:19 -0500 Subject: [PATCH] feat: make deriver modular --- scripts/backfill_data_2.3.2.py | 629 +++++++++++++++++++++++++ scripts/backfill_data_2.4.0.py | 633 ++++++++++++++++++++++++++ src/config.py | 2 + src/crud/representation.py | 21 +- src/deriver/deriver.py | 140 +++--- src/deriver/prompts.py | 113 ++++- src/deriver/reasoners/__init__.py | 7 + src/deriver/reasoners/base.py | 28 ++ src/deriver/reasoners/deductive.py | 118 +++++ src/deriver/reasoners/explicit.py | 115 +++++ src/schemas.py | 4 +- src/templates/deriver/deductive.jinja | 231 ++++++++++ src/templates/deriver/explicit.jinja | 133 ++++++ src/utils/representation.py | 175 ++++++- src/utils/types.py | 2 +- 15 files changed, 2265 insertions(+), 86 deletions(-) create mode 100644 scripts/backfill_data_2.3.2.py create mode 100644 scripts/backfill_data_2.4.0.py create mode 100644 src/deriver/reasoners/__init__.py create mode 100644 src/deriver/reasoners/base.py create mode 100644 src/deriver/reasoners/deductive.py create mode 100644 src/deriver/reasoners/explicit.py create mode 100644 src/templates/deriver/deductive.jinja create mode 100644 src/templates/deriver/explicit.jinja diff --git a/scripts/backfill_data_2.3.2.py b/scripts/backfill_data_2.3.2.py new file mode 100644 index 00000000..56506f1c --- /dev/null +++ b/scripts/backfill_data_2.3.2.py @@ -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() diff --git a/scripts/backfill_data_2.4.0.py b/scripts/backfill_data_2.4.0.py new file mode 100644 index 00000000..9c33ca78 --- /dev/null +++ b/scripts/backfill_data_2.4.0.py @@ -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() diff --git a/src/config.py b/src/config.py index fbd245b0..6033fc61 100644 --- a/src/config.py +++ b/src/config.py @@ -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): diff --git a/src/crud/representation.py b/src/crud/representation.py index 4a41cdc6..8b7c90f6 100644 --- a/src/crud/representation.py +++ b/src/crud/representation.py @@ -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 diff --git a/src/deriver/deriver.py b/src/deriver/deriver.py index 5990a29b..b4e746d3 100644 --- a/src/deriver/deriver.py +++ b/src/deriver/deriver.py @@ -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}", diff --git a/src/deriver/prompts.py b/src/deriver/prompts.py index ba27c13f..ff25be10 100644 --- a/src/deriver/prompts.py +++ b/src/deriver/prompts.py @@ -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, + }, + ) diff --git a/src/deriver/reasoners/__init__.py b/src/deriver/reasoners/__init__.py new file mode 100644 index 00000000..8c3cd7e1 --- /dev/null +++ b/src/deriver/reasoners/__init__.py @@ -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"] diff --git a/src/deriver/reasoners/base.py b/src/deriver/reasoners/base.py new file mode 100644 index 00000000..59a43542 --- /dev/null +++ b/src/deriver/reasoners/base.py @@ -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 diff --git a/src/deriver/reasoners/deductive.py b/src/deriver/reasoners/deductive.py new file mode 100644 index 00000000..cbe71d93 --- /dev/null +++ b/src/deriver/reasoners/deductive.py @@ -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 diff --git a/src/deriver/reasoners/explicit.py b/src/deriver/reasoners/explicit.py new file mode 100644 index 00000000..c2b12ecc --- /dev/null +++ b/src/deriver/reasoners/explicit.py @@ -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 diff --git a/src/schemas.py b/src/schemas.py index 4d99dd42..e83e5db5 100644 --- a/src/schemas.py +++ b/src/schemas.py @@ -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, diff --git a/src/templates/deriver/deductive.jinja b/src/templates/deriver/deductive.jinja new file mode 100644 index 00000000..666f68c2 --- /dev/null +++ b/src/templates/deriver/deductive.jinja @@ -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_section }} + + +**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" + ] + }, + ] +} +``` \ No newline at end of file diff --git a/src/templates/deriver/explicit.jinja b/src/templates/deriver/explicit.jinja new file mode 100644 index 00000000..809e68be --- /dev/null +++ b/src/templates/deriver/explicit.jinja @@ -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 }} + + +Current understanding of {{ peer_id }}: + +{{ working_representation }} + + +Recent conversation history for context: + +{{ history }} + + +New conversation turns to analyze: + +{{ 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" + ] +} +``` \ No newline at end of file diff --git a/src/utils/representation.py b/src/utils/representation.py index c0f499f0..f71d0bee 100644 --- a/src/utils/representation.py +++ b/src/utils/representation.py @@ -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 diff --git a/src/utils/types.py b/src/utils/types.py index 80dfc389..2b1ec9d5 100644 --- a/src/utils/types.py +++ b/src/utils/types.py @@ -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"]