holetron-lab/fleet-memory
1
0
Fork 0
Code Releases Packages Activity Actions

fix web init

Browse source
This commit is contained in:
Nicolò Boschi 2025-11-07 12:47:50 +01:00
parent fcc5250656
commit 18c7e0845b
7 changed files with 321 additions and 36 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

178
benchmarks/common/benchmark_runner.py
View file

@ -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")

13
benchmarks/longmemeval/longmemeval_benchmark.py
View file

@ -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(

14
generate_openapi.py
View file

@ -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()

4
memora/utils.py
View file

@ -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

9
memora/web/__init__.py
View file

@ -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"]

91
memora/web/server.py
View file

@ -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)

48
scripts/start-server.sh
View file

@ -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[@]}"