diff --git a/hindsight-api-slim/hindsight_api/engine/reflect/agent.py b/hindsight-api-slim/hindsight_api/engine/reflect/agent.py index a9594710e2..da12047e4a 100644 --- a/hindsight-api-slim/hindsight_api/engine/reflect/agent.py +++ b/hindsight-api-slim/hindsight_api/engine/reflect/agent.py @@ -17,10 +17,15 @@ from ..llm_interface import LLM_TOOL_CHOICE_AUTO, LLMToolChoice from .models import DirectiveInfo, LLMCall, ReflectAgentResult, StructuredOutputResult, TokenUsageSummary, ToolCall from .prompts import ( + _SPLIT_SYNTHESIS_WARN_CHUNKS, + CLAIMS_SYSTEM_PROMPT, _extract_directive_rules, + build_chunk_claims_prompt, build_final_prompt, build_final_system_prompt, + build_reduce_prompt, build_system_prompt_for_tools, + split_context_history, ) from .tokenization import count_cl100k_tokens from .tools_schema import get_reflect_tools @@ -650,6 +655,106 @@ def _log_completion(answer: str, iterations: int, forced: bool = False): f"total={elapsed_ms}ms" ) + async def _tracked_llm_call(prompt: str, trace_scope: str, system_prompt: str, completion_cap: int | None) -> str: + """One tool-less LLM call with usage/trace accounting folded in.""" + nonlocal total_input_tokens, total_output_tokens, total_cached_tokens, total_thoughts_tokens + llm_start = time.time() + response, usage = await llm_config.call( + messages=[ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": prompt}, + ], + scope="reflect", + max_completion_tokens=completion_cap, + return_usage=True, + ) + llm_duration = int((time.time() - llm_start) * 1000) + total_input_tokens += usage.input_tokens + total_output_tokens += usage.output_tokens + total_cached_tokens += getattr(usage, "cached_tokens", 0) or 0 + total_thoughts_tokens += getattr(usage, "thoughts_tokens", 0) or 0 + llm_trace.append( + { + "scope": trace_scope, + "duration_ms": llm_duration, + "input_tokens": usage.input_tokens, + "output_tokens": usage.output_tokens, + } + ) + return response.strip() + + async def _forced_final_synthesis(iterations_completed: int) -> ReflectAgentResult: + """Answer without tools from the accumulated tool results. + + When the accumulated results fit the prompt budget this is one LLM call, + exactly as before. When they exceed it, they are SPLIT — not truncated: + each budget-sized chunk is compressed in parallel into dated, cited + claims, and one reduce call synthesizes the answer from every chunk's + claims. The old behavior dropped any over-budget block whole (plus all + older ones), which produced confident "no information" answers carrying + hundreds of citations the synthesis model never saw (#3122). + """ + nonlocal total_input_tokens, total_output_tokens, total_cached_tokens, total_thoughts_tokens + final_system = build_final_system_prompt(bank_profile.get("mission"), llm_output_language, directives) + chunks = split_context_history(context_history, max_context_tokens) + # Every call below uses the transport-level cap, never the caller's + # max_tokens: that is a visible-length target carried as a prompt + # directive (#3365), and capping the transport with it would truncate + # thinking models mid-word — or, on the map calls, starve the evidence + # extraction. + if len(chunks) <= 1: + prompt = build_final_prompt( + query, + context_history, + bank_profile, + context, + max_context_tokens=max_context_tokens, + max_tokens=max_tokens, + ) + answer = await _tracked_llm_call(prompt, "final", final_system, synthesis_max_completion_tokens) + else: + log = logger.warning if len(chunks) > _SPLIT_SYNTHESIS_WARN_CHUNKS else logger.info + log( + f"[REFLECT {reflect_id}] Retrieved data exceeds the context budget; " + f"split synthesis over {len(chunks)} chunks." + ) + # Map: each chunk in parallel. + claim_sections = await asyncio.gather( + *( + _tracked_llm_call( + build_chunk_claims_prompt(query, chunk), + f"final_map_{i}", + CLAIMS_SYSTEM_PROMPT, + synthesis_max_completion_tokens, + ) + for i, chunk in enumerate(chunks, 1) + ) + ) + # Reduce: one synthesis call over every chunk's claims. + prompt = build_reduce_prompt(query, list(claim_sections), bank_profile, context, max_tokens=max_tokens) + answer = await _tracked_llm_call(prompt, "final", final_system, synthesis_max_completion_tokens) + + structured_output = None + if response_schema and answer: + struct = await _generate_structured_output(answer, response_schema, llm_config, reflect_id, max_tokens) + structured_output = struct.structured_output + total_input_tokens += struct.input_tokens + total_output_tokens += struct.output_tokens + total_cached_tokens += struct.cached_tokens + total_thoughts_tokens += struct.thoughts_tokens + + _log_completion(answer, iterations_completed, forced=True) + return ReflectAgentResult( + text=answer, + structured_output=structured_output, + iterations=iterations_completed, + tools_called=total_tools_called, + tool_trace=tool_trace, + llm_trace=_get_llm_trace(), + usage=_get_usage(), + directives_applied=directives_applied, + ) + consecutive_errors = 0 # When a forced ``search_mental_models`` returns fresh, usable models on a # low/mid-budget call, we stop forcing the lower retrieval layers from this @@ -668,65 +773,7 @@ def _log_completion(answer: str, iterations: int, forced: bool = False): if is_last: # Force text response on last iteration - no tools - prompt = build_final_prompt( - query, - context_history, - bank_profile, - context, - max_context_tokens=max_context_tokens, - max_tokens=max_tokens, - ) - llm_start = time.time() - response, usage = await llm_config.call( - messages=[ - { - "role": "system", - "content": build_final_system_prompt( - bank_profile.get("mission"), llm_output_language, directives - ), - }, - {"role": "user", "content": prompt}, - ], - scope="reflect", - max_completion_tokens=synthesis_max_completion_tokens, - return_usage=True, - ) - llm_duration = int((time.time() - llm_start) * 1000) - total_input_tokens += usage.input_tokens - total_output_tokens += usage.output_tokens - total_cached_tokens += getattr(usage, "cached_tokens", 0) or 0 - total_thoughts_tokens += getattr(usage, "thoughts_tokens", 0) or 0 - llm_trace.append( - { - "scope": "final", - "duration_ms": llm_duration, - "input_tokens": usage.input_tokens, - "output_tokens": usage.output_tokens, - } - ) - answer = response.strip() - - # Generate structured output if schema provided - structured_output = None - if response_schema and answer: - struct = await _generate_structured_output(answer, response_schema, llm_config, reflect_id, max_tokens) - structured_output = struct.structured_output - total_input_tokens += struct.input_tokens - total_output_tokens += struct.output_tokens - total_cached_tokens += struct.cached_tokens - total_thoughts_tokens += struct.thoughts_tokens - - _log_completion(answer, iteration + 1, forced=True) - return ReflectAgentResult( - text=answer, - structured_output=structured_output, - iterations=iteration + 1, - tools_called=total_tools_called, - tool_trace=tool_trace, - llm_trace=_get_llm_trace(), - usage=_get_usage(), - directives_applied=directives_applied, - ) + return await _forced_final_synthesis(iteration + 1) # Proactive context-window guard: if accumulated messages would exceed the # configured token budget, bail out early and synthesize from what we have. @@ -738,64 +785,7 @@ def _log_completion(answer: str, iterations: int, forced: bool = False): f"[REFLECT {reflect_id}] Context budget exceeded on iteration {iteration + 1}: " f"~{estimated_tokens} tokens >= {max_context_tokens} limit. Forcing final synthesis." ) - prompt = build_final_prompt( - query, - context_history, - bank_profile, - context, - max_context_tokens=max_context_tokens, - max_tokens=max_tokens, - ) - llm_start = time.time() - response, usage = await llm_config.call( - messages=[ - { - "role": "system", - "content": build_final_system_prompt( - bank_profile.get("mission"), llm_output_language, directives - ), - }, - {"role": "user", "content": prompt}, - ], - scope="reflect", - max_completion_tokens=synthesis_max_completion_tokens, - return_usage=True, - ) - llm_duration = int((time.time() - llm_start) * 1000) - total_input_tokens += usage.input_tokens - total_output_tokens += usage.output_tokens - total_cached_tokens += getattr(usage, "cached_tokens", 0) or 0 - total_thoughts_tokens += getattr(usage, "thoughts_tokens", 0) or 0 - llm_trace.append( - { - "scope": "final", - "duration_ms": llm_duration, - "input_tokens": usage.input_tokens, - "output_tokens": usage.output_tokens, - } - ) - answer = response.strip() - - structured_output = None - if response_schema and answer: - struct = await _generate_structured_output(answer, response_schema, llm_config, reflect_id, max_tokens) - structured_output = struct.structured_output - total_input_tokens += struct.input_tokens - total_output_tokens += struct.output_tokens - total_cached_tokens += struct.cached_tokens - total_thoughts_tokens += struct.thoughts_tokens - - _log_completion(answer, iteration + 1, forced=True) - return ReflectAgentResult( - text=answer, - structured_output=structured_output, - iterations=iteration + 1, - tools_called=total_tools_called, - tool_trace=tool_trace, - llm_trace=_get_llm_trace(), - usage=_get_usage(), - directives_applied=directives_applied, - ) + return await _forced_final_synthesis(iteration + 1) # Call LLM with tools llm_start = time.time() @@ -886,65 +876,7 @@ def _log_completion(answer: str, iterations: int, forced: bool = False): # For other errors: retry if no evidence yet (but cap consecutive errors to avoid long hangs) elif not has_gathered_evidence and iteration < max_iterations - 1 and consecutive_errors < 2: continue - prompt = build_final_prompt( - query, - context_history, - bank_profile, - context, - max_context_tokens=max_context_tokens, - max_tokens=max_tokens, - ) - llm_start = time.time() - response, usage = await llm_config.call( - messages=[ - { - "role": "system", - "content": build_final_system_prompt( - bank_profile.get("mission"), llm_output_language, directives - ), - }, - {"role": "user", "content": prompt}, - ], - scope="reflect", - max_completion_tokens=synthesis_max_completion_tokens, - return_usage=True, - ) - llm_duration = int((time.time() - llm_start) * 1000) - total_input_tokens += usage.input_tokens - total_output_tokens += usage.output_tokens - total_cached_tokens += getattr(usage, "cached_tokens", 0) or 0 - total_thoughts_tokens += getattr(usage, "thoughts_tokens", 0) or 0 - llm_trace.append( - { - "scope": "final", - "duration_ms": llm_duration, - "input_tokens": usage.input_tokens, - "output_tokens": usage.output_tokens, - } - ) - answer = response.strip() - - # Generate structured output if schema provided - structured_output = None - if response_schema and answer: - struct = await _generate_structured_output(answer, response_schema, llm_config, reflect_id, max_tokens) - structured_output = struct.structured_output - total_input_tokens += struct.input_tokens - total_output_tokens += struct.output_tokens - total_cached_tokens += struct.cached_tokens - total_thoughts_tokens += struct.thoughts_tokens - - _log_completion(answer, iteration + 1, forced=True) - return ReflectAgentResult( - text=answer, - structured_output=structured_output, - iterations=iteration + 1, - tools_called=total_tools_called, - tool_trace=tool_trace, - llm_trace=_get_llm_trace(), - usage=_get_usage(), - directives_applied=directives_applied, - ) + return await _forced_final_synthesis(iteration + 1) # No tool calls this turn. if not result.tool_calls: @@ -969,65 +901,7 @@ def _log_completion(answer: str, iterations: int, forced: bool = False): ) # Model tool-called earlier and is now stopping: fall through to a clean # forced final synthesis (tools disabled, prose expected). - prompt = build_final_prompt( - query, - context_history, - bank_profile, - context, - max_context_tokens=max_context_tokens, - max_tokens=max_tokens, - ) - llm_start = time.time() - response, usage = await llm_config.call( - messages=[ - { - "role": "system", - "content": build_final_system_prompt( - bank_profile.get("mission"), llm_output_language, directives - ), - }, - {"role": "user", "content": prompt}, - ], - scope="reflect", - max_completion_tokens=synthesis_max_completion_tokens, - return_usage=True, - ) - llm_duration = int((time.time() - llm_start) * 1000) - total_input_tokens += usage.input_tokens - total_output_tokens += usage.output_tokens - total_cached_tokens += getattr(usage, "cached_tokens", 0) or 0 - total_thoughts_tokens += getattr(usage, "thoughts_tokens", 0) or 0 - llm_trace.append( - { - "scope": "final", - "duration_ms": llm_duration, - "input_tokens": usage.input_tokens, - "output_tokens": usage.output_tokens, - } - ) - answer = response.strip() - - # Generate structured output if schema provided - structured_output = None - if response_schema and answer: - struct = await _generate_structured_output(answer, response_schema, llm_config, reflect_id, max_tokens) - structured_output = struct.structured_output - total_input_tokens += struct.input_tokens - total_output_tokens += struct.output_tokens - total_cached_tokens += struct.cached_tokens - total_thoughts_tokens += struct.thoughts_tokens - - _log_completion(answer, iteration + 1, forced=True) - return ReflectAgentResult( - text=answer, - structured_output=structured_output, - iterations=iteration + 1, - tools_called=total_tools_called, - tool_trace=tool_trace, - llm_trace=_get_llm_trace(), - usage=_get_usage(), - directives_applied=directives_applied, - ) + return await _forced_final_synthesis(iteration + 1) # The model produced at least one tool call reflect could parse: it can # drive the loop, so a later text-only turn is a legitimate stop, not a diff --git a/hindsight-api-slim/hindsight_api/engine/reflect/prompts.py b/hindsight-api-slim/hindsight_api/engine/reflect/prompts.py index 12afb1becc..dba0754284 100644 --- a/hindsight-api-slim/hindsight_api/engine/reflect/prompts.py +++ b/hindsight-api-slim/hindsight_api/engine/reflect/prompts.py @@ -443,35 +443,153 @@ def build_system_prompt_for_tools( return "\n".join(parts) -def build_final_prompt( - query: str, - context_history: list[dict], - bank_profile: dict, - additional_context: str | None = None, - max_context_tokens: int = 100_000, - max_tokens: int | None = None, -) -> str: - """Build the final prompt when forcing a text response (no tools). +#: Result-list keys a tool output can carry; an over-budget block is split on +#: these entry boundaries so no retrieved evidence is dropped. +_SPLITTABLE_RESULT_KEYS = ("observations", "memories", "results") + +#: Above this many synthesis chunks the retrieval volume is pathological +#: (each chunk is ~0.8 * max_context_tokens); we still process everything, +#: but loudly, so the real cause (an unbounded tool result) gets looked at. +_SPLIT_SYNTHESIS_WARN_CHUNKS = 4 + +#: Floor for the per-chunk budget during splitting. A tiny configured +#: ``max_context_tokens`` (tests use 1) would otherwise shred the history into +#: one chunk per result entry — an LLM call per fact. A ~1k-token prompt is +#: safe for any real model, so the floor caps fan-out without dropping data. +_MIN_SPLIT_CHUNK_TOKENS = 1024 + +_FINAL_INSTRUCTIONS = ( + "Provide a thoughtful answer by synthesizing and reasoning from the retrieved data above. " + "You can make reasonable inferences from the memories, but don't completely fabricate information. " + "If the exact answer isn't stated, use what IS stated to give the best possible answer. " + "Only say 'I don't have information' if the retrieved data is truly unrelated to the question.\n\n" + "IMPORTANT: Output ONLY the final answer. Do NOT include meta-commentary like " + '"I\'ll search..." or "Let me analyze...". Do NOT explain your reasoning process. ' + "Just provide the direct synthesized answer." +) + + +def _render_history_block(entry: dict) -> str: + """Render one context-history entry as a fenced JSON block.""" + tool = entry["tool"] + output = entry["output"] + try: + output_str = json.dumps(output, indent=2, default=str, ensure_ascii=False) + except (TypeError, ValueError): + output_str = str(output) + return f"\n### From {tool}:\n```json\n{output_str}\n```" - ``max_tokens`` is the desired *visible* length of the answer (e.g. a mental - model page's ``max_tokens``). It is communicated here as a soft directive - rather than enforced by truncating the provider call: on thinking models the - provider budget is consumed by reasoning tokens, so a hard cap cuts the page - off mid-word (#3365). The hard length guarantee is the post-hoc rewrite in - the agent; this directive just steers the model toward the target so the - rewrite rarely has to fire. - """ - parts = [] - # Bank identity +def _cut_entry_to_budget(entry: dict, token_budget: int) -> dict: + """Token-bound one indivisible over-budget entry by cutting its serialized text. + + Only reachable when a single result entry (or a list-less output like a + document expand) alone exceeds the whole per-chunk budget — the one case + where "split, don't drop" cannot be honored without exceeding the model's + window. The cut text is wrapped back into an output dict so the entry + renders like any other block. + """ + output = entry["output"] + try: + output_str = json.dumps(output, indent=2, default=str, ensure_ascii=False) + except (TypeError, ValueError): + output_str = str(output) + tokens = count_cl100k_tokens(output_str) + while output_str and tokens > token_budget: + # Proportional shrink with a safety margin; the loop guards against the + # estimate landing high, and always makes progress. + keep = min(len(output_str) - 1, max(1, int(len(output_str) * token_budget / tokens * 0.95))) + output_str = output_str[:keep] + tokens = count_cl100k_tokens(output_str) + return {**entry, "output": {"truncated": True, "content": output_str}} + + +def split_context_history(context_history: list[dict], max_context_tokens: int) -> list[list[dict]]: + """Partition tool-result history into chunks that each fit the prompt budget. + + Greedy chronological packing: blocks keep their order, and a chunk closes + when the next block would push its rendered size past the budget. A single + block bigger than the whole budget is split on result-entry boundaries + (``observations``/``memories``/``results``) into synthetic partial blocks, + so evidence is split across chunks rather than dropped — the failure mode + of the old ``break`` was answering from nothing while citing everything + (#3122). Only an *indivisible* over-budget entry gets token-cut. + + Returns at least one chunk when history is non-empty; every original + result entry appears in exactly one chunk. + """ + budget = max(_MIN_SPLIT_CHUNK_TOKENS, int(max_context_tokens * _FINAL_PROMPT_CONTEXT_FRACTION)) + chunks: list[list[dict]] = [] + current: list[dict] = [] + current_tokens = 0 + + def _close_current() -> None: + nonlocal current, current_tokens + if current: + chunks.append(current) + current = [] + current_tokens = 0 + + def _append_block(entry: dict, tokens: int) -> None: + nonlocal current_tokens + if current and current_tokens + tokens > budget: + _close_current() + current.append(entry) + current_tokens += tokens + + for entry in context_history: + tokens = count_cl100k_tokens(_render_history_block(entry)) + if tokens <= budget: + _append_block(entry, tokens) + continue + + # Over-budget block: split it on result-entry boundaries. + output = entry["output"] + split_key = next( + ( + k + for k in _SPLITTABLE_RESULT_KEYS + if isinstance(output, dict) and isinstance(output.get(k), list) and output.get(k) + ), + None, + ) + if split_key is None: + cut = _cut_entry_to_budget(entry, budget) + _append_block(cut, count_cl100k_tokens(_render_history_block(cut))) + continue + + items = output[split_key] + piece: list = [] + for item in items: + candidate = {**entry, "output": {**output, split_key: piece + [item]}} + if piece and count_cl100k_tokens(_render_history_block(candidate)) > budget: + partial = {**entry, "output": {**output, split_key: piece}} + _append_block(partial, count_cl100k_tokens(_render_history_block(partial))) + piece = [] + candidate = {**entry, "output": {**output, split_key: [item]}} + single_tokens = count_cl100k_tokens(_render_history_block(candidate)) + if not piece and single_tokens > budget: + cut = _cut_entry_to_budget({**entry, "output": {**output, split_key: [item]}}, budget) + _append_block(cut, count_cl100k_tokens(_render_history_block(cut))) + else: + piece.append(item) + if piece: + partial = {**entry, "output": {**output, split_key: piece}} + _append_block(partial, count_cl100k_tokens(_render_history_block(partial))) + + _close_current() + return chunks + + +def _bank_identity_section(bank_profile: dict, additional_context: str | None) -> list[str]: + """The shared bank-identity/disposition/context head of a synthesis prompt.""" name = bank_profile.get("name", "Assistant") mission = bank_profile.get("mission", "") - parts.append(f"## Memory Bank Context\nName: {name}") + parts = [f"## Memory Bank Context\nName: {name}"] if mission: parts.append(f"Mission: {mission}") - # Disposition traits if present disposition = bank_profile.get("disposition", {}) if disposition: traits = [] @@ -484,9 +602,51 @@ def build_final_prompt( if traits: parts.append(f"Disposition: {', '.join(traits)}") - # Additional context from caller if additional_context: parts.append(f"\n## Additional Context\n{additional_context}") + return parts + + +def _length_directive(max_tokens: int | None) -> str | None: + """Soft visible-length directive for a synthesis prompt, or None. + + ``max_tokens`` is the desired *visible* length of the answer (e.g. a mental + model page's ``max_tokens``). It is communicated as a prompt directive + rather than enforced by truncating the provider call: on thinking models the + provider budget is consumed by reasoning tokens, so a hard cap cuts the page + off mid-word (#3365). The hard length guarantee is the post-hoc rewrite in + the agent; this directive just steers the model toward the target so the + rewrite rarely has to fire. + """ + if max_tokens is None: + return None + return ( + "\n## Length\n" + f"Aim for a complete, self-contained answer of approximately {max_tokens} tokens. " + "Finishing cleanly matters more than length: end on a complete sentence and NEVER stop " + "mid-word, mid-list, or mid-code-fence. If you near the budget, wrap up gracefully rather " + "than cutting off." + ) + + +def build_final_prompt( + query: str, + context_history: list[dict], + bank_profile: dict, + additional_context: str | None = None, + max_context_tokens: int = 100_000, + max_tokens: int | None = None, +) -> str: + """Build the final prompt when forcing a text response (no tools). + + ``max_tokens`` is the soft visible-length target (see ``_length_directive``). + + Callers overflow-proof this via ``split_context_history``: when the whole + history fits one chunk this renders it directly, and the per-block budget + walk below never trims. (The walk is kept as a defensive bound for direct + callers that skip splitting.) + """ + parts = _bank_identity_section(bank_profile, additional_context) # Tool call history — include as many entries as fit within the token budget, # preferring the most recent calls (they tend to be the most targeted). @@ -497,13 +657,7 @@ def build_final_prompt( rendered: list[str] = [] truncated = False for entry in reversed(context_history): - tool = entry["tool"] - output = entry["output"] - try: - output_str = json.dumps(output, indent=2, default=str, ensure_ascii=False) - except (TypeError, ValueError): - output_str = str(output) - block = f"\n### From {tool}:\n```json\n{output_str}\n```" + block = _render_history_block(entry) block_tokens = count_cl100k_tokens(block) if block_tokens > token_budget: truncated = True @@ -521,25 +675,92 @@ def build_final_prompt( parts.append(f"\n## Question\n{query}") # Final instructions + parts.append("\n## Instructions\n" + _FINAL_INSTRUCTIONS) + + length_directive = _length_directive(max_tokens) + if length_directive is not None: + parts.append(length_directive) + + return "\n".join(parts) + + +#: System prompt for the intermediate (map) calls of split synthesis. They do +#: NOT answer the question — they compress one chunk of retrieved data into +#: dated, cited claims that the reduce call can reason over. Dates and ids are +#: mandatory because conflicting facts can land in different chunks: only the +#: reduce call sees every chunk's claims, and it needs each claim's +#: ``mentioned_at`` to apply the latest-statement-wins supersession rule. +CLAIMS_SYSTEM_PROMPT = ( + "You extract evidence from retrieved memory data. You MUST ONLY use information " + "from the provided data. NEVER make up names, people, events, or entities.\n\n" + "Output a markdown bulleted list of factual claims relevant to the question. For EVERY claim:\n" + "- state the fact in one sentence, in the same language as the question;\n" + "- append its provenance in parentheses, exactly: " + "(mentioned_at: ; occurred: ; memory_ids: )\n\n" + "Rules:\n" + "- Be exhaustive over RELEVANT evidence; skip clearly irrelevant entries.\n" + "- Do NOT synthesize, conclude, resolve conflicts, or answer the question — report conflicting " + "claims as separate bullets with their dates; a later pass reconciles them.\n" + "- Copy memory ids exactly as they appear in the data.\n" + "- If nothing in the data is relevant, output exactly: (no relevant evidence)" +) + + +def build_chunk_claims_prompt(query: str, chunk: list[dict]) -> str: + """Build the user prompt for one intermediate (map) call of split synthesis.""" + parts = ["## Retrieved Data (extract relevant claims from this data)"] + for entry in chunk: + parts.append(_render_history_block(entry)) + parts.append(f"\n## Question\n{query}") parts.append( "\n## Instructions\n" - "Provide a thoughtful answer by synthesizing and reasoning from the retrieved data above. " - "You can make reasonable inferences from the memories, but don't completely fabricate information. " - "If the exact answer isn't stated, use what IS stated to give the best possible answer. " - "Only say 'I don't have information' if the retrieved data is truly unrelated to the question.\n\n" - "IMPORTANT: Output ONLY the final answer. Do NOT include meta-commentary like " - '"I\'ll search..." or "Let me analyze...". Do NOT explain your reasoning process. ' - "Just provide the direct synthesized answer." + "List every claim in the retrieved data relevant to the question, one bullet per claim, " + "each with its (mentioned_at: ...; occurred: ...; memory_ids: ...) provenance. " + "Do not answer the question." ) + return "\n".join(parts) - if max_tokens is not None: - parts.append( - "\n## Length\n" - f"Aim for a complete, self-contained answer of approximately {max_tokens} tokens. " - "Finishing cleanly matters more than length: end on a complete sentence and NEVER stop " - "mid-word, mid-list, or mid-code-fence. If you near the budget, wrap up gracefully rather " - "than cutting off." - ) + +def build_reduce_prompt( + query: str, + claim_sections: list[str], + bank_profile: dict, + additional_context: str | None = None, + max_tokens: int | None = None, +) -> str: + """Build the final prompt that synthesizes the answer from per-chunk claims. + + The retrieved data exceeded the context budget, so it was split into chunks + and each chunk was compressed to dated, cited claims by a parallel LLM call. + This prompt hands ALL the claim sets to one model. Conflicting facts may sit + in different sections — that is why the claims carry ``mentioned_at``: the + supersession rule (latest statement wins) must be applied across sections, + not within one. + """ + parts = _bank_identity_section(bank_profile, additional_context) + + parts.append( + "\n## Retrieved Evidence (synthesize and reason from these claims)\n" + "The retrieved data was processed in parallel passes; each section below holds one pass's " + "extracted claims with provenance dates and memory ids. Treat the sections as ONE evidence " + "pool: related and conflicting claims may appear in different sections." + ) + for i, section in enumerate(claim_sections, 1): + parts.append(f"\n### Evidence pass {i}:\n{section}") + + parts.append(f"\n## Question\n{query}") + + parts.append( + "\n## Instructions\n" + "When claims about the same fact conflict, the claim with the LATEST mentioned_at date is " + "authoritative — later statements supersede earlier ones, regardless of which section they " + "appear in. If equally-recent claims disagree and nothing resolves them, say so explicitly " + "rather than picking one.\n\n" + _FINAL_INSTRUCTIONS + ) + + length_directive = _length_directive(max_tokens) + if length_directive is not None: + parts.append(length_directive) return "\n".join(parts) diff --git a/hindsight-api-slim/tests/test_reflect_agent.py b/hindsight-api-slim/tests/test_reflect_agent.py index 3527a05764..8669182f0e 100644 --- a/hindsight-api-slim/tests/test_reflect_agent.py +++ b/hindsight-api-slim/tests/test_reflect_agent.py @@ -899,7 +899,8 @@ def mock_functions_with_large_output(self): async def test_proactive_guard_fires_when_budget_exceeded(self, mock_llm, mock_functions_with_large_output): """When token count exceeds max_context_tokens after a tool call, the agent should immediately synthesize from gathered evidence instead of making - another LLM call that would overflow.""" + another LLM call that would overflow. Evidence beyond the prompt budget + is split-synthesized (parallel claim extraction + reduce), never dropped.""" # First call: LLM calls recall (forced by iter 0 with no mental models) mock_llm.call_with_tools.return_value = LLMToolCallResult( tool_calls=[LLMToolCall(id="1", name="recall", arguments={"query": "test"})], @@ -920,8 +921,12 @@ async def test_proactive_guard_fires_when_budget_exceeded(self, mock_llm, mock_f # call_with_tools was called once (for the forced recall), then the guard # kicked in — no further tool-call iterations assert mock_llm.call_with_tools.call_count == 1 - # llm.call() was invoked to generate the final synthesis - mock_llm.call.assert_called_once() + # The synthesis ran: at least one no-tools call, ending with the final + # (single-shot or reduce) call. Whether the history split depends on its + # rendered size vs the floored chunk budget — both shapes are valid here. + assert mock_llm.call.call_count >= 1 + scopes = [c.scope for c in result.llm_trace] + assert scopes[-1] == "final" @pytest.mark.asyncio async def test_context_overflow_error_skips_retry(self, mock_llm, mock_functions_with_large_output): diff --git a/hindsight-api-slim/tests/test_reflect_split_synthesis.py b/hindsight-api-slim/tests/test_reflect_split_synthesis.py new file mode 100644 index 0000000000..5203569035 --- /dev/null +++ b/hindsight-api-slim/tests/test_reflect_split_synthesis.py @@ -0,0 +1,352 @@ +"""Split synthesis: forced final synthesis must not drop retrieved evidence. + +When the reflect agent is forced to answer without tools (context guard, last +iteration, LLM error, or a clean stop) and the accumulated tool results exceed +the prompt budget, the old ``build_final_prompt`` dropped any over-budget block +whole — plus every older one — so the synthesis model could see an empty +Retrieved Data section while the response still attached hundreds of citations +(#3122). Split synthesis partitions the history into budget-sized chunks, +compresses each in parallel into dated, cited claims, and synthesizes the +answer from every chunk's claims: nothing retrieved is dropped. + +The splitter tests are pure functions; the agent tests drive the map/reduce +flow with a mock LLM. +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from hindsight_api.engine.reflect.agent import run_reflect_agent +from hindsight_api.engine.reflect.prompts import ( + _MIN_SPLIT_CHUNK_TOKENS, + _render_history_block, + build_chunk_claims_prompt, + build_reduce_prompt, + split_context_history, +) +from hindsight_api.engine.reflect.tokenization import count_cl100k_tokens +from hindsight_api.engine.response_models import LLMToolCall, LLMToolCallResult, TokenUsage + +# The splitter floors its per-chunk budget at _MIN_SPLIT_CHUNK_TOKENS, so tests +# use budgets above the floor to exercise the packing logic itself. +_BUDGET_TOKENS = 2048 +_MAX_CONTEXT = int(_BUDGET_TOKENS / 0.8) + + +def _entry(tool: str, key: str, n_items: int, item_chars: int, id_prefix: str = "mem") -> dict: + items = [{"id": f"{id_prefix}-{i}", "text": f"fact {i}: " + "x" * item_chars} for i in range(n_items)] + return {"tool": tool, "output": {key: items, "query": "q"}} + + +def _ids_in(chunks: list[list[dict]]) -> list[str]: + ids = [] + for chunk in chunks: + for entry in chunk: + output = entry["output"] + for key in ("observations", "memories", "results"): + for item in output.get(key, []) if isinstance(output, dict) else []: + ids.append(item["id"]) + return ids + + +class TestSplitContextHistory: + def test_small_history_is_one_chunk(self): + history = [_entry("recall", "memories", 3, 50)] + chunks = split_context_history(history, _MAX_CONTEXT) + assert chunks == [history] + + def test_empty_history_is_no_chunks(self): + assert split_context_history([], _MAX_CONTEXT) == [] + + def test_blocks_pack_greedily_in_order(self): + """Several fitting blocks distribute across chunks without reordering + and without losing a single result entry.""" + history = [_entry("recall", "memories", 8, 300, id_prefix=f"b{b}") for b in range(6)] + chunks = split_context_history(history, _MAX_CONTEXT) + + assert len(chunks) > 1 + for chunk in chunks: + rendered = "".join(_render_history_block(e) for e in chunk) + assert count_cl100k_tokens(rendered) <= _BUDGET_TOKENS + # Every entry survives, in original order. + original_ids = [item["id"] for e in history for item in e["output"]["memories"]] + assert _ids_in(chunks) == original_ids + + def test_oversized_block_splits_on_entry_boundaries(self): + """One block bigger than the whole budget is split into partial blocks — + the exact case the old code dropped entirely.""" + history = [_entry("search_observations", "observations", 40, 400)] + chunks = split_context_history(history, _MAX_CONTEXT) + + assert len(chunks) > 1 + for chunk in chunks: + rendered = "".join(_render_history_block(e) for e in chunk) + assert count_cl100k_tokens(rendered) <= _BUDGET_TOKENS + # Partial blocks keep the tool name and sibling keys. + assert all(e["tool"] == "search_observations" for e in chunk) + assert all(e["output"]["query"] == "q" for e in chunk) + assert _ids_in(chunks) == [f"mem-{i}" for i in range(40)] + + def test_indivisible_oversized_entry_is_token_cut(self): + """A single entry (or list-less output) bigger than the budget is the one + case that cannot be split; it gets token-cut instead of dropped.""" + giant = {"tool": "expand", "output": {"full_text": "y" * 40_000}} + chunks = split_context_history([giant], _MAX_CONTEXT) + + assert len(chunks) == 1 and len(chunks[0]) == 1 + cut = chunks[0][0] + assert cut["output"]["truncated"] is True + assert cut["output"]["content"] + assert count_cl100k_tokens(_render_history_block(cut)) <= _BUDGET_TOKENS + 32 + + def test_budget_floor_prevents_per_entry_fanout(self): + """A tiny configured budget must not shred the history into one chunk + per fact — the floor keeps chunks ~1k tokens.""" + history = [_entry("recall", "memories", 30, 60)] + chunks = split_context_history(history, max_context_tokens=1) + total_entries = sum(len(c) for c in chunks) + assert len(chunks) <= 4 + assert _ids_in(chunks) == [f"mem-{i}" for i in range(30)] + assert total_entries < 30, "history was shredded into per-entry chunks" + + +class TestSplitSynthesisPrompts: + def test_chunk_claims_prompt_carries_evidence_and_question(self): + chunk = [_entry("recall", "memories", 2, 30)] + prompt = build_chunk_claims_prompt("what happened?", chunk) + assert "mem-0" in prompt and "mem-1" in prompt + assert "## Question\nwhat happened?" in prompt + assert "Do not answer the question" in prompt + + def test_reduce_prompt_carries_all_sections_and_supersession_rule(self): + sections = ["- claim A (mentioned_at: 2026-01-01; ...)", "- claim B (mentioned_at: 2026-02-01; ...)"] + prompt = build_reduce_prompt("what happened?", sections, {"name": "Bank", "mission": "m"}) + assert "### Evidence pass 1:" in prompt and "### Evidence pass 2:" in prompt + assert "claim A" in prompt and "claim B" in prompt + assert "LATEST mentioned_at" in prompt + assert "## Question\nwhat happened?" in prompt + + +class TestSplitSynthesisAgentFlow: + """Drives the forced-synthesis path with a mock LLM and asserts the + map/reduce call pattern.""" + + @staticmethod + def _mock_llm(final_answer: str = "Reduced answer."): + llm = MagicMock() + llm.call_with_tools = AsyncMock() + + async def _call(messages, **kwargs): + # Map calls use the claims system prompt; the reduce/final call uses + # the final system prompt. Answer accordingly so the test can tell + # which output made it into the result. + if "extract evidence" in messages[0]["content"]: + return ( + f"- claim from prompt of {count_cl100k_tokens(messages[1]['content'])} tokens " + "(mentioned_at: 2026-01-01; occurred: unknown; memory_ids: mem-1)", + TokenUsage(input_tokens=10, output_tokens=5, total_tokens=15), + ) + return (final_answer, TokenUsage(input_tokens=20, output_tokens=10, total_tokens=30)) + + llm.call = AsyncMock(side_effect=_call) + return llm + + @staticmethod + def _functions(recall_payload: dict): + return { + "search_mental_models_fn": AsyncMock(return_value={"mental_models": []}), + "search_observations_fn": AsyncMock(return_value={"observations": []}), + "recall_fn": AsyncMock(return_value=recall_payload), + "expand_fn": AsyncMock(return_value={"memories": []}), + } + + @pytest.mark.asyncio + async def test_overflowing_history_triggers_map_reduce(self): + """History beyond the budget → one map call per chunk, then one reduce + call whose output is the final answer.""" + big = {"memories": [{"id": f"mem-{i}", "text": "fact " + "z" * 600} for i in range(60)]} + llm = self._mock_llm() + llm.call_with_tools.return_value = LLMToolCallResult( + tool_calls=[LLMToolCall(id="1", name="recall", arguments={"query": "q"})], + finish_reason="tool_calls", + ) + + result = await run_reflect_agent( + llm_config=llm, + bank_id="b", + query="what do you know?", + bank_profile={"name": "Test", "mission": "Testing"}, + max_context_tokens=int(_MIN_SPLIT_CHUNK_TOKENS / 0.8) + 10, + **self._functions(big), + ) + + scopes = [c.scope for c in result.llm_trace] + map_scopes = [s for s in scopes if s.startswith("final_map_")] + assert len(map_scopes) >= 2, f"expected parallel map calls, got scopes {scopes}" + assert scopes[-1] == "final", "the reduce call must be the last LLM call" + assert result.text == "Reduced answer." + + # The reduce prompt must carry every map output (one claim per chunk). + reduce_call = llm.call.await_args_list[-1] + reduce_prompt = reduce_call.kwargs.get("messages", reduce_call.args[0] if reduce_call.args else None)[1][ + "content" + ] + assert reduce_prompt.count("- claim from prompt of") == len(map_scopes) + + @pytest.mark.asyncio + async def test_caller_max_tokens_is_a_directive_not_a_transport_cap(self): + """The caller's max_tokens must not cap any synthesis call at the + transport level (#3365 decoupling): it reaches the reduce prompt as a + visible-length directive, while every map/reduce call carries the + config transport cap (None by default).""" + big = {"memories": [{"id": f"mem-{i}", "text": "fact " + "z" * 600} for i in range(60)]} + llm = self._mock_llm() + llm.call_with_tools.return_value = LLMToolCallResult( + tool_calls=[LLMToolCall(id="1", name="recall", arguments={"query": "q"})], + finish_reason="tool_calls", + ) + + await run_reflect_agent( + llm_config=llm, + bank_id="b", + query="q?", + bank_profile={"name": "Test", "mission": "Testing"}, + max_context_tokens=int(_MIN_SPLIT_CHUNK_TOKENS / 0.8) + 10, + max_tokens=64, + **self._functions(big), + ) + + caps = [c.kwargs["max_completion_tokens"] for c in llm.call.await_args_list] + assert all(cap is None for cap in caps), ( + "no synthesis call may inherit the caller's answer budget as a transport cap" + ) + reduce_prompt = llm.call.await_args_list[-1].kwargs["messages"][1]["content"] + assert "approximately 64 tokens" in reduce_prompt, "length target must reach the reduce prompt" + + @pytest.mark.asyncio + async def test_fitting_history_stays_single_call(self): + """No overflow → exactly the pre-existing single forced-synthesis call.""" + small = {"memories": [{"id": "mem-1", "text": "one small fact"}]} + llm = self._mock_llm("Direct answer.") + # First turn recalls; second turn stops with no tool calls → forced synthesis. + llm.call_with_tools.side_effect = [ + LLMToolCallResult( + tool_calls=[LLMToolCall(id="1", name="recall", arguments={"query": "q"})], + finish_reason="tool_calls", + ), + LLMToolCallResult(tool_calls=[], finish_reason="stop", content="done"), + ] + + result = await run_reflect_agent( + llm_config=llm, + bank_id="b", + query="q?", + bank_profile={"name": "Test", "mission": "Testing"}, + **self._functions(small), + ) + + assert result.text == "Direct answer." + scopes = [c.scope for c in result.llm_trace] + assert scopes == ["agent_1", "agent_2", "final"], f"unexpected scopes {scopes}" + + @pytest.mark.asyncio + async def test_map_prompts_partition_the_evidence(self): + """Every retrieved memory id reaches exactly one map prompt — the + no-evidence-dropped guarantee, asserted end to end.""" + n = 60 + big = {"memories": [{"id": f"mem-{i}", "text": "fact " + "z" * 600} for i in range(n)]} + llm = self._mock_llm() + llm.call_with_tools.return_value = LLMToolCallResult( + tool_calls=[LLMToolCall(id="1", name="recall", arguments={"query": "q"})], + finish_reason="tool_calls", + ) + + await run_reflect_agent( + llm_config=llm, + bank_id="b", + query="q?", + bank_profile={"name": "Test", "mission": "Testing"}, + max_context_tokens=int(_MIN_SPLIT_CHUNK_TOKENS / 0.8) + 10, + **self._functions(big), + ) + + map_prompts = [ + c.kwargs["messages"][1]["content"] + for c in llm.call.await_args_list + if "extract evidence" in c.kwargs["messages"][0]["content"] + ] + for i in range(n): + holders = [p for p in map_prompts if f'"mem-{i}"' in p] + assert len(holders) == 1, f"mem-{i} appears in {len(holders)} map prompts" + + +@pytest.mark.hs_llm_core +class TestSplitSynthesisRealLLM: + """Real-LLM, judge-verified coverage of the map/reduce path. + + The mock tests above pin the mechanics (partitioning, call pattern, budget + caps). What only a real model can verify is that evidence actually SURVIVES + the pipeline: a distinctive fact placed in the first chunk and another + placed in the last must both be extractable from the final answer. Under + the old drop-whole-blocks behavior the answer was a confident "no + information" — this is the regression #3122 describes, judged rather than + string-matched because the model paraphrases. + """ + + @pytest.mark.asyncio + async def test_facts_from_distinct_chunks_reach_the_answer(self, llm_config): + from tests.llm_judge import assert_meets_criteria + + # ~60 filler memories force several chunks under the floored budget. + # The two distinctive facts sit at the extremes so they are guaranteed + # to land in different map calls. + memories = [{"id": "mem-zara", "text": "Zara keeps bees on her rooftop.", "mentioned_at": "2026-01-05"}] + memories += [ + { + "id": f"mem-filler-{i}", + "text": f"Team member {i} attended the weekly sync and reported routine progress on task {i}. " + + "Nothing notable happened. " * 8, + "mentioned_at": "2026-01-10", + } + for i in range(60) + ] + memories.append( + {"id": "mem-marco", "text": "Marco collects vintage synthesizers.", "mentioned_at": "2026-02-01"} + ) + + functions = { + "search_mental_models_fn": AsyncMock(return_value={"mental_models": []}), + "search_observations_fn": AsyncMock(return_value={"observations": []}), + "recall_fn": AsyncMock(return_value={"memories": memories}), + "expand_fn": AsyncMock(return_value={"memories": []}), + } + + result = await run_reflect_agent( + llm_config=llm_config, + bank_id="split-synth-real", + query="What hobbies do the people in memory have?", + bank_profile={"name": "Test", "mission": "Remember the team."}, + max_context_tokens=int(_MIN_SPLIT_CHUNK_TOKENS / 0.8) + 10, + **functions, + ) + + scopes = [c.scope for c in result.llm_trace] + assert any(s.startswith("final_map_") for s in scopes), f"split synthesis did not engage: {scopes}" + + await assert_meets_criteria( + response=result.text, + criteria=( + "The answer mentions BOTH hobbies: Zara's beekeeping (bees/beehives) AND Marco's " + "vintage synthesizer collecting. Mentioning only one, or answering that there is " + "no information, fails." + ), + context=( + "The memory data contained two hobby facts separated by dozens of filler entries: " + "'Zara keeps bees on her rooftop' and 'Marco collects vintage synthesizers'. The " + "synthesis pipeline splits the data into chunks, so each fact traversed a different " + "intermediate extraction call." + ), + )