diff --git a/hindsight-api/hindsight_api/config.py b/hindsight-api/hindsight_api/config.py index b2b6560f..be58c0d2 100644 --- a/hindsight-api/hindsight_api/config.py +++ b/hindsight-api/hindsight_api/config.py @@ -285,6 +285,7 @@ ENV_FILE_DELETE_AFTER_RETAIN = "HINDSIGHT_API_FILE_DELETE_AFTER_RETAIN" # Observations settings (consolidated knowledge from facts) ENV_ENABLE_OBSERVATIONS = "HINDSIGHT_API_ENABLE_OBSERVATIONS" ENV_CONSOLIDATION_BATCH_SIZE = "HINDSIGHT_API_CONSOLIDATION_BATCH_SIZE" +ENV_CONSOLIDATION_LLM_BATCH_SIZE = "HINDSIGHT_API_CONSOLIDATION_LLM_BATCH_SIZE" ENV_CONSOLIDATION_MAX_TOKENS = "HINDSIGHT_API_CONSOLIDATION_MAX_TOKENS" ENV_OBSERVATIONS_MISSION = "HINDSIGHT_API_OBSERVATIONS_MISSION" @@ -426,7 +427,8 @@ DEFAULT_FILE_DELETE_AFTER_RETAIN = True # Delete file bytes after retain (saves # 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 +DEFAULT_CONSOLIDATION_LLM_BATCH_SIZE = 8 # Facts per LLM call (1 = no batching; >1 = batch mode) +DEFAULT_CONSOLIDATION_MAX_TOKENS = 512 # Max tokens for recall when finding related observations DEFAULT_OBSERVATIONS_MISSION = None # Declarative spec of what observations are for this bank # Database migrations @@ -679,6 +681,7 @@ class HindsightConfig: # Observations settings (consolidated knowledge from facts) enable_observations: bool consolidation_batch_size: int + consolidation_llm_batch_size: int consolidation_max_tokens: int observations_mission: str | None @@ -1097,6 +1100,9 @@ class HindsightConfig: consolidation_batch_size=int( os.getenv(ENV_CONSOLIDATION_BATCH_SIZE, str(DEFAULT_CONSOLIDATION_BATCH_SIZE)) ), + consolidation_llm_batch_size=int( + os.getenv(ENV_CONSOLIDATION_LLM_BATCH_SIZE, str(DEFAULT_CONSOLIDATION_LLM_BATCH_SIZE)) + ), consolidation_max_tokens=int( os.getenv(ENV_CONSOLIDATION_MAX_TOKENS, str(DEFAULT_CONSOLIDATION_MAX_TOKENS)) ), diff --git a/hindsight-api/hindsight_api/engine/consolidation/consolidator.py b/hindsight-api/hindsight_api/engine/consolidation/consolidator.py index 3eae314f..966f8917 100644 --- a/hindsight-api/hindsight_api/engine/consolidation/consolidator.py +++ b/hindsight-api/hindsight_api/engine/consolidation/consolidator.py @@ -15,6 +15,7 @@ import json import logging import time import uuid +from dataclasses import dataclass, field from datetime import datetime, timezone from typing import TYPE_CHECKING, Any @@ -23,7 +24,7 @@ from pydantic import BaseModel from ...config import get_config from ..memory_engine import fq_table from ..retain import embedding_utils -from .prompts import build_consolidation_prompt +from .prompts import build_batch_consolidation_prompt if TYPE_CHECKING: from asyncpg import Connection @@ -35,15 +36,34 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -class _ConsolidationAction(BaseModel): - action: str # "update" | "create" +class _CreateAction(BaseModel): text: str - reason: str = "" - learning_id: str | None = None # required for "update" actions + source_fact_ids: list[str] # memory UUIDs from the NEW FACTS list -class _ConsolidationResponse(BaseModel): - actions: list[_ConsolidationAction] +class _UpdateAction(BaseModel): + text: str + observation_id: str # UUID of the existing observation to update + source_fact_ids: list[str] # memory UUIDs from the NEW FACTS list + + +class _DeleteAction(BaseModel): + observation_id: str # UUID of the observation to remove + + +class _ConsolidationBatchResponse(BaseModel): + creates: list[_CreateAction] = [] + updates: list[_UpdateAction] = [] + deletes: list[_DeleteAction] = [] + + +@dataclass +class _BatchLLMResult: + creates: list[_CreateAction] = field(default_factory=list) + updates: list[_UpdateAction] = field(default_factory=list) + deletes: list[_DeleteAction] = field(default_factory=list) + obs_count: int = 0 + prompt_chars: int = 0 class ConsolidationPerfLog: @@ -54,6 +74,9 @@ class ConsolidationPerfLog: self.start_time = time.time() self.lines: list[str] = [] self.timings: dict[str, float] = {} + self.llm_calls: int = 0 + self.total_obs_in_context: int = 0 + self.total_prompt_chars: int = 0 def log(self, message: str) -> None: """Add a log line.""" @@ -66,6 +89,12 @@ class ConsolidationPerfLog: else: self.timings[key] = duration + def record_llm_call(self, obs_count: int, prompt_chars: int) -> None: + """Record stats for a single LLM call.""" + self.llm_calls += 1 + self.total_obs_in_context += obs_count + self.total_prompt_chars += prompt_chars + def flush(self) -> None: """Flush all log lines to the logger.""" total_time = time.time() - self.start_time @@ -98,6 +127,7 @@ async def run_consolidation_job( config = await memory_engine._config_resolver.resolve_full_config(bank_id, request_context) perf = ConsolidationPerfLog(bank_id) max_memories_per_batch = config.consolidation_batch_size + llm_batch_size = max(1, config.consolidation_llm_batch_size) # Check if consolidation is enabled if not config.enable_observations: @@ -156,15 +186,8 @@ async def run_consolidation_job( # Track all unique tags from consolidated memories for mental model refresh filtering consolidated_tags: set[str] = set() - batch_num = 0 - last_progress_timings = {} # Track timings at last progress log + llm_batch_num = 0 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() @@ -186,95 +209,90 @@ async def run_consolidation_job( if not memories: break # No more unconsolidated memories - for memory in memories: - mem_start = time.time() + # Group memories by exact tag set before batching — security requirement: + # memories with different tags must never share an LLM call. + tag_groups: dict[tuple[str, ...], list[dict[str, Any]]] = {} + for m in memories: + tag_key = tuple(sorted(m.get("tags") or [])) + tag_groups.setdefault(tag_key, []).append(dict(m)) - # Track tags from this memory for mental model refresh filtering - memory_tags = memory.get("tags") or [] - if memory_tags: - consolidated_tags.update(memory_tags) + # Flatten into LLM batches respecting both tag groups and llm_batch_size + llm_batches: list[list[dict[str, Any]]] = [] + for group in tag_groups.values(): + for i in range(0, len(group), llm_batch_size): + llm_batches.append(group[i : i + llm_batch_size]) + + for llm_batch in llm_batches: + llm_batch_num += 1 + llm_batch_start = time.time() + + # Snapshot perf and stats before this LLM batch + snap_timings = perf.timings.copy() + snap_llm_calls = perf.llm_calls + snap_total_chars = perf.total_prompt_chars + snap_stats = stats.copy() + + # Track tags for mental model refresh filtering + for memory in llm_batch: + memory_tags = memory.get("tags") or [] + if memory_tags: + consolidated_tags.update(memory_tags) - # Process the memory (uses its own connection internally) async with pool.acquire() as conn: - result = await _process_memory( + results = await _process_memory_batch( conn=conn, memory_engine=memory_engine, bank_id=bank_id, - memory=dict(memory), + memories=llm_batch, request_context=request_context, perf=perf, config=config, ) - - # Mark memory as consolidated (committed immediately) - await conn.execute( - f""" - UPDATE {fq_table("memory_units")} - SET consolidated_at = NOW() - WHERE id = $1 - """, - memory["id"], + await conn.executemany( + f"UPDATE {fq_table('memory_units')} SET consolidated_at = NOW() WHERE id = $1", + [(m["id"],) for m in llm_batch], ) - mem_time = time.time() - mem_start - perf.record_timing("process_memory_total", mem_time) + for result in results: + stats["memories_processed"] += 1 + action = result.get("action") + if action == "created": + stats["observations_created"] += 1 + stats["actions_executed"] += 1 + elif action == "updated": + stats["observations_updated"] += 1 + stats["actions_executed"] += 1 + elif action == "merged": + stats["observations_merged"] += 1 + stats["actions_executed"] += 1 + elif action == "multiple": + stats["observations_created"] += result.get("created", 0) + stats["observations_updated"] += result.get("updated", 0) + stats["observations_merged"] += result.get("merged", 0) + stats["actions_executed"] += result.get("total_actions", 0) + elif action == "skipped": + stats["skipped"] += 1 - stats["memories_processed"] += 1 - - action = result.get("action") - if action == "created": - stats["observations_created"] += 1 - stats["actions_executed"] += 1 - elif action == "updated": - stats["observations_updated"] += 1 - stats["actions_executed"] += 1 - elif action == "merged": - stats["observations_merged"] += 1 - stats["actions_executed"] += 1 - elif action == "multiple": - stats["observations_created"] += result.get("created", 0) - stats["observations_updated"] += result.get("updated", 0) - stats["observations_merged"] += result.get("merged", 0) - stats["actions_executed"] += result.get("total_actions", 0) - elif action == "skipped": - stats["skipped"] += 1 - - # 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{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 + # Per-LLM-batch log + llm_batch_time = time.time() - llm_batch_start + timing_parts = [] + for key in ["recall", "llm", "embedding", "db_write"]: + if key in perf.timings: + delta = perf.timings[key] - snap_timings.get(key, 0) + timing_parts.append(f"{key}={delta:.3f}s") + input_tokens = int((perf.total_prompt_chars - snap_total_chars) / 4) + batch_created = stats["observations_created"] - snap_stats["observations_created"] + batch_updated = stats["observations_updated"] - snap_stats["observations_updated"] + batch_skipped = stats["skipped"] - snap_stats["skipped"] + llm_calls_made = perf.llm_calls - snap_llm_calls logger.info( - f"[CONSOLIDATION] bank={bank_id} batch {batch_num}/{len(memories)} memories: " - f"{', '.join(timing_parts)} | avg={avg_per_memory:.3f}s/memory" + f"[CONSOLIDATION] bank={bank_id} llm_batch #{llm_batch_num}" + f" ({len(llm_batch)} memories, {llm_calls_made} llm calls)" + f" | {stats['memories_processed']}/{total_count} processed" + f" | {', '.join(timing_parts)}" + f" | created={batch_created} updated={batch_updated} skipped={batch_skipped}" + f" | input_tokens=~{input_tokens}" + f" | avg={llm_batch_time / len(llm_batch):.3f}s/memory" ) # Build summary @@ -298,6 +316,10 @@ async def run_consolidation_job( if "db_write" in perf.timings: timing_parts.append(f"db_write={perf.timings['db_write']:.3f}s") + if perf.llm_calls > 0: + timing_parts.append(f"avg_obs={perf.total_obs_in_context / perf.llm_calls:.1f}") + timing_parts.append(f"avg_prompt_tokens=~{perf.total_prompt_chars / perf.llm_calls / 4:.0f}") + if timing_parts: perf.log(f"[4] Timing breakdown: {', '.join(timing_parts)}") @@ -411,211 +433,219 @@ async def _trigger_mental_model_refreshes( return refreshed_count -async def _process_memory( +async def _process_memory_batch( conn: "Connection", memory_engine: "MemoryEngine", bank_id: str, - memory: dict[str, Any], + memories: list[dict[str, Any]], request_context: "RequestContext", perf: ConsolidationPerfLog | None = None, config: Any = None, -) -> dict[str, Any]: +) -> list[dict[str, Any]]: """ - Process a single memory for consolidation using a SINGLE LLM call. + Process a batch of memories in a single LLM call. - This function: - 1. Finds related observations (can be empty) - 2. Uses ONE LLM call to extract durable knowledge AND decide on actions - 3. Executes array of actions (can be multiple creates/updates) + Steps: + 1. Parallel recalls — one per fact (read-only; safe to parallelise) + 2. Union of retrieved observations across the batch (deduped by id) + 3. Single LLM call with all N facts + unioned observations + 4. Sequential action execution (writes remain serial for consistency) + 5. Returns one result dict per memory, in the same order as `memories` - The LLM handles all cases: - - No related observations: returns create action(s) with extracted durable knowledge - - Related observations exist: returns update/create actions based on tag routing - - Purely ephemeral fact: returns empty array (skip) - - Returns: - Dict with action summary: created/updated/merged counts + Per-fact security: action execution validates each learning_id against the + observations that were recalled specifically for that fact, so cross-tag + updates cannot occur. """ - from ...tracing import get_tracer, is_tracing_enabled + import asyncio - fact_text = memory["text"] - memory_id = memory["id"] - fact_tags = memory.get("tags") or [] - - # Create parent span for this memory's consolidation - tracer = get_tracer() - if is_tracing_enabled(): - consolidation_span = tracer.start_span("hindsight.consolidation") - consolidation_span.set_attribute("hindsight.memory_id", str(memory_id)) - consolidation_span.set_attribute("hindsight.bank_id", bank_id) - else: - consolidation_span = None - - try: - # Find related observations using the full recall system - # SECURITY: Pass tags to ensure observations don't leak across security boundaries - t0 = time.time() - recall_result = await _find_related_observations( + # 1. Parallel recalls — one per fact + t0 = time.time() + recall_tasks = [ + _find_related_observations( memory_engine=memory_engine, bank_id=bank_id, - query=fact_text, + query=m["text"], request_context=request_context, - tags=fact_tags, # Pass source memory's tags for security + tags=m.get("tags") or [], ) - if perf: - perf.record_timing("recall", time.time() - t0) + for m in memories + ] + per_fact_recalls = await asyncio.gather(*recall_tasks) + if perf: + perf.record_timing("recall", time.time() - t0) - # Single LLM call handles ALL cases (with or without existing observations) - # Note: Tags are NOT passed to LLM - they are handled algorithmically - t0 = time.time() - actions = await _consolidate_with_llm( + # 2. Build per-fact observation sets (keyed by memory ID string) for secure action validation + per_fact_obs_ids: dict[str, set[str]] = { + str(memories[i]["id"]): {str(obs.id) for obs in r.results} for i, r in enumerate(per_fact_recalls) + } + + # Union all observations (deduped by id) + seen_ids: set[str] = set() + union_observations: list["MemoryFact"] = [] + union_source_facts: dict[str, "MemoryFact"] = {} + for recall_result in per_fact_recalls: + for obs in recall_result.results: + obs_id = str(obs.id) + if obs_id not in seen_ids: + seen_ids.add(obs_id) + union_observations.append(obs) + if recall_result.source_facts: + union_source_facts.update(recall_result.source_facts) + + # 3. Single LLM call + t0 = time.time() + llm_result = await _consolidate_batch_with_llm( + memory_engine=memory_engine, + memories=memories, + union_observations=union_observations, + union_source_facts=union_source_facts, + config=config, + ) + if perf: + perf.record_timing("llm", time.time() - t0) + perf.record_llm_call(llm_result.obs_count, llm_result.prompt_chars) + + # 4. Sequential execution of creates / updates / deletes + # Track which memory indices participated so we can build per-memory results for stats + per_memory_created: set[str] = set() + per_memory_updated: set[str] = set() + + # All memories in the batch share the same tag set (enforced by batching) + fact_tags = memories[0].get("tags") or [] if memories else [] + + mem_by_id = {str(m["id"]): m for m in memories} + + for create in llm_result.creates: + source_mems = [mem_by_id[fid] for fid in create.source_fact_ids if fid in mem_by_id] + if not source_mems: + continue + await _execute_create_action( + conn=conn, memory_engine=memory_engine, - fact_text=fact_text, - recall_result=recall_result, - config=config, + bank_id=bank_id, + source_memory_ids=[m["id"] for m in source_mems], + text=create.text, + source_fact_tags=fact_tags, + event_date=_min_date(m.get("event_date") for m in source_mems), + occurred_start=_min_date(m.get("occurred_start") for m in source_mems), + occurred_end=_max_date(m.get("occurred_end") for m in source_mems), + mentioned_at=_max_date(m.get("mentioned_at") for m in source_mems), + perf=perf, ) - if perf: - perf.record_timing("llm", time.time() - t0) + for m in source_mems: + per_memory_created.add(str(m["id"])) - if not actions: - # LLM returned empty array - fact is purely ephemeral, skip - return {"action": "skipped", "reason": "no_durable_knowledge"} + for update in llm_result.updates: + source_mems = [mem_by_id[fid] for fid in update.source_fact_ids if fid in mem_by_id] + if not source_mems: + continue + # Security: the observation must have been recalled for at least one of the source facts + if not any(update.observation_id in per_fact_obs_ids.get(str(m["id"]), set()) for m in source_mems): + logger.debug( + f"Batch consolidation: rejected update — observation {update.observation_id} " + f"not in any source fact's recall" + ) + continue + await _execute_update_action( + conn=conn, + memory_engine=memory_engine, + bank_id=bank_id, + source_memory_ids=[m["id"] for m in source_mems], + observation_id=update.observation_id, + new_text=update.text, + observations=union_observations, + source_fact_tags=fact_tags, + source_occurred_start=_min_date(m.get("occurred_start") for m in source_mems), + source_occurred_end=_max_date(m.get("occurred_end") for m in source_mems), + source_mentioned_at=_max_date(m.get("mentioned_at") for m in source_mems), + perf=perf, + ) + for m in source_mems: + per_memory_updated.add(str(m["id"])) - # Execute all actions and collect results - results = [] - for action in actions: - action_type = action.get("action") - if action_type == "update": - result = await _execute_update_action( - conn=conn, - memory_engine=memory_engine, - bank_id=bank_id, - memory_id=memory_id, - action=action, - observations=recall_result.results, - source_fact_tags=fact_tags, # Pass source fact's tags for security - source_occurred_start=memory.get("occurred_start"), - source_occurred_end=memory.get("occurred_end"), - source_mentioned_at=memory.get("mentioned_at"), - perf=perf, - ) - results.append(result) - elif action_type == "create": - result = await _execute_create_action( - conn=conn, - memory_engine=memory_engine, - bank_id=bank_id, - memory_id=memory_id, - action=action, - source_fact_tags=fact_tags, # Pass source fact's tags for security - event_date=memory.get("event_date"), - occurred_start=memory.get("occurred_start"), - occurred_end=memory.get("occurred_end"), - mentioned_at=memory.get("mentioned_at"), - perf=perf, - ) - results.append(result) + for delete in llm_result.deletes: + # Security: the observation must be present in the unioned recall + if not any(str(obs.id) == delete.observation_id for obs in union_observations): + logger.debug( + f"Batch consolidation: rejected delete — observation {delete.observation_id} not in unioned recall" + ) + continue + await _execute_delete_action(conn=conn, bank_id=bank_id, observation_id=delete.observation_id) - if not results: - # No valid actions executed - return {"action": "skipped", "reason": "no_valid_actions"} + # Build per-memory result dicts for the stats tracker in the outer loop + results: list[dict[str, Any]] = [] + for m in memories: + mid = str(m["id"]) + created = mid in per_memory_created + updated = mid in per_memory_updated + if created and updated: + results.append({"action": "multiple", "created": 1, "updated": 1, "merged": 0, "total_actions": 2}) + elif created: + results.append({"action": "created"}) + elif updated: + results.append({"action": "updated"}) + else: + results.append({"action": "skipped", "reason": "no_durable_knowledge"}) - # Summarize results - created = sum(1 for r in results if r.get("action") == "created") - updated = sum(1 for r in results if r.get("action") == "updated") - merged = sum(1 for r in results if r.get("action") == "merged") + return results - if len(results) == 1: - return results[0] - return { - "action": "multiple", - "created": created, - "updated": updated, - "merged": merged, - "total_actions": len(results), - } - finally: - if consolidation_span: - consolidation_span.end() +def _min_date(dates: "Any") -> "datetime | None": + """Return the minimum non-None datetime from an iterable.""" + return min((d for d in dates if d is not None), default=None) + + +def _max_date(dates: "Any") -> "datetime | None": + """Return the maximum non-None datetime from an iterable.""" + return max((d for d in dates if d is not None), default=None) async def _execute_update_action( conn: "Connection", memory_engine: "MemoryEngine", bank_id: str, - memory_id: uuid.UUID, - action: dict[str, Any], + source_memory_ids: list[uuid.UUID], + observation_id: str, + new_text: str, observations: list["MemoryFact"], source_fact_tags: list[str] | None = None, source_occurred_start: datetime | None = None, source_occurred_end: datetime | None = None, source_mentioned_at: datetime | None = None, perf: ConsolidationPerfLog | None = None, -) -> dict[str, Any]: +) -> None: """ - Execute an update action on an existing observation. + Update an existing observation. - Updates the observation text, adds to history, increments proof_count, - and updates temporal fields: - - occurred_start: uses LEAST to keep the earliest start time - - occurred_end: uses GREATEST to keep the most recent end time - - mentioned_at: uses GREATEST to keep the most recent mention time - - SECURITY: Merges source fact's tags into the observation's existing tags. - This ensures all contributors can see the observation they contributed to. - For example, if Lisa's observation (tags=['user_lisa']) is updated with - Mike's fact (tags=['user_mike']), the observation will have both tags. + Extends source_memory_ids with all contributing memories, updates temporal fields + (LEAST for occurred_start, GREATEST for occurred_end / mentioned_at), and merges tags. """ - learning_id = action.get("learning_id") - new_text = action.get("text") - reason = action.get("reason", "Updated with new fact") - - if not learning_id or not new_text: - return {"action": "skipped", "reason": "missing_learning_id_or_text"} - - # Find the observation - model = next((m for m in observations if m.id == learning_id), None) + model = next((m for m in observations if str(m.id) == observation_id), None) if not model: - return {"action": "skipped", "reason": "learning_not_found"} + logger.debug(f"Update skipped: observation {observation_id} not found in recall results") + return - # Build history entry (history is fetched fresh from DB on update to avoid stale state) history = [ { "previous_text": model.text, "changed_at": datetime.now(timezone.utc).isoformat(), - "reason": reason, - "source_memory_id": str(memory_id), + "source_memory_ids": [str(mid) for mid in source_memory_ids], } ] - # Update source_memory_ids - source_ids = list(model.source_fact_ids or []) - source_ids.append(memory_id) + source_ids = list(model.source_fact_ids or []) + source_memory_ids - # SECURITY: Merge source fact's tags into existing observation tags - # This ensures all contributors can see the observation they contributed to + # SECURITY: Merge source fact's tags into existing observation tags so all contributors can see it existing_tags = set(model.tags or []) source_tags = set(source_fact_tags or []) - merged_tags = list(existing_tags | source_tags) # Union of both tag sets - if source_tags and source_tags != existing_tags: - logger.debug( - f"Security: Merging tags for observation {learning_id}: " - f"existing={list(existing_tags)}, source={list(source_tags)}, merged={merged_tags}" - ) + merged_tags = list(existing_tags | source_tags) - # Generate new embedding for updated text t0 = time.time() embeddings = await embedding_utils.generate_embeddings_batch(memory_engine.embeddings, [new_text]) embedding_str = str(embeddings[0]) if embeddings else None if perf: perf.record_timing("embedding", time.time() - t0) - # Update the observation - # - occurred_start: LEAST keeps the earliest start time across all source facts - # - occurred_end: GREATEST keeps the most recent end time across all source facts - # - mentioned_at: GREATEST keeps the most recent mention time - # - tags: merged from existing + source fact (for visibility) t0 = time.time() await conn.execute( f""" @@ -637,73 +667,65 @@ async def _execute_update_action( json.dumps(history), source_ids, len(source_ids), - uuid.UUID(learning_id), + uuid.UUID(observation_id), source_occurred_start, source_occurred_end, source_mentioned_at, merged_tags, ) - - # Create links from memory to observation - await _create_memory_links(conn, memory_id, uuid.UUID(learning_id)) if perf: perf.record_timing("db_write", time.time() - t0) - logger.debug(f"Updated observation {learning_id} with memory {memory_id}") - - return {"action": "updated", "observation_id": learning_id} + logger.debug(f"Updated observation {observation_id} from {len(source_memory_ids)} source memories") async def _execute_create_action( conn: "Connection", memory_engine: "MemoryEngine", bank_id: str, - memory_id: uuid.UUID, - action: dict[str, Any], + source_memory_ids: list[uuid.UUID], + text: str, source_fact_tags: list[str] | None = None, event_date: datetime | None = None, occurred_start: datetime | None = None, occurred_end: datetime | None = None, mentioned_at: datetime | None = None, perf: ConsolidationPerfLog | None = None, -) -> dict[str, Any]: +) -> None: """ - Execute a create action for a new observation. + Create a new observation from one or more source memories. - Creates a new observation with the specified text. - The text comes directly from the classify LLM - no second LLM call needed. - - Tags are determined algorithmically (not by LLM): - - Observations always inherit their source fact's tags - - This ensures visibility scope is maintained (security) + Tags are inherited from the source facts (determined algorithmically, not by LLM) + to maintain visibility scope. """ - text = action.get("text") - - # Tags are determined algorithmically - always use source fact's tags - # This ensures private memories create private observations - tags = source_fact_tags or [] - - if not text: - return {"action": "skipped", "reason": "missing_text"} - - # Use text directly from classify - skip the redundant LLM call - result = await _create_observation_directly( + await _create_observation_directly( conn=conn, memory_engine=memory_engine, bank_id=bank_id, - source_memory_id=memory_id, - observation_text=text, # Text already processed by classify LLM - tags=tags, + source_memory_ids=source_memory_ids, + observation_text=text, + tags=source_fact_tags or [], event_date=event_date, occurred_start=occurred_start, occurred_end=occurred_end, mentioned_at=mentioned_at, perf=perf, ) + logger.debug(f"Created observation from {len(source_memory_ids)} source memories") - logger.debug(f"Created observation {result.get('observation_id')} from memory {memory_id} (tags: {tags})") - return result +async def _execute_delete_action( + conn: "Connection", + bank_id: str, + observation_id: str, +) -> None: + """Delete a superseded or contradicted observation.""" + await conn.execute( + f"DELETE FROM {fq_table('memory_units')} WHERE id = $1 AND bank_id = $2 AND fact_type = 'observation'", + uuid.UUID(observation_id), + bank_id, + ) + logger.debug(f"Deleted observation {observation_id}") async def _create_memory_links( @@ -803,7 +825,6 @@ def _build_observations_for_llm( "id": obs.id, "text": obs.text, "proof_count": len(obs.source_fact_ids or []) or 1, - "tags": obs.tags or [], } if obs.occurred_start: obs_data["occurred_start"] = obs.occurred_start @@ -811,74 +832,91 @@ def _build_observations_for_llm( obs_data["occurred_end"] = obs.occurred_end if obs.mentioned_at: obs_data["mentioned_at"] = obs.mentioned_at - source_memories = [ - {"text": sf.text, "occurred_start": sf.occurred_start} - for sid in (obs.source_fact_ids or [])[:3] - if (sf := source_facts.get(sid)) is not None - ] + source_memories = [] + for sid in obs.source_fact_ids or []: + sf = source_facts.get(sid) + if sf is None: + continue + sf_data: dict[str, Any] = {"text": sf.text} + if sf.context: + sf_data["context"] = sf.context + if sf.occurred_start: + sf_data["occurred_start"] = sf.occurred_start + if sf.occurred_end: + sf_data["occurred_end"] = sf.occurred_end + if sf.mentioned_at: + sf_data["mentioned_at"] = sf.mentioned_at + source_memories.append(sf_data) if source_memories: obs_data["source_memories"] = source_memories obs_list.append(obs_data) return obs_list -async def _consolidate_with_llm( +async def _consolidate_batch_with_llm( memory_engine: "MemoryEngine", - fact_text: str, - recall_result: "RecallResult", + memories: list[dict[str, Any]], + union_observations: "list[MemoryFact]", + union_source_facts: "dict[str, MemoryFact]", config: Any = None, -) -> list[dict[str, Any]]: - """ - Single LLM call to extract durable knowledge and decide on consolidation actions. - - This handles ALL cases: - - No related observations: extracts durable knowledge, returns create action - - Related observations exist: compares and returns update/create actions - - Purely ephemeral fact: returns empty array - - Note: Tags are NOT handled by the LLM. They are determined algorithmically: - - CREATE: observation inherits source fact's tags - - UPDATE: observation merges source fact's tags with existing tags - - Returns: - List of actions, each being: - - {"action": "update", "learning_id": "uuid", "text": "...", "reason": "..."} - - {"action": "create", "text": "...", "reason": "..."} - - [] if fact is purely ephemeral (no durable knowledge) - """ - observations = recall_result.results - source_facts = recall_result.source_facts or {} - - if observations: - obs_list = _build_observations_for_llm(observations, source_facts) +) -> _BatchLLMResult: + """Single LLM call for a batch of facts against a pooled set of observations.""" + if union_observations: + obs_list = _build_observations_for_llm(union_observations, union_source_facts) observations_text = json.dumps(obs_list, indent=2) else: observations_text = "[]" + def _fact_line(m: dict[str, Any]) -> str: + parts = [f"[{m['id']}] {m['text']}"] + if m.get("occurred_start"): + parts.append(f"occurred_start={m['occurred_start']}") + if m.get("occurred_end"): + parts.append(f"occurred_end={m['occurred_end']}") + if m.get("mentioned_at"): + parts.append(f"mentioned_at={m['mentioned_at']}") + return " | ".join(parts) + + facts_lines = "\n".join(_fact_line(m) for m in memories) + observations_mission = config.observations_mission if config is not None else None - prompt_template = build_consolidation_prompt(observations_mission) + prompt_template = build_batch_consolidation_prompt(observations_mission) prompt = prompt_template.format( - fact_text=fact_text, + facts_text=facts_lines, observations_text=observations_text, ) - messages = [ - {"role": "user", "content": prompt}, - ] + max_attempts = 3 + last_exc: Exception | None = None + for attempt in range(1, max_attempts + 1): + try: + response: _ConsolidationBatchResponse = await memory_engine._consolidation_llm_config.call( + messages=[{"role": "user", "content": prompt}], + response_format=_ConsolidationBatchResponse, + scope="consolidation", + ) + return _BatchLLMResult( + creates=response.creates, + updates=response.updates, + deletes=response.deletes, + obs_count=len(union_observations), + prompt_chars=len(prompt), + ) + except Exception as exc: + last_exc = exc + logger.warning(f"[CONSOLIDATION] LLM batch call failed (attempt {attempt}/{max_attempts}): {exc}") - response: _ConsolidationResponse = await memory_engine._consolidation_llm_config.call( - messages=messages, - response_format=_ConsolidationResponse, - scope="consolidation", + logger.error( + f"[CONSOLIDATION] LLM batch call failed after {max_attempts} attempts, skipping batch. Last error: {last_exc}" ) - return [a.model_dump() for a in response.actions] + return _BatchLLMResult(obs_count=len(union_observations), prompt_chars=len(prompt)) async def _create_observation_directly( conn: "Connection", memory_engine: "MemoryEngine", bank_id: str, - source_memory_id: uuid.UUID, + source_memory_ids: list[uuid.UUID], observation_text: str, tags: list[str] | None = None, event_date: datetime | None = None, @@ -887,12 +925,7 @@ async def _create_observation_directly( mentioned_at: datetime | None = None, perf: ConsolidationPerfLog | None = None, ) -> dict[str, Any]: - """ - Create an observation directly with pre-processed text (no LLM call). - - Used when the classify LLM has already provided the learning text. - This avoids the redundant second LLM call. - """ + """Create an observation from one or more source memories with pre-processed text.""" # Generate embedding for the observation (convert to string for pgvector) t0 = time.time() embeddings = await embedding_utils.generate_embeddings_batch(memory_engine.embeddings, [observation_text]) @@ -942,7 +975,7 @@ async def _create_observation_directly( bank_id, observation_text, embedding_str, - [source_memory_id], + source_memory_ids, obs_tags, obs_event_date, obs_occurred_start, @@ -950,11 +983,9 @@ async def _create_observation_directly( obs_mentioned_at, ) - # Create links between memory and observation (includes entity links, memory_links) - await _create_memory_links(conn, source_memory_id, observation_id) if perf: perf.record_timing("db_write", time.time() - t0) - logger.debug(f"Created observation {observation_id} from memory {source_memory_id} (tags: {obs_tags})") + logger.debug(f"Created observation {observation_id} from {len(source_memory_ids)} memories (tags: {obs_tags})") return {"action": "created", "observation_id": str(row["id"]), "tags": obs_tags} diff --git a/hindsight-api/hindsight_api/engine/consolidation/prompts.py b/hindsight-api/hindsight_api/engine/consolidation/prompts.py index 13f1b873..58f890c2 100644 --- a/hindsight-api/hindsight_api/engine/consolidation/prompts.py +++ b/hindsight-api/hindsight_api/engine/consolidation/prompts.py @@ -1,21 +1,21 @@ """Prompts for the consolidation engine.""" -# Output format instructions -_OUTPUT_FORMAT = """ -Output a JSON object with an "actions" array: -{{"actions": [ - {{"action": "update", "learning_id": "uuid-from-observations", "text": "...", "reason": "..."}}, - {{"action": "create", "text": "...", "reason": "..."}} -]}} +# Default mission when no bank-specific mission is set +_DEFAULT_MISSION = "Track every detail: names, numbers, dates, places, and relationships. Prefer specifics over abstractions, never generalise." -Return {{"actions": []}} if the fact contains no durable knowledge. -Do NOT include "tags" in output — tags are handled automatically.""" +# Processing rules — always present regardless of mission +_PROCESSING_RULES = """Processing rules (always apply): +- REDUNDANT: same info worded differently → UPDATE the existing observation. +- CONTRADICTION/UPDATE: capture both states with temporal markers ("used to X, now Y"). +- RESOLVE REFERENCES: when a new fact provides a concrete value resolving a vague placeholder in an existing observation (e.g. "home country", "hometown", "birthplace", "native language", "her ex", "that city"), UPDATE the observation to embed the resolved value explicitly. Example: new fact says "grandma in Sweden" + existing observation says "moved from her home country" → update to "home country is Sweden". +- NEVER merge observations about different people or unrelated topics.""" -# Data section - holds the dynamic per-call data -_DATA_SECTION = """ -NEW FACT: {fact_text} +# Data section — format placeholders {facts_text} and {observations_text} are substituted at call time +_BATCH_DATA_SECTION = """ +NEW FACTS: +{facts_text} -EXISTING OBSERVATIONS (JSON array with source memories and dates): +EXISTING OBSERVATIONS (JSON array, pooled from recalls across all facts above): {observations_text} Each observation includes: @@ -25,34 +25,42 @@ Each observation includes: - occurred_start/occurred_end: temporal range of source facts - source_memories: array of supporting facts with their text and dates -Compare the new fact against existing observations: -- Same topic → UPDATE with learning_id -- New topic → CREATE new observation -- Purely ephemeral → return empty actions list""" +Compare the facts against existing observations: +- Same topic as an existing observation → UPDATE it (observation_id + source_fact_ids) +- New topic with durable knowledge → CREATE a new observation (source_fact_ids) +- Cross-reference facts within the batch: a later fact may resolve a vague reference in an earlier one +- Purely ephemeral facts → omit them (no create/update needed)""" -# Default rules used when no observations_mission is set -_DEFAULT_RULES = """Extract DURABLE KNOWLEDGE from facts — the stable truth implied by an event, not transient state. +# Output format — JSON braces escaped as {{ }} so .format() leaves them literal +_BATCH_OUTPUT_FORMAT = """ +Output a JSON object with three arrays. -Example: "User moved to Room 203" → observe "Room 203 exists", not "User is in Room 203". +Example (showing the required UUID format for all IDs): +{{"creates": [{{"text": "Alice lives in Berlin", "source_fact_ids": ["a1b2c3d4-e5f6-7890-abcd-ef1234567890", "b2c3d4e5-f6a7-8901-bcde-f12345678901"]}}], + "updates": [{{"text": "Alice works at Acme Corp as a senior engineer", "observation_id": "c3d4e5f6-a7b8-9012-cdef-123456789012", "source_fact_ids": ["d4e5f6a7-b8c9-0123-defa-234567890123"]}}], + "deletes": [{{"observation_id": "e5f6a7b8-c9d0-1234-efab-345678901234"}}]}} Rules: -- Keep specifics: names, numbers, locations. Never abstract into general principles. -- NEVER merge observations about different people or unrelated topics. -- REDUNDANT: same info worded differently → update existing. -- CONTRADICTION/UPDATE: capture both states with temporal markers ("used to X, now Y"). -- RESOLVE REFERENCES: When a new fact provides a concrete value that resolves a vague placeholder in an existing observation (e.g., a location that corresponds to "home country", "hometown", "birthplace", "native language", "her ex", "that city"), UPDATE the existing observation to embed the resolved value explicitly. Example: new fact mentions grandma in Sweden + existing observation says "moved from her home country" → update to state "home country is Sweden".""" +- "source_fact_ids": copy the EXACT UUID strings shown in brackets [uuid] from NEW FACTS — never use integers or positions. +- "observation_id": copy the EXACT "id" UUID string from EXISTING OBSERVATIONS. +- One create/update may reference multiple facts when they jointly support the observation. +- "deletes": only when an observation is directly superseded or contradicted by new facts. +- Do NOT include "tags" — handled automatically. +- Return {{"creates": [], "updates": [], "deletes": []}} if nothing durable is found.""" -def build_consolidation_prompt(observations_mission: str | None = None) -> str: +def build_batch_consolidation_prompt(observations_mission: str | None = None) -> str: """ - Build the consolidation prompt. + Build the consolidation prompt for batch mode (multiple facts per LLM call). - If observations_mission is provided, it replaces the default durable-knowledge rules - with bank-specific instructions for what to synthesise. Otherwise the default rules apply. + The mission defines *what* to track (customisable per bank). + Processing rules and output format are always present regardless of mission. """ - rules_section = f"## MISSION\n{observations_mission}" if observations_mission else _DEFAULT_RULES + mission = observations_mission or _DEFAULT_MISSION return ( "You are a memory consolidation system. Synthesize facts into observations " - "and merge with existing observations when appropriate.\n\n" + rules_section + _DATA_SECTION + _OUTPUT_FORMAT + "and merge with existing observations when appropriate.\n\n" + f"## MISSION\n{mission}\n\n" + f"{_PROCESSING_RULES}" + _BATCH_DATA_SECTION + _BATCH_OUTPUT_FORMAT ) diff --git a/hindsight-api/hindsight_api/main.py b/hindsight-api/hindsight_api/main.py index 2b9d5829..c950fda9 100644 --- a/hindsight-api/hindsight_api/main.py +++ b/hindsight-api/hindsight_api/main.py @@ -273,6 +273,7 @@ def main(): file_delete_after_retain=config.file_delete_after_retain, enable_observations=config.enable_observations, consolidation_batch_size=config.consolidation_batch_size, + consolidation_llm_batch_size=config.consolidation_llm_batch_size, consolidation_max_tokens=config.consolidation_max_tokens, observations_mission=config.observations_mission, skip_llm_verification=config.skip_llm_verification, diff --git a/hindsight-api/tests/test_consolidation.py b/hindsight-api/tests/test_consolidation.py index 1379453e..1100d15e 100644 --- a/hindsight-api/tests/test_consolidation.py +++ b/hindsight-api/tests/test_consolidation.py @@ -1990,35 +1990,35 @@ class TestMentalModelRefreshAfterConsolidation: def test_consolidation_prompt_default(): - """Test that the default consolidation prompt contains the built-in durable-knowledge rules.""" - from hindsight_api.engine.consolidation.prompts import build_consolidation_prompt + """Test that the default consolidation prompt contains the built-in mission and processing rules.""" + from hindsight_api.engine.consolidation.prompts import build_batch_consolidation_prompt - prompt = build_consolidation_prompt() - assert "DURABLE KNOWLEDGE" in prompt + prompt = build_batch_consolidation_prompt() assert "temporal markers" in prompt - assert "{fact_text}" in prompt + assert "RESOLVE REFERENCES" in prompt + assert "{facts_text}" in prompt assert "{observations_text}" in prompt def test_consolidation_prompt_observations_mission(): - """Test that observations_mission replaces the default rules.""" - from hindsight_api.engine.consolidation.prompts import build_consolidation_prompt + """Test that observations_mission replaces the default mission but keeps processing rules.""" + from hindsight_api.engine.consolidation.prompts import build_batch_consolidation_prompt spec = "Observations are weekly summaries of sprint outcomes and team dynamics." - prompt = build_consolidation_prompt(observations_mission=spec) + prompt = build_batch_consolidation_prompt(observations_mission=spec) # Spec is injected assert spec in prompt - # Default rules are NOT present - assert "EXTRACT DURABLE KNOWLEDGE" not in prompt - # Output format and data placeholders remain - assert "actions" in prompt - assert "{fact_text}" in prompt + # Processing rules and output format always remain + assert "RESOLVE REFERENCES" in prompt + assert "creates" in prompt + assert "updates" in prompt + assert "{facts_text}" in prompt assert "{observations_text}" in prompt # Renders cleanly - rendered = prompt.format(fact_text="Alice fixed a bug.", observations_text="[]") - assert "{fact_text}" not in rendered + rendered = prompt.format(facts_text="Alice fixed a bug.", observations_text="[]") + assert "{facts_text}" not in rendered assert spec in rendered diff --git a/hindsight-dev/benchmarks/common/benchmark_runner.py b/hindsight-dev/benchmarks/common/benchmark_runner.py index 1d98bc25..995bc420 100644 --- a/hindsight-dev/benchmarks/common/benchmark_runner.py +++ b/hindsight-dev/benchmarks/common/benchmark_runner.py @@ -455,48 +455,23 @@ class BenchmarkRunner: Get the count of memories pending consolidation. Returns: - Number of memories not yet consolidated into mental models + Number of memories not yet processed by the consolidation job """ pool = await self.memory._get_pool() from hindsight_api.engine.memory_engine import fq_table async with pool.acquire() as conn: - # Check when consolidation last ran - last_consolidated_row = await conn.fetchrow( + result = await conn.fetchrow( f""" - SELECT MAX(created_at) as last_consolidated_at + SELECT COUNT(*) as count FROM {fq_table("memory_units")} - WHERE bank_id = $1 AND fact_type = 'mental_model' + WHERE bank_id = $1 AND consolidated_at IS NULL AND fact_type IN ('experience', 'world') """, bank_id, ) - last_consolidated_at = last_consolidated_row["last_consolidated_at"] if last_consolidated_row else None - - if last_consolidated_at: - # Count memories created after last consolidation - result = await conn.fetchrow( - f""" - SELECT COUNT(*) as count - FROM {fq_table("memory_units")} - WHERE bank_id = $1 AND fact_type IN ('experience', 'world') - AND created_at > $2 - """, - bank_id, - last_consolidated_at, - ) - else: - # If never consolidated, count all experience/world memories - result = await conn.fetchrow( - f""" - SELECT COUNT(*) as count - FROM {fq_table("memory_units")} - WHERE bank_id = $1 AND fact_type IN ('experience', 'world') - """, - bank_id, - ) return result["count"] if result else 0 - async def _wait_for_consolidation(self, bank_id: str, poll_interval: float = 2.0, timeout: float = 300.0) -> None: + async def _wait_for_consolidation(self, bank_id: str, poll_interval: float = 2.0, timeout: float = 3000.0) -> None: """ Wait for consolidation to complete (pending_consolidation reaches 0). @@ -860,6 +835,7 @@ class BenchmarkRunner: question_semaphore: asyncio.Semaphore, eval_semaphore_size: int = 8, clear_this_agent: bool = True, + wait_consolidation: bool = False, ) -> Dict: """ Process a single item (ingest + evaluate). @@ -867,6 +843,7 @@ class BenchmarkRunner: Args: clear_this_agent: Whether to clear this agent's data before ingesting. Set to False to skip clearing (e.g., when agent_id is shared and already cleared) + wait_consolidation: If True, wait for consolidation to complete before evaluating QA. Returns: Result dict with metrics @@ -891,6 +868,12 @@ class BenchmarkRunner: else: num_sessions = -1 + # Wait for consolidation before evaluating if requested + if wait_consolidation: + step += 1 + console.print(f" [{step}] Waiting for consolidation...") + await self._wait_for_consolidation(agent_id) + # Evaluate QA step += 1 qa_pairs = self.dataset.get_qa_pairs(item) @@ -934,6 +917,7 @@ class BenchmarkRunner: max_concurrent_items: int = 1, # Max concurrent items (conversations) to process in parallel output_path: Optional[Path] = None, # Path to save results incrementally merge_with_existing: bool = False, # Whether to merge with existing results + wait_consolidation: bool = False, # Wait for consolidation to complete before evaluating QA ) -> Dict[str, Any]: """ Run the full benchmark evaluation. @@ -1011,6 +995,7 @@ class BenchmarkRunner: max_concurrent_items, output_path, merge_with_existing, + wait_consolidation, ) async def _run_single_phase( @@ -1028,6 +1013,7 @@ class BenchmarkRunner: max_concurrent_items: int = 1, output_path: Optional[Path] = None, merge_with_existing: bool = False, + wait_consolidation: bool = False, ) -> Dict[str, Any]: """Original single-phase approach: process each item independently.""" # Create semaphore for question processing @@ -1049,6 +1035,7 @@ class BenchmarkRunner: max_concurrent_items, output_path, merge_with_existing, + wait_consolidation, ) else: # Sequential item processing (original behavior) @@ -1065,6 +1052,7 @@ class BenchmarkRunner: filln, output_path, merge_with_existing, + wait_consolidation, ) # Calculate overall metrics @@ -1100,6 +1088,7 @@ class BenchmarkRunner: filln: bool, output_path: Optional[Path] = None, merge_with_existing: bool = False, + wait_consolidation: bool = False, ) -> List[Dict]: """Process items sequentially (original behavior).""" all_results = [] @@ -1147,6 +1136,7 @@ class BenchmarkRunner: question_semaphore, eval_semaphore_size, clear_this_agent, + wait_consolidation, ) # Replace existing result or append new one @@ -1178,6 +1168,7 @@ class BenchmarkRunner: max_concurrent_items: int, output_path: Optional[Path] = None, merge_with_existing: bool = False, + wait_consolidation: bool = False, ) -> List[Dict]: """Process items in parallel (requires unique agent IDs per item).""" # Load existing results if merge_with_existing is True @@ -1222,6 +1213,7 @@ class BenchmarkRunner: question_semaphore, eval_semaphore_size, clear_this_agent=True, # Always clear for parallel processing + wait_consolidation=wait_consolidation, ) return result diff --git a/hindsight-dev/benchmarks/locomo/locomo_benchmark.py b/hindsight-dev/benchmarks/locomo/locomo_benchmark.py index 2d2bea73..d5391c65 100644 --- a/hindsight-dev/benchmarks/locomo/locomo_benchmark.py +++ b/hindsight-dev/benchmarks/locomo/locomo_benchmark.py @@ -295,6 +295,7 @@ async def run_benchmark( only_failed: bool = False, only_invalid: bool = False, question_index: int = None, + wait_consolidation: bool = False, ): """ Run the LoComo benchmark. @@ -461,6 +462,7 @@ async def run_benchmark( max_concurrent_items=concurrent_items, output_path=output_path, # Save results incrementally merge_with_existing=merge_with_existing, + wait_consolidation=wait_consolidation, ) # Display results (final save already happened incrementally) @@ -591,6 +593,11 @@ if __name__ == "__main__": default=None, help="Run only the question at this 0-based index within each conversation (e.g., 11)", ) + parser.add_argument( + "--wait-consolidation", + action="store_true", + help="Wait for consolidation to complete after ingestion (or immediately when using --skip-ingestion) before evaluating QA.", + ) args = parser.parse_args() @@ -610,5 +617,6 @@ if __name__ == "__main__": only_failed=args.only_failed, only_invalid=args.only_invalid, question_index=args.question_index, + wait_consolidation=args.wait_consolidation, ) ) diff --git a/hindsight-docs/docs/developer/configuration.md b/hindsight-docs/docs/developer/configuration.md index 4208ba72..24d3d17d 100644 --- a/hindsight-docs/docs/developer/configuration.md +++ b/hindsight-docs/docs/developer/configuration.md @@ -733,6 +733,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_CONSOLIDATION_LLM_BATCH_SIZE` | Number of facts sent to the LLM in a single consolidation call. Higher values reduce LLM calls and improve throughput at the cost of larger prompts. Set to `1` to disable batching. | `8` | | `HINDSIGHT_API_OBSERVATIONS_MISSION` | What this bank should synthesise into durable observations. Replaces the built-in consolidation rules — leave unset to use the server default. | - | #### Customizing observations: when to use what