From 18c7e0845bc145b3d2f34536550473ec525303fa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicol=C3=B2=20Boschi?= Date: Fri, 7 Nov 2025 12:47:50 +0100 Subject: [PATCH] fix web init --- benchmarks/common/benchmark_runner.py | 178 +++++++++++++++++- .../longmemeval/longmemeval_benchmark.py | 13 +- generate_openapi.py | 14 +- memora/utils.py | 4 +- memora/web/__init__.py | 9 +- memora/web/server.py | 91 +++++++-- scripts/start-server.sh | 48 ++++- 7 files changed, 321 insertions(+), 36 deletions(-) diff --git a/benchmarks/common/benchmark_runner.py b/benchmarks/common/benchmark_runner.py index ced4c3ac..8ec96a86 100644 --- a/benchmarks/common/benchmark_runner.py +++ b/benchmarks/common/benchmark_runner.py @@ -229,6 +229,37 @@ class BenchmarkRunner: memory_llm_base_url=os.getenv("MEMORY_LLM_BASE_URL") or None, # Use None to get provider defaults ) + def calculate_data_stats(self, items: List[Dict[str, Any]]) -> Dict[str, Any]: + """ + Calculate statistics about the data to be ingested. + + Returns: + Dict with statistics: total_sessions, total_chars, avg_session_length, etc. + """ + total_sessions = 0 + total_chars = 0 + session_lengths = [] + + for item in items: + batch_contents = self.dataset.prepare_sessions_for_ingestion(item) + total_sessions += len(batch_contents) + + for session in batch_contents: + content_len = len(session['content']) + total_chars += content_len + session_lengths.append(content_len) + + avg_length = total_chars / total_sessions if total_sessions > 0 else 0 + + return { + 'total_sessions': total_sessions, + 'total_chars': total_chars, + 'total_items': len(items), + 'avg_session_length': avg_length, + 'min_session_length': min(session_lengths) if session_lengths else 0, + 'max_session_length': max(session_lengths) if session_lengths else 0 + } + async def ingest_conversation( self, item: Dict[str, Any], @@ -547,6 +578,7 @@ class BenchmarkRunner: eval_semaphore_size: int = 8, clear_agent_per_item: bool = False, specific_item: Optional[str] = None, + separate_ingestion_phase: bool = False, ) -> Dict[str, Any]: """ Run the full benchmark evaluation. @@ -561,8 +593,9 @@ class BenchmarkRunner: skip_ingestion: Skip ingestion and use existing data max_concurrent_questions: Max concurrent question processing eval_semaphore_size: Max concurrent LLM judge requests - clear_agent_per_item: Clear agent data before each item (for isolation) + clear_agent_per_item: Use unique agent ID per item for isolation (deprecated when separate_ingestion_phase=True) specific_item: If provided, only run this specific item ID (e.g., conversation) + separate_ingestion_phase: If True, ingest all data first, then evaluate all questions (single agent) Returns: Dict with complete benchmark results @@ -588,6 +621,35 @@ class BenchmarkRunner: console.print(f"\n[2] Initializing memory system...") console.print(f" [green]✓[/green] Memory system initialized") + if separate_ingestion_phase: + # New two-phase approach: ingest all, then evaluate all + return await self._run_two_phase( + items, agent_id, thinking_budget, max_tokens, + skip_ingestion, max_questions_per_item, + max_concurrent_questions, eval_semaphore_size + ) + else: + # Original approach: process each item independently + return await self._run_single_phase( + items, agent_id, thinking_budget, max_tokens, + skip_ingestion, max_questions_per_item, + max_concurrent_questions, eval_semaphore_size, + clear_agent_per_item + ) + + async def _run_single_phase( + self, + items: List[Dict[str, Any]], + agent_id: str, + thinking_budget: int, + max_tokens: int, + skip_ingestion: bool, + max_questions_per_item: Optional[int], + max_concurrent_questions: int, + eval_semaphore_size: int, + clear_agent_per_item: bool, + ) -> Dict[str, Any]: + """Original single-phase approach: process each item independently.""" # Create semaphore for question processing question_semaphore = asyncio.Semaphore(max_concurrent_questions) @@ -595,12 +657,12 @@ class BenchmarkRunner: all_results = [] for i, item in enumerate(items, 1): - # Clear agent per item if requested (for isolation in benchmarks like LongMemEval) - if clear_agent_per_item and i > 1 and not skip_ingestion: - await self.memory.delete_agent(agent_id) + # Use unique agent ID per item if requested (for isolation in benchmarks like LongMemEval) + # This avoids deadlocks from deleting agent data + item_agent_id = f"{agent_id}_item_{i-1}" if clear_agent_per_item else agent_id result = await self.process_single_item( - item, agent_id, i, len(items), + item, item_agent_id, i, len(items), thinking_budget, max_tokens, max_questions_per_item, skip_ingestion, question_semaphore, eval_semaphore_size, ) @@ -624,6 +686,112 @@ class BenchmarkRunner: 'item_results': all_results } + async def _run_two_phase( + self, + items: List[Dict[str, Any]], + agent_id: str, + thinking_budget: int, + max_tokens: int, + skip_ingestion: bool, + max_questions_per_item: Optional[int], + max_concurrent_questions: int, + eval_semaphore_size: int, + ) -> Dict[str, Any]: + """ + Two-phase approach: ingest all data into single agent, then evaluate all questions. + + More realistic scenario where agent accumulates memories over time. + """ + # Phase 1: Ingestion + if not skip_ingestion: + # Calculate and display data statistics + console.print(f"\n[3] Analyzing data to be ingested...") + stats = self.calculate_data_stats(items) + console.print(f" [cyan]Total items:[/cyan] {stats['total_items']}") + console.print(f" [cyan]Total sessions:[/cyan] {stats['total_sessions']}") + console.print(f" [cyan]Total characters:[/cyan] {stats['total_chars']:,}") + console.print(f" [cyan]Avg session length:[/cyan] {stats['avg_session_length']:.0f} chars") + console.print(f" [cyan]Session length range:[/cyan] {stats['min_session_length']}-{stats['max_session_length']} chars") + + console.print(f"\n[4] Phase 1: Ingesting all data into agent '{agent_id}'...") + console.print(f" [yellow]Clearing previous agent data...[/yellow]") + await self.memory.delete_agent(agent_id) + console.print(f" [green]✓[/green] Cleared agent data") + + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + BarColumn(), + TextColumn("[progress.percentage]{task.percentage:>3.0f}%"), + console=console + ) as progress: + task = progress.add_task( + f"[cyan]Ingesting {len(items)} items...", + total=len(items) + ) + + total_sessions = 0 + for item in items: + num_sessions = await self.ingest_conversation(item, agent_id) + total_sessions += num_sessions + progress.update(task, advance=1) + + console.print(f" [green]✓[/green] Ingested {total_sessions} sessions from {len(items)} items") + else: + console.print(f"\n[3] Skipping ingestion (using existing data)") + + # Phase 2: Evaluation + console.print(f"\n[5] Phase 2: Evaluating all questions...") + + # Create semaphore for question processing + question_semaphore = asyncio.Semaphore(max_concurrent_questions) + + all_results = [] + for i, item in enumerate(items, 1): + item_id = self.dataset.get_item_id(item) + console.print(f"\n[bold blue]Item {i}/{len(items)}[/bold blue] (ID: {item_id})") + + # Get QA pairs + qa_pairs = self.dataset.get_qa_pairs(item) + console.print(f" Evaluating {len(qa_pairs)} QA pairs (parallel)...") + + qa_results = await self.evaluate_qa_task( + agent_id, + qa_pairs, + item_id, + thinking_budget, + max_tokens, + max_questions_per_item, + question_semaphore, + ) + + # Calculate metrics + metrics = await self.calculate_metrics(qa_results, eval_semaphore_size) + console.print(f" [green]✓[/green] Accuracy: {metrics['accuracy']:.2f}% ({metrics['correct']}/{metrics['total']})") + + all_results.append({ + 'item_id': item_id, + 'metrics': metrics, + 'num_sessions': -1 # Not tracked in two-phase mode + }) + + # Calculate overall metrics + total_correct = sum(r['metrics']['correct'] for r in all_results) + total_questions = sum(r['metrics']['total'] for r in all_results) + total_invalid = sum(r['metrics'].get('invalid', 0) for r in all_results) + total_valid = total_questions - total_invalid + overall_accuracy = (total_correct / total_valid * 100) if total_valid > 0 else 0 + + return { + 'overall_accuracy': overall_accuracy, + 'total_correct': total_correct, + 'total_questions': total_questions, + 'total_invalid': total_invalid, + 'total_valid': total_valid, + 'num_items': len(items), + 'item_results': all_results + } + def display_results(self, results: Dict[str, Any]): """Display benchmark results in a formatted table.""" console.print("\n[bold green]✓ Benchmark Complete![/bold green]\n") diff --git a/benchmarks/longmemeval/longmemeval_benchmark.py b/benchmarks/longmemeval/longmemeval_benchmark.py index ddfa7eb4..c4fb85ef 100644 --- a/benchmarks/longmemeval/longmemeval_benchmark.py +++ b/benchmarks/longmemeval/longmemeval_benchmark.py @@ -6,6 +6,9 @@ Provides dataset, answer generator, and evaluator for the LongMemEval benchmark. import sys from pathlib import Path +from benchmarks.common.benchmark_runner import BenchmarkRunner +from memora import TemporalSemanticMemory + # Add parent directory to path for imports sys.path.insert(0, str(Path(__file__).parent.parent.parent)) @@ -227,7 +230,8 @@ async def run_benchmark( ) # Run benchmark - # Note: LongMemEval requires clearing agent per item for isolation + # Two-phase approach: ingest all 500 conversations into single agent, then evaluate all questions + # This is more realistic and tests retrieval from a large memory base results = await runner.run( dataset_path=dataset_path, agent_id="longmemeval", @@ -236,9 +240,9 @@ async def run_benchmark( thinking_budget=thinking_budget, max_tokens=max_tokens, skip_ingestion=skip_ingestion, - max_concurrent_questions=8, # Lower for LongMemEval (each has full conversation) + max_concurrent_questions=8, eval_semaphore_size=8, - clear_agent_per_item=True # Clear agent data per item for isolation + separate_ingestion_phase=True # Ingest all data first, then evaluate all questions ) # Display and save results @@ -268,6 +272,9 @@ def download_dataset(dataset_path: Path) -> bool: console.print(f"[dim]URL: {url}[/dim]") console.print(f"[dim]Destination: {dataset_path}[/dim]") + # Create parent directory if it doesn't exist + dataset_path.parent.mkdir(parents=True, exist_ok=True) + try: # Use curl to download with progress result = subprocess.run( diff --git a/generate_openapi.py b/generate_openapi.py index b77ae8f7..6019a0e0 100644 --- a/generate_openapi.py +++ b/generate_openapi.py @@ -6,15 +6,27 @@ This script imports the FastAPI app and exports its OpenAPI schema to a JSON fil """ import json import sys +import os from pathlib import Path # Add parent directory to path to import memory module sys.path.insert(0, str(Path(__file__).parent)) -from memora.web.server import app +from memora.web.server import create_app +from memora import TemporalSemanticMemory def generate_openapi_spec(output_path: str = "openapi.json"): """Generate OpenAPI spec and save to file.""" + # Create a temporary memory instance for OpenAPI generation + _memory = TemporalSemanticMemory( + db_url=os.getenv("DATABASE_URL", "postgresql://user:pass@localhost:5432/memora"), + memory_llm_provider=os.getenv("MEMORY_LLM_PROVIDER", "groq"), + memory_llm_api_key=os.getenv("MEMORY_LLM_API_KEY", "dummy"), + memory_llm_model=os.getenv("MEMORY_LLM_MODEL", "openai/gpt-oss-120b"), + memory_llm_base_url=os.getenv("MEMORY_LLM_BASE_URL") or None, + ) + app = create_app(_memory) + # Get the OpenAPI schema from the app openapi_schema = app.openapi() diff --git a/memora/utils.py b/memora/utils.py index 30415bee..8f7cf052 100644 --- a/memora/utils.py +++ b/memora/utils.py @@ -1,6 +1,7 @@ """ Utility functions for memory system. """ +import logging from datetime import datetime from typing import List, Dict, TYPE_CHECKING @@ -38,7 +39,8 @@ async def extract_facts(text: str, event_date: datetime, context: str = "", llm_ fact_dicts = await extract_facts_from_text(text, event_date, context, llm_config=llm_config) if not fact_dicts: - raise Exception(f"LLM extracted 0 facts from text of length {len(text)}. This may indicate the text contains no meaningful information, or the LLM failed to extract facts.") + logging.warning(f"LLM extracted 0 facts from text of length {len(text)}. This may indicate the text contains no meaningful information, or the LLM failed to extract facts. Full text: {text}") + return [] return fact_dicts diff --git a/memora/web/__init__.py b/memora/web/__init__.py index 924cf08d..3a3d1597 100644 --- a/memora/web/__init__.py +++ b/memora/web/__init__.py @@ -3,6 +3,13 @@ Web interface for memory system. Provides FastAPI app and visualization interface. """ -from .server import app, create_app +from .server import create_app + +# Lazy import of app to avoid initialization issues +def __getattr__(name): + if name == "app": + from .server import _get_app + return _get_app() + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") __all__ = ["app", "create_app"] diff --git a/memora/web/server.py b/memora/web/server.py index bd3ca259..516f0602 100644 --- a/memora/web/server.py +++ b/memora/web/server.py @@ -662,30 +662,83 @@ def _register_routes(app: FastAPI): -# Create default app instance -# Initialize memory system with environment variables -_memory = TemporalSemanticMemory( - db_url=os.getenv("DATABASE_URL"), - memory_llm_provider=os.getenv("MEMORY_LLM_PROVIDER", "groq"), - memory_llm_api_key=os.getenv("MEMORY_LLM_API_KEY"), - memory_llm_model=os.getenv("MEMORY_LLM_MODEL", "openai/gpt-oss-120b"), - memory_llm_base_url=os.getenv("MEMORY_LLM_BASE_URL") or None, # Use None to get provider defaults -) -app = create_app(_memory) +def _create_default_app(): + """Create app instance with default environment configuration.""" + _memory = TemporalSemanticMemory( + db_url=os.getenv("DATABASE_URL"), + memory_llm_provider=os.getenv("MEMORY_LLM_PROVIDER", "groq"), + memory_llm_api_key=os.getenv("MEMORY_LLM_API_KEY"), + memory_llm_model=os.getenv("MEMORY_LLM_MODEL", "openai/gpt-oss-120b"), + memory_llm_base_url=os.getenv("MEMORY_LLM_BASE_URL") or None, + ) + return create_app(_memory) + + +# Module-level app instance (lazy initialization) +_app_instance = None + + +def _get_app(): + """Get or create the app instance.""" + global _app_instance + if _app_instance is None: + _app_instance = _create_default_app() + return _app_instance if __name__ == "__main__": import uvicorn + import argparse + + # Parse CLI arguments + parser = argparse.ArgumentParser(description="Memory Graph API Server") + parser.add_argument("--host", default="0.0.0.0", help="Host to bind to (default: 0.0.0.0)") + parser.add_argument("--port", type=int, default=8080, help="Port to bind to (default: 8080)") + parser.add_argument("--reload", action="store_true", help="Enable auto-reload on code changes") + parser.add_argument("--workers", type=int, default=1, help="Number of worker processes (default: 1)") + parser.add_argument("--log-level", default="info", choices=["critical", "error", "warning", "info", "debug", "trace"], + help="Log level (default: info)") + parser.add_argument("--access-log", action="store_true", help="Enable access log") + parser.add_argument("--no-access-log", dest="access_log", action="store_false", help="Disable access log") + parser.add_argument("--proxy-headers", action="store_true", help="Enable X-Forwarded-Proto, X-Forwarded-For headers") + parser.add_argument("--forwarded-allow-ips", default=None, help="Comma separated list of IPs to trust with proxy headers") + parser.add_argument("--ssl-keyfile", default=None, help="SSL key file") + parser.add_argument("--ssl-certfile", default=None, help="SSL certificate file") + parser.set_defaults(access_log=False) + + args = parser.parse_args() + + # Create app after parsing args (so --help works without DB connection) + app = _create_default_app() + print("\n" + "=" * 80) print("Memory Graph API Server") print("=" * 80) - print("\nStarting server at http://localhost:8080") - print("\nEndpoints:") - print(" GET / - Visualization UI") - print(" GET /api/graph - Get graph data") - print(" POST /api/search - Run search with trace") - print(" POST /api/memories/batch - Store multiple memories in batch") - print(" GET /api/agents - List available agents") - print("\n" + "=" * 80 + "\n") + print(f"Host: {args.host}") + print(f"Port: {args.port}") + print(f"Reload: {args.reload}") + print(f"Workers: {args.workers}") + print(f"Log Level: {args.log_level}") + print("=" * 80 + "\n") - uvicorn.run("memora.web.server:app", host="0.0.0.0", port=8080, reload=True) + # Prepare uvicorn config + uvicorn_config = { + "app": app, + "host": args.host, + "port": args.port, + "reload": args.reload, + "workers": args.workers, + "log_level": args.log_level, + "access_log": args.access_log, + "proxy_headers": args.proxy_headers, + } + + # Add optional parameters if provided + if args.forwarded_allow_ips: + uvicorn_config["forwarded_allow_ips"] = args.forwarded_allow_ips + if args.ssl_keyfile: + uvicorn_config["ssl_keyfile"] = args.ssl_keyfile + if args.ssl_certfile: + uvicorn_config["ssl_certfile"] = args.ssl_certfile + + uvicorn.run(**uvicorn_config) diff --git a/scripts/start-server.sh b/scripts/start-server.sh index b98c1c82..735ea1c8 100755 --- a/scripts/start-server.sh +++ b/scripts/start-server.sh @@ -5,6 +5,7 @@ cd "$(dirname "$0")/.." # Parse arguments ENV_MODE="local" +SERVER_ARGS=() while [[ $# -gt 0 ]]; do case $1 in @@ -16,13 +17,34 @@ while [[ $# -gt 0 ]]; do fi shift 2 ;; - *) - echo "Usage: $0 [--env local|dev]" + --help|-h) + echo "Usage: $0 [--env local|dev] [uvicorn options...]" echo "" echo "Options:" echo " --env local Use local environment (default)" echo " --env dev Use dev environment" - exit 1 + echo "" + echo "Uvicorn options (passed to server):" + echo " --host HOST Host to bind to (default: 0.0.0.0)" + echo " --port PORT Port to bind to (default: 8080)" + echo " --reload Enable auto-reload on code changes" + echo " --workers WORKERS Number of worker processes (default: 1)" + echo " --log-level LEVEL Log level: critical/error/warning/info/debug/trace" + echo " --access-log Enable access log" + echo " --no-access-log Disable access log" + echo " --proxy-headers Enable X-Forwarded-Proto, X-Forwarded-For headers" + echo " --forwarded-allow-ips Comma separated list of IPs to trust" + echo " --ssl-keyfile FILE SSL key file" + echo " --ssl-certfile FILE SSL certificate file" + echo "" + echo "Example:" + echo " $0 --env dev --reload --port 8000 --log-level debug" + exit 0 + ;; + *) + # Pass all other arguments to the server + SERVER_ARGS+=("$1") + shift ;; esac done @@ -43,8 +65,22 @@ set -a source "$ENV_FILE" set +a -echo "Server will be available at: http://localhost:8080" +# Extract port from SERVER_ARGS if provided, otherwise use default +PORT=8080 +for ((i=0; i<${#SERVER_ARGS[@]}; i++)); do + if [[ "${SERVER_ARGS[$i]}" == "--port" ]]; then + PORT="${SERVER_ARGS[$((i+1))]}" + break + fi +done + +echo "Server will be available at: http://localhost:${PORT}" echo "" -open http://localhost:8080 -uv run uvicorn memora.web.server:app --reload --host 0.0.0.0 --port 8080 +# Set default arguments if not provided +if [[ ${#SERVER_ARGS[@]} -eq 0 ]]; then + SERVER_ARGS=(--reload --host 0.0.0.0 --port 8080) +fi + +open "http://localhost:${PORT}" +uv run python -m memora.web.server "${SERVER_ARGS[@]}"