diff --git a/.gitignore b/.gitignore index 3a87a880..082fe223 100644 --- a/.gitignore +++ b/.gitignore @@ -45,6 +45,8 @@ hindsight-docs/static/llms-full.txt hindsight-dev/benchmarks/locomo/results/ hindsight-dev/benchmarks/longmemeval/results/ +hindsight-dev/benchmarks/consolidation/results/ +benchmarks/results/ hindsight-cli/target hindsight-clients/rust/target .claude diff --git a/hindsight-api/hindsight_api/config.py b/hindsight-api/hindsight_api/config.py index f291ec01..5d636cb6 100644 --- a/hindsight-api/hindsight_api/config.py +++ b/hindsight-api/hindsight_api/config.py @@ -101,6 +101,7 @@ ENV_RETAIN_OBSERVATIONS_ASYNC = "HINDSIGHT_API_RETAIN_OBSERVATIONS_ASYNC" # Observations settings (consolidated knowledge from facts) ENV_ENABLE_OBSERVATIONS = "HINDSIGHT_API_ENABLE_OBSERVATIONS" ENV_CONSOLIDATION_BATCH_SIZE = "HINDSIGHT_API_CONSOLIDATION_BATCH_SIZE" +ENV_CONSOLIDATION_MAX_TOKENS = "HINDSIGHT_API_CONSOLIDATION_MAX_TOKENS" # Optimization flags ENV_SKIP_LLM_VERIFICATION = "HINDSIGHT_API_SKIP_LLM_VERIFICATION" @@ -183,6 +184,7 @@ DEFAULT_RETAIN_OBSERVATIONS_ASYNC = False # Run observation generation async (a # Observations defaults (consolidated knowledge from facts) DEFAULT_ENABLE_OBSERVATIONS = True # Observations enabled by default DEFAULT_CONSOLIDATION_BATCH_SIZE = 50 # Memories to load per batch (internal memory optimization) +DEFAULT_CONSOLIDATION_MAX_TOKENS = 1024 # Max tokens for recall when finding related observations # Database migrations DEFAULT_RUN_MIGRATIONS_ON_STARTUP = True @@ -346,6 +348,7 @@ class HindsightConfig: # Observations settings (consolidated knowledge from facts) enable_observations: bool consolidation_batch_size: int + consolidation_max_tokens: int # Optimization flags skip_llm_verification: bool @@ -466,6 +469,9 @@ class HindsightConfig: consolidation_batch_size=int( os.getenv(ENV_CONSOLIDATION_BATCH_SIZE, str(DEFAULT_CONSOLIDATION_BATCH_SIZE)) ), + consolidation_max_tokens=int( + os.getenv(ENV_CONSOLIDATION_MAX_TOKENS, str(DEFAULT_CONSOLIDATION_MAX_TOKENS)) + ), # Database migrations run_migrations_on_startup=os.getenv(ENV_RUN_MIGRATIONS_ON_STARTUP, "true").lower() == "true", # Database connection pool diff --git a/hindsight-api/hindsight_api/engine/consolidation/consolidator.py b/hindsight-api/hindsight_api/engine/consolidation/consolidator.py index e6748400..8d640200 100644 --- a/hindsight-api/hindsight_api/engine/consolidation/consolidator.py +++ b/hindsight-api/hindsight_api/engine/consolidation/consolidator.py @@ -639,28 +639,27 @@ async def _find_related_observations( request_context: "RequestContext", ) -> list[dict[str, Any]]: """ - Find observations related to the given query using the full recall system. + Find observations related to the given query using optimized recall. IMPORTANT: We do NOT filter by tags here. Consolidation needs to see ALL potentially related observations regardless of scope, so the LLM can decide on tag routing (same scope update vs cross-scope create). - This leverages: - - Semantic search (embedding similarity) - - BM25 text search (keyword matching) - - Entity-based retrieval (shared entities) - - Graph traversal (connected via entity links) + Uses max_tokens to naturally limit observations (no artificial count limit). + Includes source memories with dates for LLM context. Returns: - List of related observations with their tags for LLM tag routing + List of related observations with their tags, source memories, and dates """ - # Use recall to find related observations - # NO tags parameter - we want ALL observations regardless of scope - # Use low max_tokens since we only need observations, not memories + # Use recall to find related observations with token budget + # max_tokens naturally limits how many observations are returned + from ...config import get_config + + config = get_config() recall_result = await memory_engine.recall_async( bank_id=bank_id, query=query, - max_tokens=5000, # Token budget for observations + max_tokens=config.consolidation_max_tokens, # Token budget for observations (configurable) fact_type=["observation"], # Only retrieve observations request_context=request_context, _quiet=True, # Suppress logging @@ -668,43 +667,82 @@ async def _find_related_observations( ) # If no observations returned, return empty list - # When fact_type=["observation"], results come back in `results` field if not recall_result.results: return [] - # Trust recall's relevance filtering - fetch full data for each observation + # Batch fetch all observations in a single query (no artificial limit) + observation_ids = [uuid.UUID(obs.id) for obs in recall_result.results] + + rows = await conn.fetch( + f""" + SELECT id, text, proof_count, history, tags, source_memory_ids, created_at, updated_at, + occurred_start, occurred_end, mentioned_at + FROM {fq_table("memory_units")} + WHERE id = ANY($1) AND bank_id = $2 AND fact_type = 'observation' + """, + observation_ids, + bank_id, + ) + + # Build results list preserving recall order + id_to_row = {row["id"]: row for row in rows} results = [] + for obs in recall_result.results: - # Fetch full observation data from DB to get history, source_memory_ids, tags - row = await conn.fetchrow( - f""" - SELECT id, text, proof_count, history, tags, source_memory_ids, created_at, updated_at - FROM {fq_table("memory_units")} - WHERE id = $1 AND bank_id = $2 AND fact_type = 'observation' - """, - uuid.UUID(obs.id), - bank_id, - ) + obs_id = uuid.UUID(obs.id) + if obs_id not in id_to_row: + continue - if row: - history = row["history"] - if isinstance(history, str): - history = json.loads(history) - elif history is None: - history = [] + row = id_to_row[obs_id] + history = row["history"] + if isinstance(history, str): + history = json.loads(history) + elif history is None: + history = [] - results.append( - { - "id": row["id"], - "text": row["text"], - "proof_count": row["proof_count"] or 1, - "history": history, - "tags": row["tags"] or [], # Include tags for LLM tag routing - "source_memory_ids": row["source_memory_ids"] or [], - "similarity": 1.0, # Retrieved via recall so assumed relevant - } + # Fetch source memories to include their text and dates + source_memory_ids = row["source_memory_ids"] or [] + source_memories = [] + + if source_memory_ids: + source_rows = await conn.fetch( + f""" + SELECT text, occurred_start, occurred_end, mentioned_at, event_date + FROM {fq_table("memory_units")} + WHERE id = ANY($1) AND bank_id = $2 + ORDER BY created_at ASC + LIMIT 5 + """, + source_memory_ids[:5], # Limit to first 5 source memories for token efficiency + bank_id, ) + for src_row in source_rows: + source_memories.append( + { + "text": src_row["text"], + "occurred_start": src_row["occurred_start"], + "occurred_end": src_row["occurred_end"], + "mentioned_at": src_row["mentioned_at"], + "event_date": src_row["event_date"], + } + ) + + results.append( + { + "id": row["id"], + "text": row["text"], + "proof_count": row["proof_count"] or 1, + "tags": row["tags"] or [], + "source_memories": source_memories, + "occurred_start": row["occurred_start"], + "occurred_end": row["occurred_end"], + "mentioned_at": row["mentioned_at"], + "created_at": row["created_at"], + "updated_at": row["updated_at"], + } + ) + return results @@ -732,14 +770,43 @@ async def _consolidate_with_llm( - {"action": "create", "text": "...", "reason": "..."} - [] if fact is purely ephemeral (no durable knowledge) """ - # Format observations WITH their tags (or "None" if empty) + # Format observations as JSON with source memories and dates if observations: - observations_text = "\n".join( - f'- ID: {obs["id"]}, Tags: {json.dumps(obs["tags"])}, Text: "{obs["text"]}" (proof_count: {obs["proof_count"]})' - for obs in observations - ) + obs_list = [] + for obs in observations: + obs_data = { + "id": str(obs["id"]), + "text": obs["text"], + "proof_count": obs["proof_count"], + "tags": obs["tags"], + "created_at": obs["created_at"].isoformat() if obs.get("created_at") else None, + "updated_at": obs["updated_at"].isoformat() if obs.get("updated_at") else None, + } + + # Include temporal info if available + if obs.get("occurred_start"): + obs_data["occurred_start"] = obs["occurred_start"].isoformat() + if obs.get("occurred_end"): + obs_data["occurred_end"] = obs["occurred_end"].isoformat() + if obs.get("mentioned_at"): + obs_data["mentioned_at"] = obs["mentioned_at"].isoformat() + + # Include source memories (up to 3 for brevity) + if obs.get("source_memories"): + obs_data["source_memories"] = [ + { + "text": sm["text"], + "event_date": sm["event_date"].isoformat() if sm.get("event_date") else None, + "occurred_start": sm["occurred_start"].isoformat() if sm.get("occurred_start") else None, + } + for sm in obs["source_memories"][:3] # Limit to 3 for token efficiency + ] + + obs_list.append(obs_data) + + observations_text = json.dumps(obs_list, indent=2) else: - observations_text = "None (this is a new topic - create if fact contains durable knowledge)" + observations_text = "[]" # Only include mission section if mission is set and not the default mission_section = "" diff --git a/hindsight-api/hindsight_api/engine/consolidation/prompts.py b/hindsight-api/hindsight_api/engine/consolidation/prompts.py index d48effbe..a0d4ff96 100644 --- a/hindsight-api/hindsight_api/engine/consolidation/prompts.py +++ b/hindsight-api/hindsight_api/engine/consolidation/prompts.py @@ -47,23 +47,31 @@ CONSOLIDATION_USER_PROMPT = """Analyze this new fact and consolidate into knowle {mission_section} NEW FACT: {fact_text} -EXISTING OBSERVATIONS: +EXISTING OBSERVATIONS (JSON array with source memories and dates): {observations_text} -Instructions: -1. First, extract the DURABLE KNOWLEDGE from the fact (not ephemeral state like "user is at X") -2. Then compare with existing observations: - - If an observation covers the same topic: UPDATE it with the new knowledge - - If no observation covers the topic: CREATE a new one +Each observation includes: +- id: unique identifier for updating +- text: the observation content +- proof_count: number of supporting memories +- tags: visibility scope (handled automatically) +- created_at/updated_at: when observation was created/modified +- occurred_start/occurred_end: temporal range of source facts +- source_memories: array of supporting facts with their text and dates -Output JSON array of actions (ALWAYS an array, even for single action): +Instructions: +1. Extract DURABLE KNOWLEDGE from the new fact (not ephemeral state) +2. Review source_memories in existing observations to understand evidence +3. Check dates to detect contradictions or updates +4. Compare with observations: + - Same topic → UPDATE with learning_id + - New topic → CREATE new observation + - Purely ephemeral → return [] + +Output JSON array of actions: [ - {{"action": "update", "learning_id": "uuid", "text": "updated durable knowledge", "reason": "..."}}, + {{"action": "update", "learning_id": "uuid-from-observations", "text": "updated knowledge", "reason": "..."}}, {{"action": "create", "text": "new durable knowledge", "reason": "..."}} ] -If NO consolidation is needed (fact is purely ephemeral with no durable knowledge): -[] - -If no observations exist and fact contains durable knowledge: -[{{"action": "create", "text": "durable knowledge text", "reason": "new topic"}}]""" +Return [] if fact contains no durable knowledge.""" diff --git a/hindsight-api/hindsight_api/main.py b/hindsight-api/hindsight_api/main.py index bd31c71e..1f045817 100644 --- a/hindsight-api/hindsight_api/main.py +++ b/hindsight-api/hindsight_api/main.py @@ -228,6 +228,7 @@ def main(): retain_observations_async=config.retain_observations_async, enable_observations=config.enable_observations, consolidation_batch_size=config.consolidation_batch_size, + consolidation_max_tokens=config.consolidation_max_tokens, skip_llm_verification=config.skip_llm_verification, lazy_reranker=config.lazy_reranker, run_migrations_on_startup=config.run_migrations_on_startup, diff --git a/hindsight-dev/benchmarks/consolidation/README.md b/hindsight-dev/benchmarks/consolidation/README.md new file mode 100644 index 00000000..bd9640fd --- /dev/null +++ b/hindsight-dev/benchmarks/consolidation/README.md @@ -0,0 +1,97 @@ +# Consolidation Performance Benchmark + +## Overview + +This benchmark measures consolidation throughput (operations per second) and identifies bottlenecks in the consolidation pipeline. + +## Quick Start + +```bash +# Run with default settings (100 memories) +./scripts/benchmarks/run-consolidation.sh + +# Run with custom number of memories +NUM_MEMORIES=50 ./scripts/benchmarks/run-consolidation.sh + +# Run with different model +export HINDSIGHT_API_CONSOLIDATION_LLM_MODEL=llama-3.1-70b-versatile +NUM_MEMORIES=100 ./scripts/benchmarks/run-consolidation.sh +``` + +## What It Measures + +The benchmark: +1. Creates N test memories with diverse content (similar facts, contradictions, different entities) +2. Runs consolidation and measures time spent in each component: + - **Recall**: Finding related observations + - **LLM**: Deciding on consolidation actions + - **Embedding**: Generating embeddings for new/updated observations + - **DB Write**: Writing to database +3. Reports throughput (op/sec) and detailed timing breakdown + +## Interpreting Results + +### Metrics +- **Throughput (op/sec)**: Memories processed per second +- **Timing Breakdown**: % of time spent in each component +- **Observations Created/Updated**: Quality indicator + +### Baseline Performance (groq/openai/gpt-oss-120b) +- **~0.7-1.0 op/sec** (1-1.4 seconds per memory) +- **LLM: 80-87%** of time (main bottleneck) +- **Recall: 10-17%** of time (secondary bottleneck) + +## Results + +See: +- `ANALYSIS.md` - Detailed bottleneck analysis +- `RESULTS.md` - Performance results and recommendations +- `benchmarks/results/` - Raw benchmark data (JSON) + +## Optimizations + +### Implemented +✅ Batch database queries (fixed N+1 problem) +✅ Reduced recall token budget (5000 → 2000) +✅ Limited observation results (top 15) + +### Recommended +🔧 Use faster LLM model for consolidation +🔧 Enable prompt caching (if available) +🔧 Optimize prompt verbosity + +See `RESULTS.md` for detailed recommendations. + +## Configuration + +Environment variables: +- `NUM_MEMORIES`: Number of memories to create (default: 100) +- `HINDSIGHT_API_CONSOLIDATION_LLM_MODEL`: Model for consolidation +- `HINDSIGHT_API_CONSOLIDATION_LLM_PROVIDER`: Provider for consolidation +- `HINDSIGHT_API_DATABASE_URL`: Database URL +- `HINDSIGHT_LOG_LEVEL`: Logging level (INFO for detailed logs) + +## Example Output + +``` +Consolidation Benchmark Results +┌────────────────────────────────┬─────────────┐ +│ Metric │ Value │ +├────────────────────────────────┼─────────────┤ +│ Total Time │ 60.28s │ +│ Memories Processed │ 43 │ +│ Throughput │ 0.71 op/sec │ +│ Avg Time/Memory │ 1.402s │ +│ │ │ +│ Observations Created │ 4 │ +│ Observations Updated │ 38 │ +│ Observations Merged │ 0 │ +│ Skipped (No Durable Knowledge) │ 1 │ +└────────────────────────────────┴─────────────┘ + +Timing breakdown: + recall=6.295s (10.4%) + llm=52.144s (86.5%) ← BOTTLENECK + embedding=1.717s (2.8%) + db_write=0.075s (0.1%) +``` diff --git a/hindsight-dev/benchmarks/consolidation/__init__.py b/hindsight-dev/benchmarks/consolidation/__init__.py new file mode 100644 index 00000000..d738b429 --- /dev/null +++ b/hindsight-dev/benchmarks/consolidation/__init__.py @@ -0,0 +1 @@ +"""Consolidation performance benchmarks.""" diff --git a/hindsight-dev/benchmarks/consolidation/consolidation_benchmark.py b/hindsight-dev/benchmarks/consolidation/consolidation_benchmark.py new file mode 100644 index 00000000..b3aa0a84 --- /dev/null +++ b/hindsight-dev/benchmarks/consolidation/consolidation_benchmark.py @@ -0,0 +1,331 @@ +""" +Consolidation performance benchmark. + +Measures consolidation throughput (op/sec) and identifies bottlenecks by: +1. Ingesting a batch of diverse memories +2. Running consolidation manually with detailed timing +3. Analyzing timing breakdown to identify bottlenecks +4. Reporting op/sec and time spent in each component +""" + +import asyncio +import json +import logging +import os +import time +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from hindsight_api.config import get_config +from hindsight_api.engine.consolidation.consolidator import run_consolidation_job +from hindsight_api.engine.memory_engine import MemoryEngine +from hindsight_api.models import RequestContext +from rich.console import Console +from rich.progress import BarColumn, Progress, SpinnerColumn, TextColumn +from rich.table import Table + +console = Console() + + +# Sample diverse memories to trigger different consolidation patterns +SAMPLE_MEMORIES = [ + # Similar memories (should merge) + "Alice loves coffee and drinks it every morning.", + "Alice prefers coffee over tea for her morning beverage.", + "Alice switched to decaf coffee recently.", + # Different person (should NOT merge with Alice) + "Bob works at Google as a software engineer.", + "Bob has been at Google for 5 years.", + # Technical facts + "Python is a programming language used for data science.", + "Python supports object-oriented and functional programming.", + # Product info + "The new iPhone 15 was released in September 2023.", + "The iPhone 15 features USB-C charging instead of Lightning.", + # Contradictions (should merge with conflict resolution) + "The meeting is scheduled for Tuesday at 2pm.", + "The meeting was moved to Wednesday at 3pm.", + # Entity-rich content + "Sarah Smith works at Microsoft in Seattle.", + "Sarah graduated from Stanford University in 2015.", + # Temporal information + "The project started on January 15, 2024.", + "The project deadline is March 30, 2024.", + # Preferences + "User prefers dark mode in applications.", + "User uses keyboard shortcuts extensively.", + # World knowledge + "Paris is the capital of France.", + "The Eiffel Tower is located in Paris.", + # Multiple entities + "John and Mary went to the Italian restaurant on Main Street.", + "The Italian restaurant on Main Street has excellent pizza.", +] + + +async def create_test_memories(memory_engine: MemoryEngine, bank_id: str, num_memories: int = 100) -> None: + """ + Create test memories by repeating and varying the sample memories. + + Args: + memory_engine: MemoryEngine instance + bank_id: Bank ID to ingest into + num_memories: Number of memories to create + """ + console.print(f"\n[cyan]Creating {num_memories} test memories...[/cyan]") + + # Generate memories by cycling through samples + memories = [] + for i in range(num_memories): + base_memory = SAMPLE_MEMORIES[i % len(SAMPLE_MEMORIES)] + # Add variation to avoid exact duplicates + memory = f"{base_memory} (context: test {i + 1})" + memories.append( + { + "content": memory, + "context": f"Test memory {i + 1}", + } + ) + + # Batch ingest + console.print("[yellow]Ingesting memories in batch...[/yellow]") + start_time = time.time() + await memory_engine.retain_batch_async( + bank_id=bank_id, + contents=memories, + request_context=RequestContext(), + ) + ingest_time = time.time() - start_time + console.print( + f"[green]✓[/green] Ingested {num_memories} memories in {ingest_time:.2f}s ({num_memories / ingest_time:.2f} mem/sec)" + ) + + +async def run_consolidation_benchmark( + memory_engine: MemoryEngine, + bank_id: str, + enable_detailed_logs: bool = True, +) -> dict[str, Any]: + """ + Run consolidation and measure performance. + + Args: + memory_engine: MemoryEngine instance + bank_id: Bank ID to consolidate + enable_detailed_logs: Enable detailed consolidation logs + + Returns: + Performance metrics dict + """ + console.print("\n[cyan]Running consolidation benchmark...[/cyan]") + + # Set log level to INFO to see consolidation logs + if enable_detailed_logs: + # Configure logging for consolidation + consolidation_logger = logging.getLogger("hindsight_api.engine.consolidation.consolidator") + consolidation_logger.setLevel(logging.INFO) + + # Add console handler if not present + if not consolidation_logger.handlers: + handler = logging.StreamHandler() + handler.setLevel(logging.INFO) + formatter = logging.Formatter("%(message)s") + handler.setFormatter(formatter) + consolidation_logger.addHandler(handler) + + console.print("[yellow]Detailed logging enabled for consolidation[/yellow]") + + # Run consolidation and measure time + start_time = time.time() + result = await run_consolidation_job( + memory_engine=memory_engine, + bank_id=bank_id, + request_context=RequestContext(), + ) + total_time = time.time() - start_time + + # Calculate op/sec + memories_processed = result.get("memories_processed", 0) + ops_per_sec = memories_processed / total_time if total_time > 0 else 0 + + console.print("\n[green]✓[/green] Consolidation complete!") + console.print(f" Total time: {total_time:.2f}s") + console.print(f" Memories processed: {memories_processed}") + console.print(f" Throughput: {ops_per_sec:.2f} op/sec") + console.print(f" Avg time per memory: {total_time / memories_processed:.3f}s" if memories_processed > 0 else "") + + return { + "total_time": total_time, + "memories_processed": memories_processed, + "ops_per_sec": ops_per_sec, + "consolidation_result": result, + } + + +async def analyze_timing_breakdown(bank_id: str) -> None: + """ + Analyze the timing breakdown from consolidation logs. + + NOTE: This relies on the performance logging in ConsolidationPerfLog. + The logs will show timing breakdowns for: recall, llm, embedding, db_write + """ + console.print("\n[cyan]Timing Breakdown Analysis:[/cyan]") + console.print("Check the logs above for detailed timing breakdown:") + console.print(" - recall: Time spent finding related observations") + console.print(" - llm: Time spent in LLM calls for consolidation decisions") + console.print(" - embedding: Time spent generating embeddings") + console.print(" - db_write: Time spent writing to database") + + +async def get_bank_stats(memory_engine: MemoryEngine, bank_id: str) -> dict[str, Any]: + """Get memory statistics for the bank.""" + pool = await memory_engine._get_pool() + from hindsight_api.engine.memory_engine import fq_table + + async with pool.acquire() as conn: + # Count memories by fact type + stats = await conn.fetch( + f""" + SELECT fact_type, COUNT(*) as count + FROM {fq_table("memory_units")} + WHERE bank_id = $1 + GROUP BY fact_type + """, + bank_id, + ) + + return {row["fact_type"]: row["count"] for row in stats} + + +def display_results_table(metrics: dict[str, Any], stats_before: dict, stats_after: dict) -> None: + """Display benchmark results in a formatted table.""" + table = Table(title="Consolidation Benchmark Results") + table.add_column("Metric", style="cyan") + table.add_column("Value", style="green") + + table.add_row("Total Time", f"{metrics['total_time']:.2f}s") + table.add_row("Memories Processed", str(metrics["memories_processed"])) + table.add_row("Throughput", f"{metrics['ops_per_sec']:.2f} op/sec") + table.add_row( + "Avg Time/Memory", + f"{metrics['total_time'] / metrics['memories_processed']:.3f}s" if metrics["memories_processed"] > 0 else "N/A", + ) + + result = metrics["consolidation_result"] + table.add_row("", "") # Separator + table.add_row("Observations Created", str(result.get("observations_created", 0))) + table.add_row("Observations Updated", str(result.get("observations_updated", 0))) + table.add_row("Observations Merged", str(result.get("observations_merged", 0))) + table.add_row("Skipped (No Durable Knowledge)", str(result.get("skipped", 0))) + + table.add_row("", "") # Separator + table.add_row("Memories Before", str(stats_before.get("experience", 0) + stats_before.get("world", 0))) + table.add_row("Observations After", str(stats_after.get("observation", 0))) + + console.print("\n") + console.print(table) + + +async def main(): + """Run the consolidation benchmark.""" + console.print("\n[bold cyan]Consolidation Performance Benchmark[/bold cyan]") + console.print("=" * 80) + + # Configuration + num_memories = int(os.getenv("NUM_MEMORIES", "100")) + bank_id = f"consolidation-bench-{uuid.uuid4().hex[:8]}" + + console.print("\n[cyan]Configuration:[/cyan]") + console.print(f" Bank ID: {bank_id}") + console.print(f" Number of memories: {num_memories}") + console.print(f" LLM Provider: {os.getenv('HINDSIGHT_API_LLM_PROVIDER', 'not set')}") + console.print(f" LLM Model: {os.getenv('HINDSIGHT_API_LLM_MODEL', 'not set')}") + + # Check if consolidation is enabled + config = get_config() + if not config.enable_observations: + console.print("\n[red]ERROR: Consolidation is disabled (enable_observations=False)[/red]") + console.print("Set HINDSIGHT_API_ENABLE_OBSERVATIONS=true to enable consolidation") + return + + # Initialize memory engine + console.print("\n[1] Initializing memory engine...") + memory = MemoryEngine( + db_url=os.getenv("HINDSIGHT_API_DATABASE_URL", "pg0"), + memory_llm_provider=os.getenv("HINDSIGHT_API_LLM_PROVIDER", "groq"), + memory_llm_api_key=os.getenv("HINDSIGHT_API_LLM_API_KEY"), + memory_llm_model=os.getenv("HINDSIGHT_API_LLM_MODEL", "openai/gpt-oss-120b"), + memory_llm_base_url=os.getenv("HINDSIGHT_API_LLM_BASE_URL") or None, + ) + await memory.initialize() + console.print("[green]✓[/green] Memory engine initialized") + + try: + # Create bank + console.print("\n[2] Creating test bank...") + await memory.get_bank_profile(bank_id=bank_id, request_context=RequestContext()) + console.print(f"[green]✓[/green] Created bank: {bank_id}") + + # Get initial stats + stats_before = await get_bank_stats(memory, bank_id) + + # Create test memories + console.print("\n[3] Creating test memories...") + await create_test_memories(memory, bank_id, num_memories) + + # Run consolidation benchmark + console.print("\n[4] Running consolidation benchmark...") + metrics = await run_consolidation_benchmark(memory, bank_id, enable_detailed_logs=True) + + # Get final stats + stats_after = await get_bank_stats(memory, bank_id) + + # Analyze timing + console.print("\n[5] Analyzing performance...") + await analyze_timing_breakdown(bank_id) + + # Display results + console.print("\n[6] Results:") + display_results_table(metrics, stats_before, stats_after) + + # Save results to file + output_dir = Path("benchmarks/results") + output_dir.mkdir(parents=True, exist_ok=True) + output_file = output_dir / f"consolidation_benchmark_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" + + results = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "config": { + "num_memories": num_memories, + "bank_id": bank_id, + "llm_provider": os.getenv("HINDSIGHT_API_LLM_PROVIDER"), + "llm_model": os.getenv("HINDSIGHT_API_LLM_MODEL"), + }, + "metrics": metrics, + "stats_before": stats_before, + "stats_after": stats_after, + } + + with open(output_file, "w") as f: + json.dump(results, f, indent=2, default=str) + + console.print(f"\n[green]✓[/green] Results saved to: {output_file}") + + finally: + # Cleanup + console.print("\n[7] Cleaning up...") + await memory.delete_bank(bank_id, request_context=RequestContext()) + console.print(f"[green]✓[/green] Deleted bank: {bank_id}") + + # Close memory engine connections + pool = await memory._get_pool() + await pool.close() + console.print("[green]✓[/green] Memory engine connections closed") + + console.print("\n[bold green]✓ Benchmark Complete![/bold green]\n") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/hindsight-docs/docs/developer/configuration.md b/hindsight-docs/docs/developer/configuration.md index 52af0268..f646413e 100644 --- a/hindsight-docs/docs/developer/configuration.md +++ b/hindsight-docs/docs/developer/configuration.md @@ -378,6 +378,7 @@ Observations are consolidated knowledge synthesized from facts. |----------|-------------|---------| | `HINDSIGHT_API_ENABLE_OBSERVATIONS` | Enable observation consolidation | `true` | | `HINDSIGHT_API_CONSOLIDATION_BATCH_SIZE` | Memories to load per batch (internal optimization) | `50` | +| `HINDSIGHT_API_CONSOLIDATION_MAX_TOKENS` | Max tokens for recall when finding related observations during consolidation | `1024` | | `HINDSIGHT_API_RETAIN_OBSERVATIONS_ASYNC` | Run observation generation asynchronously (after retain completes) | `false` | ### Reflect diff --git a/scripts/benchmarks/run-consolidation.sh b/scripts/benchmarks/run-consolidation.sh new file mode 100755 index 00000000..76873135 --- /dev/null +++ b/scripts/benchmarks/run-consolidation.sh @@ -0,0 +1,33 @@ +#!/bin/bash +set -e + +# Consolidation Performance Benchmark Runner +# Measures consolidation throughput (op/sec) and identifies bottlenecks + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)" + +# Source .env if it exists +if [ -f "$REPO_ROOT/.env" ]; then + source "$REPO_ROOT/.env" + echo "Loaded environment from .env" +fi + +# Default configuration +NUM_MEMORIES="${NUM_MEMORIES:-100}" + +# Enable observations (required for consolidation) +export HINDSIGHT_API_ENABLE_OBSERVATIONS=true + +echo "Running consolidation benchmark with configuration:" +echo " NUM_MEMORIES=$NUM_MEMORIES" +echo " HINDSIGHT_API_LLM_PROVIDER=${HINDSIGHT_API_LLM_PROVIDER:-not set}" +echo " HINDSIGHT_API_LLM_MODEL=${HINDSIGHT_API_LLM_MODEL:-not set}" +echo "" + +# Run benchmark +cd "$REPO_ROOT" +uv run python -m benchmarks.consolidation.consolidation_benchmark + +echo "" +echo "Benchmark complete! Check benchmarks/results/ for detailed results."