Compare commits

...
1 Commits
egn ... ma
Author SHA1 Message Date
Nicolò Boschi c949191953 feat: add real-time timing breakdown logging for consolidation
- Log timing breakdown after each batch (every 50 memories by default)
- Log timing breakdown in progress logs (every 10 memories)
- Shows recall, llm, embedding, db_write times incrementally
- Includes avg time per memory for quick diagnosis
- Helps diagnose performance issues in production without waiting for job completion

Example output (every 10 memories):
[CONSOLIDATION] bank=xyz progress: 10/39303 memories processed | recall=2.09s, llm=11.03s, embedding=0.48s, db_write=0.02s

Example output (per batch):
[CONSOLIDATION] bank=xyz batch 1/50 memories: recall=7.3s, llm=57.5s, embedding=2.0s, db_write=0.09s | avg=1.3s/memory
2026-01-29 17:37:01 +01:00
2 changed files with 36 additions and 7 deletions
@@ -144,10 +144,14 @@ async def run_consolidation_job(
}
batch_num = 0
last_progress_timings = {} # Track timings at last progress log
while True:
batch_num += 1
batch_start = time.time()
# Snapshot timings at batch start for per-batch calculation
batch_start_timings = perf.timings.copy()
# Fetch next batch of unconsolidated memories
async with pool.acquire() as conn:
t0 = time.time()
@@ -217,19 +221,44 @@ async def run_consolidation_job(
elif action == "skipped":
stats["skipped"] += 1
# Log progress periodically
# Log progress periodically with timing breakdown
if stats["memories_processed"] % 10 == 0:
# Calculate timing deltas since last progress log
timing_parts = []
for key in ["recall", "llm", "embedding", "db_write"]:
if key in perf.timings:
delta = perf.timings[key] - last_progress_timings.get(key, 0)
timing_parts.append(f"{key}={delta:.2f}s")
timing_str = f" | {', '.join(timing_parts)}" if timing_parts else ""
logger.info(
f"[CONSOLIDATION] bank={bank_id} progress: "
f"{stats['memories_processed']}/{total_count} memories processed"
f"{stats['memories_processed']}/{total_count} memories processed{timing_str}"
)
# Update last progress snapshot
last_progress_timings = perf.timings.copy()
batch_time = time.time() - batch_start
perf.log(
f"[2] Batch {batch_num}: {len(memories)} memories in {batch_time:.3f}s "
f"(avg {batch_time / len(memories):.3f}s/memory)"
)
# Log timing breakdown after each batch (delta from batch start)
timing_parts = []
for key in ["recall", "llm", "embedding", "db_write"]:
if key in perf.timings:
delta = perf.timings[key] - batch_start_timings.get(key, 0)
timing_parts.append(f"{key}={delta:.3f}s")
if timing_parts:
avg_per_memory = batch_time / len(memories) if memories else 0
logger.info(
f"[CONSOLIDATION] bank={bank_id} batch {batch_num}/{len(memories)} memories: "
f"{', '.join(timing_parts)} | avg={avg_per_memory:.3f}s/memory"
)
# Build summary
perf.log(
f"[3] Results: {stats['memories_processed']} memories -> "
Generated
+5 -5
View File
@@ -1295,7 +1295,7 @@ wheels = [
[[package]]
name = "hindsight-all"
version = "0.4.0"
version = "0.4.1"
source = { editable = "hindsight" }
dependencies = [
{ name = "hindsight-api" },
@@ -1319,7 +1319,7 @@ provides-extras = ["test"]
[[package]]
name = "hindsight-api"
version = "0.4.0"
version = "0.4.1"
source = { editable = "hindsight-api" }
dependencies = [
{ name = "aiohttp" },
@@ -1447,7 +1447,7 @@ dev = [
[[package]]
name = "hindsight-client"
version = "0.4.0"
version = "0.4.1"
source = { editable = "hindsight-clients/python" }
dependencies = [
{ name = "aiohttp" },
@@ -1481,7 +1481,7 @@ provides-extras = ["test"]
[[package]]
name = "hindsight-dev"
version = "0.4.0"
version = "0.4.1"
source = { editable = "hindsight-dev" }
dependencies = [
{ name = "hindsight-api" },
@@ -1527,7 +1527,7 @@ dev = [
[[package]]
name = "hindsight-embed"
version = "0.4.0"
version = "0.4.1"
source = { editable = "hindsight-embed" }
dependencies = [
{ name = "httpx" },