diff --git a/agentigrid/README.md b/agentigrid/README.md index cb1f0ebb..59f6fdda 100644 --- a/agentigrid/README.md +++ b/agentigrid/README.md @@ -367,7 +367,7 @@ agentigrid ./data/exago/examples/case_ACTIVSg200.m "Find the maximum load scalin # With options agentigrid ./data/exago/examples/case_ACTIVSg200.m "Minimize generation cost" \ - --backend anthropic --model claude-sonnet-4-20250514 \ + --backend anthropic --model claude-sonnet-4-6 \ --app opflow --max-iter 30 --verbose # Quiet mode (only show final summary) @@ -424,7 +424,7 @@ python -m agentigrid ./data/exago/examples/case_ACTIVSg200.m "test goal" --dry-r Version 0.1.0 ============================================================ Backend: anthropic - Model: claude-sonnet-4-20250514 + Model: claude-sonnet-4-6 Application: opflow Base case: data/exago/examples/case_ACTIVSg200.m Goal: Find the maximum load scaling factor @@ -435,7 +435,7 @@ python -m agentigrid ./data/exago/examples/case_ACTIVSg200.m "test goal" --dry-r [Iter 0] Running base case simulation... [Iter 0] Base case: CONVERGED, cost=$27,557.57 -[Iter 1] Sending prompt to anthropic (claude-sonnet-4-20250514)... +[Iter 1] Sending prompt to anthropic (claude-sonnet-4-6)... [Iter 1] LLM action: modify — "Scale all loads +20%" [Iter 1] Applied 1 command(s), 0 skipped [Iter 1] Simulation completed in 0.04s — CONVERGED, cost=$33,019.55 @@ -450,7 +450,7 @@ python -m agentigrid ./data/exago/examples/case_ACTIVSg200.m "test goal" --dry-r ============================================================ Goal: Find the maximum load scaling factor Application: opflow - Backend: anthropic (claude-sonnet-4-20250514) + Backend: anthropic (claude-sonnet-4-6) Iterations: 6 (of max 20) Duration: 18.3 seconds Tokens used: ~12,450 (prompt: 9,200, completion: 3,250) diff --git a/agentigrid/agentigrid/backends/__init__.py b/agentigrid/agentigrid/backends/__init__.py index dd3651b0..15114dfa 100644 --- a/agentigrid/agentigrid/backends/__init__.py +++ b/agentigrid/agentigrid/backends/__init__.py @@ -2,10 +2,10 @@ from __future__ import annotations -from agentigrid.backends.base import LLMBackend, LLMResponse +from agentigrid.backends.base import LLMBackend, LLMResponse, UsageMeter from agentigrid.config import LLMConfig -__all__ = ["LLMBackend", "LLMResponse", "create_backend"] +__all__ = ["LLMBackend", "LLMResponse", "UsageMeter", "create_backend"] def create_backend(config: LLMConfig) -> LLMBackend: diff --git a/agentigrid/agentigrid/backends/anthropic_backend.py b/agentigrid/agentigrid/backends/anthropic_backend.py index 10bbea96..594ba46d 100644 --- a/agentigrid/agentigrid/backends/anthropic_backend.py +++ b/agentigrid/agentigrid/backends/anthropic_backend.py @@ -27,6 +27,9 @@ def __init__(self, config: LLMConfig) -> None: raise ImportError("anthropic package is required for the Anthropic backend") self._config = config + # None = not yet known; True/False once a call has succeeded / the + # model has rejected the parameter (see complete()). + self.temperature_sent: Optional[bool] = None api_key = os.environ.get(config.api_key_env) if not api_key: logger.warning( @@ -40,6 +43,26 @@ def name(self) -> str: """Return the backend name.""" return "anthropic" + def _system_param(self, system_prompt: str): + """System prompt in the form the Messages API expects. + + With ``prompt_cache`` on, the system prompt is sent as a single text block + carrying an ephemeral ``cache_control`` breakpoint. AgentiGrid builds the + system prompt once per session, so every later iteration reads it from the + cache instead of paying the full input price again. This affects billing + only; the model sees identical text either way. Prompts below the model's + minimum cacheable length are silently processed uncached by the API. + """ + if not getattr(self._config, "prompt_cache", False): + return system_prompt + return [ + { + "type": "text", + "text": system_prompt, + "cache_control": {"type": "ephemeral"}, + } + ] + def supports_json_mode(self) -> bool: """Anthropic does not have a native JSON output mode.""" return False @@ -54,20 +77,45 @@ def complete( temp = temperature if temperature is not None else self._config.temperature try: - response = self._client.messages.create( + kwargs = dict( model=self._config.model, max_tokens=self._config.max_tokens, - temperature=temp, - system=system_prompt, + system=self._system_param(system_prompt), messages=[{"role": "user", "content": user_prompt}], ) + if self.temperature_sent is not False: + kwargs["temperature"] = temp + try: + response = self._client.messages.create(**kwargs) + if "temperature" in kwargs: + self.temperature_sent = True + except Exception as exc: + # Newer models reject the sampling-temperature parameter + # ("`temperature` is deprecated for this model"). Retry once + # without it and stop sending it for the rest of the session; + # the journal records temperature_sent=False. + if "temperature" in kwargs and "temperature" in str(exc).lower(): + logger.warning("Model %s rejects `temperature`; retrying without it.", self._config.model) + kwargs.pop("temperature") + self.temperature_sent = False + response = self._client.messages.create(**kwargs) + else: + raise raw_text = "" + content_types = [] for block in response.content: + content_types.append(getattr(block, "type", None)) if block.type == "text": raw_text += block.text + stop_reason = getattr(response, "stop_reason", None) + if stop_reason == "max_tokens": + logger.warning("Response stopped at max_tokens=%s (blocks: %s, %d text chars)", + self._config.max_tokens, content_types, len(raw_text)) - prompt_tokens = getattr(response.usage, "input_tokens", None) + prompt_tokens, cache_creation, cache_read = _input_token_breakdown( + response.usage + ) completion_tokens = getattr(response.usage, "output_tokens", None) json_data, json_error = extract_json(raw_text) @@ -80,6 +128,10 @@ def complete( backend=self.name(), prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, + cache_creation_tokens=cache_creation, + cache_read_tokens=cache_read, + stop_reason=stop_reason, + content_types=content_types, ) except Exception as exc: @@ -93,4 +145,23 @@ def complete( backend=self.name(), prompt_tokens=None, completion_tokens=None, + api_error=True, ) + + +def _input_token_breakdown(usage) -> tuple[Optional[int], Optional[int], Optional[int]]: + """Return (total_input, cache_creation, cache_read) from an API usage object. + + With caching, the API's ``input_tokens`` counts only tokens after the last + cache breakpoint. The total input is ``input_tokens + cache_creation_input_tokens + + cache_read_input_tokens``; returning that total keeps ``prompt_tokens`` + comparable with runs made before caching was enabled. + """ + base = getattr(usage, "input_tokens", None) + creation = getattr(usage, "cache_creation_input_tokens", None) + read = getattr(usage, "cache_read_input_tokens", None) + creation = creation if isinstance(creation, int) else None + read = read if isinstance(read, int) else None + if not isinstance(base, int): + return None, creation, read + return base + (creation or 0) + (read or 0), creation, read diff --git a/agentigrid/agentigrid/backends/base.py b/agentigrid/agentigrid/backends/base.py index 32c4581a..2e681924 100644 --- a/agentigrid/agentigrid/backends/base.py +++ b/agentigrid/agentigrid/backends/base.py @@ -18,6 +18,20 @@ class LLMResponse: backend: str prompt_tokens: Optional[int] completion_tokens: Optional[int] + # Prompt-cache accounting (Anthropic). ``prompt_tokens`` always reports the + # TOTAL input tokens (uncached + cache writes + cache reads), so its meaning + # is the same with or without caching; these break that total down. + cache_creation_tokens: Optional[int] = None + cache_read_tokens: Optional[int] = None + # True when the request itself failed (network, auth, invalid request...), + # as opposed to the model answering with unparseable text. Lets the agent + # loop record API failures separately from rejected proposals. + api_error: bool = False + # Why generation stopped ("end_turn", "max_tokens", ...) and the content + # block types returned, when the backend reports them. A "max_tokens" stop + # is an output-cap limit, not a model proposal, and is recorded as such. + stop_reason: Optional[str] = None + content_types: Optional[list] = None class LLMBackend(ABC): @@ -51,3 +65,61 @@ def name(self) -> str: def supports_json_mode(self) -> bool: """Whether this backend supports a native JSON output mode.""" ... + + +class UsageMeter(LLMBackend): + """Transparent wrapper that meters token usage across EVERY backend call. + + The agent loop's own counters cover only the main per-iteration call; the + post-search analysis calls are not counted there. Wrapping the backend once + gives complete per-run totals (including prompt-cache writes and reads) for + cost accounting, without touching any call site. Behaviour is unchanged: + every call is delegated verbatim. + """ + + def __init__(self, inner: LLMBackend) -> None: + self._inner = inner + self.calls = 0 + self.prompt_tokens = 0 + self.completion_tokens = 0 + self.cache_creation_tokens = 0 + self.cache_read_tokens = 0 + self.api_errors = 0 + self.truncated = 0 + + def complete(self, system_prompt, user_prompt, temperature=None): + resp = self._inner.complete(system_prompt, user_prompt, temperature) + self.calls += 1 + if getattr(resp, "api_error", False): + self.api_errors += 1 + if getattr(resp, "stop_reason", None) == "max_tokens": + self.truncated += 1 + self.prompt_tokens += resp.prompt_tokens or 0 + self.completion_tokens += resp.completion_tokens or 0 + self.cache_creation_tokens += resp.cache_creation_tokens or 0 + self.cache_read_tokens += resp.cache_read_tokens or 0 + return resp + + def name(self) -> str: + return self._inner.name() + + def supports_json_mode(self) -> bool: + return self._inner.supports_json_mode() + + def __getattr__(self, item): # delegate anything backend-specific + if item == "_inner": # not yet set (e.g. during copy/unpickle) + raise AttributeError(item) + return getattr(self._inner, item) + + def totals(self) -> dict: + """Per-run usage totals, as written to the journal (``llm_usage``).""" + return { + "calls": self.calls, + "prompt_tokens": self.prompt_tokens, + "completion_tokens": self.completion_tokens, + "cache_creation_tokens": self.cache_creation_tokens, + "cache_read_tokens": self.cache_read_tokens, + "api_errors": self.api_errors, + "truncated": self.truncated, + "temperature_sent": getattr(self._inner, "temperature_sent", None), + } diff --git a/agentigrid/agentigrid/backends/ollama_backend.py b/agentigrid/agentigrid/backends/ollama_backend.py index f7fb1541..ae9faf6b 100644 --- a/agentigrid/agentigrid/backends/ollama_backend.py +++ b/agentigrid/agentigrid/backends/ollama_backend.py @@ -120,4 +120,5 @@ def complete( backend=self.name(), prompt_tokens=None, completion_tokens=None, + api_error=True, ) diff --git a/agentigrid/agentigrid/backends/openai_backend.py b/agentigrid/agentigrid/backends/openai_backend.py index 0b4261ab..47ecb8f0 100644 --- a/agentigrid/agentigrid/backends/openai_backend.py +++ b/agentigrid/agentigrid/backends/openai_backend.py @@ -104,4 +104,5 @@ def complete( backend=self.name(), prompt_tokens=None, completion_tokens=None, + api_error=True, ) diff --git a/agentigrid/agentigrid/config.py b/agentigrid/agentigrid/config.py index 7f0cb8d5..ea0c01a9 100644 --- a/agentigrid/agentigrid/config.py +++ b/agentigrid/agentigrid/config.py @@ -36,13 +36,14 @@ }, "llm": { "backend": "anthropic", - "model": "claude-sonnet-4-20250514", + "model": "claude-sonnet-4-6", "api_key_env": "ANTHROPIC_API_KEY", "openai_base_url": None, "ollama_host": "http://localhost:11434", "ollama_cloud_host": None, "temperature": 0.3, "max_tokens": 4096, + "prompt_cache": True, }, "search": { "max_iterations": 20, @@ -159,6 +160,9 @@ class LLMConfig: ollama_cloud_host: Optional[str] temperature: float max_tokens: int + # Anthropic prompt caching of the (per-session constant) system prompt. + # Billing-only: does not change model outputs. Ignored by other backends. + prompt_cache: bool = True @dataclass(frozen=True) @@ -374,6 +378,15 @@ def load_config( target[parts[-1]] = value logger.debug("Applied CLI overrides: %s", cli_overrides) + # AGENTIGRID_MAX_TOKENS overrides the output-token cap (e.g. to fix it for a + # batch of runs); the value used is recorded in the journal's llm_usage. + _mt = os.environ.get("AGENTIGRID_MAX_TOKENS", "").strip() + if _mt: + try: + merged["llm"] = {**merged["llm"], "max_tokens": int(_mt)} + except ValueError: + logger.warning("Ignoring non-integer AGENTIGRID_MAX_TOKENS=%r", _mt) + # Build frozen dataclass sections exago_path_fields = { "binary_dir", "opflow_binary", "scopflow_binary", "tcopflow_binary", diff --git a/agentigrid/agentigrid/exago_engine/agent_loop.py b/agentigrid/agentigrid/exago_engine/agent_loop.py index 20e59c1e..10165cce 100644 --- a/agentigrid/agentigrid/exago_engine/agent_loop.py +++ b/agentigrid/agentigrid/exago_engine/agent_loop.py @@ -17,7 +17,7 @@ from pathlib import Path from typing import Callable, Optional -from agentigrid.backends import create_backend +from agentigrid.backends import UsageMeter, create_backend from agentigrid.backends.base import LLMBackend, LLMResponse from agentigrid.config import AppConfig from agentigrid.exago_engine.commands import parse_command @@ -30,7 +30,7 @@ compute_pareto_labels, format_variant_results, ) -from agentigrid.exago_engine.journal import JournalEntry, ObjectiveEntry, SearchJournal +from agentigrid.exago_engine.journal import JournalEntry, ObjectiveEntry, SearchJournal, format_iteration_count from agentigrid.exago_engine.metric_extractor import available_metrics, available_metrics_for_app, extract_all_metrics from agentigrid.exago_engine.modifier import apply_modifications, build_index_maps from agentigrid.exago_engine import sweep_metrics @@ -66,6 +66,9 @@ def _count_scenario_rows(scenario_path: Path) -> int: logger = logging.getLogger("agentigrid.exago_engine.agent_loop") +# Cap on the raw model response kept for a discarded iteration (journal telemetry). +DISCARDED_RESPONSE_MAX = 4000 + _MAX_CONSECUTIVE_PARSE_FAILURES = 3 @@ -765,7 +768,9 @@ def __init__( on_explore: Callable[[int, list[dict]], None] | None = None, ) -> None: self._config = config - self._backend: LLMBackend = create_backend(config.llm) + # Metered so per-run token totals (incl. prompt-cache reads/writes) + # cover every LLM call, not only the main per-iteration call. + self._backend: LLMBackend = UsageMeter(create_backend(config.llm)) self._executor = SimulationExecutor(config.exago, config.output) self._journal = SearchJournal() self._quiet = quiet @@ -1223,17 +1228,67 @@ def run(self, base_case: Path, goal: str) -> SearchSession: # 3. Agent loop max_iter = self._config.search.max_iterations + # Optional stall breaker: stop after N consecutive no-progress iterations + # (iterations that record no new journal entry — e.g. the model repeating + # an invalid action). 0/unset disables it (the default), so batch runs always + # reach max_iter and stay comparable; the UI can set it. + try: + stall_limit = int(os.environ.get("AGENTIGRID_STALL_LIMIT", "0") or "0") + except ValueError: + stall_limit = 0 + stall = 0 for iteration in range(1, max_iter + 1): if self._stop_requested: session.termination_reason = "user_stopped" self._print("\nSearch stopped by user.") break - action_type, should_continue = self._iteration(iteration, goal) - # Notify callback after each iteration + self._iteration_api_error = False + self._iteration_response_text = None + self._iteration_stop_reason = None + self._iteration_content_types = None + try: + action_type, should_continue = self._iteration(iteration, goal) + except Exception as e: + import traceback + self._print( + f"\n[Iter {iteration}] Action failed with an internal error " + f"({type(e).__name__}: {e}); discarding and continuing." + ) + traceback.print_exc() + self._error_feedback = ( + f"The previous action raised an internal error and was discarded: " + f"{type(e).__name__}: {e}. Respond with a well-formed action that " + f"strictly follows the schema." + ) + action_type, should_continue = "error", True + internal_error = True + else: + internal_error = False + # Did THIS iteration record a journal entry? A discarded/invalid action + # (common with weak local models) appends nothing. + latest_entry = self._journal.latest + recorded = latest_entry is not None and latest_entry.iteration == iteration + if action_type == "error" and not recorded: + self._record_discarded(iteration, internal_error) + # Notify the UI only on a real new entry; otherwise show a transient + # discarded card (avoids re-emitting the previous entry as a phantom). if self._on_iteration: - latest_entry = self._journal.latest - if latest_entry: + if recorded: self._on_iteration(iteration, latest_entry, action_type, self._latest_opflow) + elif action_type == "error": + self._emit_discarded(iteration) + # Stall breaker: count consecutive no-progress iterations. + if recorded: + stall = 0 + elif stall_limit: + stall += 1 + if stall >= stall_limit: + session.termination_reason = "stalled_no_progress" + self._print( + f"\nStopping early: {stall} consecutive no-progress " + f"iterations (stall limit {stall_limit})." + ) + break if not should_continue: if not session.termination_reason: session.termination_reason = "completed" @@ -1245,16 +1300,12 @@ def run(self, base_case: Path, goal: str) -> SearchSession: if not session.termination_reason: session.termination_reason = "completed" - # Final notification with termination reason - if self._on_iteration: - latest_entry = self._journal.latest - if latest_entry: - self._on_iteration( - latest_entry.iteration, - latest_entry, - session.termination_reason, - self._latest_opflow, - ) + # No final on_iteration re-emit: the last real entry was already sent by + # the per-iteration callback above. Re-emitting it duplicated the last + # timeline card, and when the final iterations were discarded it made a + # stale, lower-numbered entry appear AFTER the final iteration (e.g. a + # phantom "Iteration 6" after 20). The UI reads session.termination_reason + # when the search finishes. session.end_time = datetime.now().isoformat() session.total_prompt_tokens = self._total_prompt_tokens @@ -1272,6 +1323,88 @@ def run(self, base_case: Path, goal: str) -> SearchSession: self._finalize(session, elapsed) return session + # ------------------------------------------------------------------ + # Discarded-iteration diagnostic + # ------------------------------------------------------------------ + + def _record_discarded(self, iteration: int, internal: bool = False) -> None: + """Journal telemetry for an iteration whose action produced no entry. + + Persisted: iteration, kind (rejected / internal / api_error / + truncated), stop_reason and content block types when known, the + truncated feedback sent back to the model and, when the model answered, + its raw response (the rejected proposal itself, capped at + DISCARDED_RESPONSE_MAX characters). The entry list and all summary + statistics are unaffected.""" + stop = getattr(self, "_iteration_stop_reason", None) + if internal: + kind = "internal" + elif getattr(self, "_iteration_api_error", False): + kind = "api_error" + elif stop == "max_tokens": + kind = "truncated" # output cap hit: a harness limit, not a rejected proposal + else: + kind = "rejected" + record = { + "iteration": iteration, + "kind": kind, + "feedback": (self._error_feedback or "")[:300], + } + if stop is not None: + record["stop_reason"] = stop + ctypes = getattr(self, "_iteration_content_types", None) + if ctypes is not None: + record["content_types"] = list(ctypes) + text = getattr(self, "_iteration_response_text", None) + if text is not None and kind != "api_error": # an API error has no model response + record["response"] = text[:DISCARDED_RESPONSE_MAX] + record["response_truncated"] = len(text) > DISCARDED_RESPONSE_MAX + self._journal.discarded_actions.append(record) + + def _emit_discarded(self, iteration: int) -> None: + """Emit a UI-only 'discarded' timeline card for an iteration whose action + was rejected (invalid/malformed proposal, unknown action, parse error) + and therefore recorded no journal entry. + + The synthetic entry is deliberately NOT added to ``self._journal``: it + never affects summary stats, best-iteration selection, or the persisted + record. ``mode="discarded"`` keeps ``is_solve_iteration`` + False, and ``feasible=False`` / ``objective_value=None`` keep it out of + the session_manager's best-cost tracking. It exists purely to tell the + user *why* the iteration produced nothing, instead of showing a phantom + duplicate or a blank card. + """ + if not self._on_iteration: + return + reason = (self._error_feedback or "The model returned an invalid proposal.").strip() + # The UI label must attribute the failure to the MODEL, not the code: + # the raw feedback ("Unknown action X. Valid actions: ...") is written for + # the LLM and reads like a code assertion in the timeline. Keep only the + # first clause for the headline and prefix it clearly; the full corrective + # text still shows in the card's expander (llm_reasoning). + headline = reason.split(". ")[0].splitlines()[0].strip().rstrip(".") + if not headline: + headline = "invalid proposal" + headline = headline if len(headline) <= 120 else headline[:117] + "..." + synthetic = JournalEntry( + iteration=iteration, + description=f"LLM output rejected — {headline} (no change applied)", + commands=[], + objective_value=None, + feasible=False, + convergence_status="FAILED", + violations_count=0, + voltage_min=0.0, + voltage_max=0.0, + max_line_loading_pct=0.0, + total_gen_mw=0.0, + total_load_mw=0.0, + llm_reasoning=reason, # full detail shown in the card's expander + mode="discarded", + elapsed_seconds=0.0, + ) + self._on_iteration(iteration, synthetic, "discarded", None) + # ------------------------------------------------------------------ # Single iteration # ------------------------------------------------------------------ @@ -1329,6 +1462,11 @@ def _iteration( response = self._backend.complete(system_prompt, user_prompt) logger.debug("LLM raw response: %s", response.raw_text[:500]) + # Kept for journal telemetry if this iteration's action is discarded: + # the exact proposal the parser / validator rejected. + self._iteration_response_text = response.raw_text or "" + self._iteration_stop_reason = getattr(response, "stop_reason", None) + self._iteration_content_types = getattr(response, "content_types", None) # Track tokens pt = response.prompt_tokens or 0 @@ -1341,6 +1479,11 @@ def _iteration( f"(cumulative: ~{self._total_prompt_tokens + self._total_completion_tokens:,})" ) + # A failed request (auth, invalid request, network) is not a model + # answer: journal it as an API error, not as a rejected action. + if getattr(response, "api_error", False): + self._iteration_api_error = True + # Parse JSON from response if response.json_data is None: self._consecutive_parse_failures += 1 @@ -1350,14 +1493,23 @@ def _iteration( _MAX_CONSECUTIVE_PARSE_FAILURES, response.json_error, ) - self._print(f"[Iter {iteration}] Failed to parse LLM response as JSON") + truncated = getattr(response, "stop_reason", None) == "max_tokens" + self._print(f"[Iter {iteration}] Failed to parse LLM response as JSON" + + (f" (output hit max_tokens={self._config.llm.max_tokens})" if truncated else "")) if self._consecutive_parse_failures >= _MAX_CONSECUTIVE_PARSE_FAILURES: self._print(f"[Iter {iteration}] Too many consecutive parse failures — aborting") return "error", False - self._error_feedback = ( - "Failed to parse JSON from your response. " - "Please respond with a valid JSON object." - ) + if truncated: + self._error_feedback = ( + "Your previous response was cut off at the output-token limit before " + "a complete JSON object was produced. Respond with the JSON action " + "only, keeping the reasoning brief." + ) + else: + self._error_feedback = ( + "Failed to parse JSON from your response. " + "Please respond with a valid JSON object." + ) return "error", True self._consecutive_parse_failures = 0 @@ -1511,6 +1663,14 @@ def _handle_modify( f"{skipped_count} skipped" ) + # A modify that applied zero commands changes nothing — make the timeline + # say so instead of showing a blank "No description" card. + if applied_count == 0: + if not description or description == "No description": + description = "No commands applied (no change)" + else: + description = f"{description} — no commands applied" + if all_errors: self._error_feedback = "Command errors:\n" + "\n".join(all_errors) @@ -1626,7 +1786,7 @@ def _handle_modify( self._active_steering_directives[-1]["directive"] if self._active_steering_directives else None ) - self._journal.add_from_results( + _modify_entry = self._journal.add_from_results( iteration=iteration, description=description, commands=raw_commands, @@ -1640,6 +1800,7 @@ def _handle_modify( gencost=modified_net.gencost if self._config.search.application == "pflow" else None, exago_command=_single_call_record(sim_result), ) + _modify_entry.skipped_commands = list(all_errors) # Extract tracked metrics for multi-objective tracking if opflow is not None: @@ -2079,6 +2240,11 @@ def _resolve_candidate_buses(self, spec: dict) -> tuple[Optional[list[int]], str Returns (candidates, "") on success or (None, error_message) on failure. Shared by the feasibility sweep and the boundary sweep. """ + if not isinstance(spec, dict): + return None, ( + f"candidate_set must be an object with a 'type' field, " + f"got {type(spec).__name__}: {spec!r}" + ) ctype = spec.get("type", "") if ctype == "all_buses": candidates = [b.bus_i for b in self._base_network.buses] @@ -2275,6 +2441,8 @@ def _handle_sweep( candidate_set_spec = data.get("candidate_set", {}) mutation_template = data.get("mutation", {}) feasibility_spec = data.get("feasibility", {}) + if not isinstance(feasibility_spec, dict): + feasibility_spec = {} if not mutation_template or "action" not in mutation_template: self._error_feedback = "sweep requires a 'mutation' dict with an 'action' key." @@ -2712,6 +2880,8 @@ def _handle_boundary_sweep( pf_spec = data.get("power_factor", s.boundary_power_factor_default) candidate_set_spec = data.get("candidate_set", {}) feasibility_spec = data.get("feasibility", {}) + if not isinstance(feasibility_spec, dict): + feasibility_spec = {} vmin = feasibility_spec.get("Vmin", 0.9) vmax = feasibility_spec.get("Vmax", 1.1) q_frac = s.boundary_gen_q_frac @@ -3370,6 +3540,8 @@ def _handle_reserve_screen( description = data.get("description", "Hot reserve / N-1 generator security") reasoning = data.get("reasoning", "") feasibility_spec = data.get("feasibility", {}) + if not isinstance(feasibility_spec, dict): + feasibility_spec = {} vmin = feasibility_spec.get("Vmin", 0.9) vmax = feasibility_spec.get("Vmax", 1.1) minimize = bool(data.get("minimize", False)) @@ -4827,6 +4999,16 @@ def _assemble_prompt( def _finalize(self, session: SearchSession, elapsed_seconds: float) -> None: """Print summary and save journal.""" + _totals = getattr(self._backend, "totals", None) + if callable(_totals): + self._journal.llm_usage = { + **_totals(), + "backend": self._backend.name(), + "model": self._config.llm.model, + "max_tokens": self._config.llm.max_tokens, + "prompt_cache": bool(getattr(self._config.llm, "prompt_cache", False)), + } + total_tokens = self._total_prompt_tokens + self._total_completion_tokens if self._pending_explore is not None: # session ended right after an explore dirs, best = self._pending_explore @@ -4889,7 +5071,7 @@ def _finalize(self, session: SearchSession, elapsed_seconds: float) -> None: print(f" Application: {session.application}") print(f" Backend: {self._backend.name()} ({self._config.llm.model})") print( - f" Iterations: {stats['total_iterations']} " + f" Iterations: {format_iteration_count(stats)} " f"(of max {self._config.search.max_iterations})" ) print(f" Duration: {elapsed_seconds:.1f} seconds") @@ -5158,16 +5340,61 @@ def resume_from(self, save_dir: Path) -> SearchSession: # Continue the agent loop from last_iteration + 1 max_iter = self._config.search.max_iterations + try: + stall_limit = int(os.environ.get("AGENTIGRID_STALL_LIMIT", "0") or "0") + except ValueError: + stall_limit = 0 + stall = 0 for iteration in range(last_iteration + 1, max_iter + 1): if self._stop_requested: session.termination_reason = "user_stopped" self._print("\nSearch stopped by user.") break - action_type, should_continue = self._iteration(iteration, goal) + self._iteration_api_error = False + self._iteration_response_text = None + self._iteration_stop_reason = None + self._iteration_content_types = None + try: + action_type, should_continue = self._iteration(iteration, goal) + except Exception as e: + import traceback + self._print( + f"\n[Iter {iteration}] Action failed with an internal error " + f"({type(e).__name__}: {e}); discarding and continuing." + ) + traceback.print_exc() + self._error_feedback = ( + f"The previous action raised an internal error and was discarded: " + f"{type(e).__name__}: {e}. Respond with a well-formed action that " + f"strictly follows the schema." + ) + action_type, should_continue = "error", True + internal_error = True + else: + internal_error = False + + # Only emit when this iteration recorded a new entry (see run() above). + latest_entry = self._journal.latest + recorded = latest_entry is not None and latest_entry.iteration == iteration + if action_type == "error" and not recorded: + self._record_discarded(iteration, internal_error) if self._on_iteration: - latest_entry = self._journal.latest - if latest_entry: + if recorded: self._on_iteration(iteration, latest_entry, action_type, self._latest_opflow) + elif action_type == "error": + self._emit_discarded(iteration) + # Stall breaker: count consecutive no-progress iterations. + if recorded: + stall = 0 + elif stall_limit: + stall += 1 + if stall >= stall_limit: + session.termination_reason = "stalled_no_progress" + self._print( + f"\nStopping early: {stall} consecutive no-progress " + f"iterations (stall limit {stall_limit})." + ) + break if not should_continue: if not session.termination_reason: session.termination_reason = "completed" @@ -5179,14 +5406,8 @@ def resume_from(self, save_dir: Path) -> SearchSession: if not session.termination_reason: session.termination_reason = "completed" - # Final notification - if self._on_iteration: - latest_entry = self._journal.latest - if latest_entry: - self._on_iteration( - latest_entry.iteration, latest_entry, - session.termination_reason, self._latest_opflow, - ) + # No final on_iteration re-emit (see run() above) — it only duplicated + # the last timeline card. session.end_time = datetime.now().isoformat() session.total_prompt_tokens = self._total_prompt_tokens diff --git a/agentigrid/agentigrid/exago_engine/journal.py b/agentigrid/agentigrid/exago_engine/journal.py index f86d4bc5..0c7eea3f 100644 --- a/agentigrid/agentigrid/exago_engine/journal.py +++ b/agentigrid/agentigrid/exago_engine/journal.py @@ -55,6 +55,10 @@ class JournalEntry: contingency_meta: Optional[dict] = None # C.5: mutation, changed element, affected elements, order, pass/fail counts reserve_meta: Optional[dict] = None # C.8: hot-reserve / N-1 generator security accounting findings: Optional[dict] = None # complete: the LLM's full findings (summary, details, ...) + # Modify iterations: commands the LLM proposed that were NOT applied (parse + # errors or validator skips), as messages. [] = every proposed command was + # applied; None = not recorded (non-modify entries, older journals). + skipped_commands: Optional[list[str]] = None def describe_contingency_scope(meta: dict) -> str: @@ -217,6 +221,11 @@ def __init__(self) -> None: self.session_best: Optional[dict] = None # load_factor: session-level load scaling factor (updated by set_load_factor action). self.load_factor: Optional[float] = None + # discarded_actions: iterations whose LLM action recorded no entry -- + # rejected by the deterministic validator / parser ("rejected"), a failed + # API request ("api_error"), output cut at max_tokens ("truncated") or + # an internal error ("internal"). Exported with the journal. + self.discarded_actions: list[dict] = [] def update_session_best( self, @@ -1122,6 +1131,10 @@ def export_json(self, path: Path) -> None: data["session_best"] = self.session_best if self.load_factor is not None: data["load_factor"] = self.load_factor + if getattr(self, "llm_usage", None) is not None: + data["llm_usage"] = self.llm_usage + if getattr(self, "discarded_actions", None): + data["discarded_actions"] = self.discarded_actions path.write_text(json.dumps(data, indent=2), encoding="utf-8") logger.info("Journal exported to %s (%d entries)", path, len(self._entries)) @@ -1139,7 +1152,7 @@ def export_csv(self, path: Path) -> None: "mode", "elapsed_seconds", "timestamp", "steering_directive", "tracked_metrics", "feasibility_detail", "solver", "num_steps", "num_scenarios", "explored_variants", "candidate_count", "feasible_buses", "exago_command", - "contingency_meta", "reserve_meta", "findings", + "contingency_meta", "reserve_meta", "findings", "skipped_commands", ] with open(path, "w", newline="", encoding="utf-8") as f: @@ -1155,6 +1168,7 @@ def export_csv(self, path: Path) -> None: row["contingency_meta"] = json.dumps(row.get("contingency_meta") or None) row["reserve_meta"] = json.dumps(row.get("reserve_meta") or None) row["findings"] = json.dumps(row.get("findings") or None) + row["skipped_commands"] = json.dumps(row.get("skipped_commands")) writer.writerow(row) logger.info("Journal CSV exported to %s (%d entries)", path, len(self._entries)) @@ -1163,6 +1177,22 @@ def export_csv(self, path: Path) -> None: # Summary statistics # ------------------------------------------------------------------ + def _iteration_counts(self) -> tuple[int, bool]: + """Agent iterations actually run, and whether a base case was solved. + + ``total_iterations`` counts journal entries: it includes the base-case + solve (iteration 0) and leaves out iterations whose LLM output was + rejected (they are kept in ``discarded_actions``, not as entries). + This counts distinct iteration numbers >= 1 across both, which is the + number of agent iterations the user asked for and the run performed. + """ + numbers = {e.iteration for e in self._entries} + for d in getattr(self, "discarded_actions", None) or []: + it = d.get("iteration") if isinstance(d, dict) else None + if isinstance(it, int): + numbers.add(it) + return sum(1 for n in numbers if n >= 1), 0 in numbers + def summary_stats( self, best_iteration_override: int | None = None, @@ -1182,9 +1212,12 @@ def summary_stats( feasible_count, infeasible_count, objective_trend, voltage_range_trend, goal_type. """ + llm_iterations, has_base_case = self._iteration_counts() if not self._entries: return { "total_iterations": 0, + "llm_iterations": llm_iterations, + "has_base_case": has_base_case, "best_objective": None, "best_iteration": None, "best_bus": None, @@ -1242,6 +1275,8 @@ def summary_stats( return { "total_iterations": len(self._entries), + "llm_iterations": llm_iterations, + "has_base_case": has_base_case, "best_objective": best_objective, "best_iteration": best_iteration, "best_bus": best_bus, @@ -1296,3 +1331,15 @@ def _best_cost_in_sweeps(self) -> tuple[float, int, int | None] | None: if best is None or c < best[0]: best = (c, e.iteration, v.get("bus")) return best + + +def format_iteration_count(stats: dict) -> str: + """Display form of the iteration count, e.g. ``"2 + base case"``. + + Falls back to ``total_iterations`` for stats dicts written before + ``llm_iterations`` existed. + """ + if "llm_iterations" not in stats: + return str(stats.get("total_iterations", 0)) + n = stats["llm_iterations"] + return f"{n} + base case" if stats.get("has_base_case") else str(n) diff --git a/agentigrid/agentigrid/exago_engine/regenerate_report.py b/agentigrid/agentigrid/exago_engine/regenerate_report.py index a18c66fa..1d6dc088 100644 --- a/agentigrid/agentigrid/exago_engine/regenerate_report.py +++ b/agentigrid/agentigrid/exago_engine/regenerate_report.py @@ -110,6 +110,7 @@ def load_journal_export(path: Path) -> SearchJournal: exago_command=ed.get("exago_command"), contingency_meta=ed.get("contingency_meta"), reserve_meta=ed.get("reserve_meta"), + skipped_commands=ed.get("skipped_commands"), )) # Rebuild the objective registry, then overwrite its history with the saved diff --git a/agentigrid/agentigrid/exago_engine/session_io.py b/agentigrid/agentigrid/exago_engine/session_io.py index cfd65e4b..013ff615 100644 --- a/agentigrid/agentigrid/exago_engine/session_io.py +++ b/agentigrid/agentigrid/exago_engine/session_io.py @@ -185,6 +185,7 @@ def load_session(save_dir: Path) -> dict[str, Any]: contingency_meta=entry_data.get("contingency_meta"), reserve_meta=entry_data.get("reserve_meta"), findings=entry_data.get("findings"), + skipped_commands=entry_data.get("skipped_commands"), ) journal_entries.append(entry) diff --git a/agentigrid/claude_code_changes.md b/agentigrid/claude_code_changes.md index 83a815a8..3c0fed69 100644 --- a/agentigrid/claude_code_changes.md +++ b/agentigrid/claude_code_changes.md @@ -4,6 +4,40 @@ This document records significant changes made by Claude Code, grouped by the pr --- +## Agent loop and LLM backends: robust iterations, honest UI, usage accounting (2026-10-03) + +**Motivation.** Long runs with local models (Ollama) showed iterations that went wrong in ways the UI hid or misreported: a malformed action crashed the run; a rejected iteration re-emitted the previous card, so the timeline showed stale or out-of-order cards (e.g. "Iteration 6" after iteration 20); analysis and completion entries were labelled FAILED; a modify that applied no command showed "No description"; and the summary counted the base case as an iteration ("Iterations 3" for max 2). Separately, Anthropic runs re-sent the same system prompt at full price on every iteration, and some newer models reject the `temperature` parameter. + +### Agent loop (`agentigrid/exago_engine/agent_loop.py`) + +- Each iteration runs inside a guard: an exception in an action handler is discarded with feedback to the model, and the run continues. Handlers that read `feasibility` / `candidate_set` tolerate non-dict values. +- The UI callback fires only when the iteration recorded a journal entry. Otherwise a UI-only "discarded" card explains why ("LLM output rejected — … (no change applied)"); it is never added to the journal. The final re-emit of the last entry is gone (it duplicated the last card). Same in `resume`. +- `journal.discarded_actions` records every iteration that produced no entry: kind (`rejected`, `api_error`, `truncated`, `internal`), the feedback sent to the model, `stop_reason`, content block types and the raw response (capped at 4,000 characters). Entries and summary statistics are unchanged. +- A response cut at `max_tokens` gets its own message and feedback asking for the JSON action only. +- A modify that applies zero commands says so in its description; modify entries record `skipped_commands`. +- Optional stall breaker: `AGENTIGRID_STALL_LIMIT=N` ends the run after N consecutive iterations without a new entry (`termination_reason = "stalled_no_progress"`). Off by default. + +### Backends (`agentigrid/backends/`) + +- Anthropic prompt caching of the per-session system prompt (`llm.prompt_cache`, default on). Billing only; the model sees the same text. `prompt_tokens` stays the total input; cache writes and reads are reported separately. +- `UsageMeter` wraps the backend and counts every call (including post-search analysis); totals go to the journal as `llm_usage`. +- Models that reject `temperature` are retried once without it, and it is not sent again in that session. +- `LLMResponse` gains `api_error`, `stop_reason`, `content_types` and cache token fields; failed requests in all backends set `api_error`. +- `AGENTIGRID_MAX_TOKENS` overrides the output-token cap. + +### UI and reports + +- Iterations shown as "N + base case" (UI summary and live panel, CLI summary, PDF report). `summary_stats()` adds `llm_iterations` and `has_base_case`; `total_iterations` is unchanged. +- Timeline cards show FAILED only for solves that did not converge, not for analysis / complete / sweep / contingency / explore entries or discarded proposals. +- Sidebar checkbox for the stall breaker (unchecked by default). +- Default Anthropic model `claude-sonnet-4-20250514` (retired) → `claude-sonnet-4-6`. + +### Tests + +`tests/llm/test_prompt_cache.py` (new: caching, token breakdown, `UsageMeter`, temperature retry, API errors, max_tokens, env override), `tests/journal/test_iteration_count.py` (new), `tests/agent_loop/test_agent_loop.py` (+8: discarded telemetry, API errors, truncation, response cap, no stale UI cards, stall breaker off/on, internal errors). The 4 new agent-loop callback/stall/error tests fail on the previous agent loop. + +--- + ## C.2/C.3 corrections — metric gating, summary aggregation, predicate audit, sweep dedup (2026-06-23) Four contained fixes from the prompt-13/15/17 runs. No capability behavior changed (capacity numbers, dispatch results, metric/predicate definitions, and the C.1 binding identifier are untouched) — only how results are gated, summarized, audited, and deduplicated. diff --git a/agentigrid/configs/default_config.yaml.template b/agentigrid/configs/default_config.yaml.template index dd5dd092..23908c84 100644 --- a/agentigrid/configs/default_config.yaml.template +++ b/agentigrid/configs/default_config.yaml.template @@ -30,7 +30,7 @@ data: # LLM backend settings llm: backend: "anthropic" # openai | anthropic | ollama | ollama-cloud - model: "claude-sonnet-4-20250514" + model: "claude-sonnet-4-6" api_key_env: "ANTHROPIC_API_KEY" # env var name containing the API key # OpenAI-specific openai_base_url: null @@ -40,6 +40,7 @@ llm: # Generation parameters temperature: 0.3 max_tokens: 4096 + prompt_cache: true # Anthropic only: cache the system prompt (billing-only; outputs unchanged) # Search settings search: diff --git a/agentigrid/configs/local_config.yaml.template b/agentigrid/configs/local_config.yaml.template index 3f4d5764..9e5cf608 100644 --- a/agentigrid/configs/local_config.yaml.template +++ b/agentigrid/configs/local_config.yaml.template @@ -29,6 +29,7 @@ llm: # ollama_cloud_host: null # set for Ollama Cloud temperature: 0.3 max_tokens: 4096 + prompt_cache: true # Anthropic only: cache the system prompt (billing-only; outputs unchanged) # Search settings search: diff --git a/agentigrid/launcher/ARCHITECTURE.md b/agentigrid/launcher/ARCHITECTURE.md index 3a8a2d98..23263b9b 100644 --- a/agentigrid/launcher/ARCHITECTURE.md +++ b/agentigrid/launcher/ARCHITECTURE.md @@ -224,7 +224,7 @@ Always visible. Contains all search parameters. **Section: LLM Backend** - Backend selector: `anthropic` | `openai` | `ollama` | `ollama-cloud` -- Model name text input (pre-filled with defaults per backend: `claude-sonnet-4-20250514` for Anthropic, `gpt-4o` for OpenAI, etc.) +- Model name text input (pre-filled with defaults per backend: `claude-sonnet-4-6` for Anthropic, `gpt-4o` for OpenAI, etc.) - Temperature slider (0.0 – 1.0, default 0.3) **Section: Search Parameters** @@ -274,7 +274,7 @@ Shown while a search is running. Two-column layout: - Key metrics: voltage range, max line loading, gen/load totals - Mode used (fresh/accumulative) - Current iteration shows a live status indicator: - - Spinner with phase text: "Sending prompt to Claude (claude-sonnet-4-20250514)..." → "Applying 3 modifications..." → "Running OPFLOW simulation..." → "Parsing results..." + - Spinner with phase text: "Sending prompt to Claude (claude-sonnet-4-6)..." → "Applying 3 modifications..." → "Running OPFLOW simulation..." → "Parsing results..." **Right column (~35%)** — Live Charts: - **Convergence chart** (Plotly line chart): Objective value vs. iteration number diff --git a/agentigrid/launcher/app.py b/agentigrid/launcher/app.py index 13613f1f..2711e738 100644 --- a/agentigrid/launcher/app.py +++ b/agentigrid/launcher/app.py @@ -36,7 +36,11 @@ reserve_trajectory_chart, ) -from agentigrid.exago_engine.journal import describe_contingency_scope +from agentigrid.exago_engine.journal import ( + NON_SOLVE_STATUSES, + describe_contingency_scope, + format_iteration_count, +) from agentigrid.exago_parsers import parse_matpower, network_summary # ── Page Configuration ─────────────────────────────────────────────────────── @@ -395,6 +399,23 @@ def render_sidebar() -> dict: max_iterations = st.slider( "Max iterations", 1, 50, 20, disabled=disabled, ) + stall_stop = st.checkbox( + "Stop early if the model stalls", + value=False, + disabled=disabled, + help=( + "End the search after N consecutive no-progress iterations " + "(the model repeating an invalid action or applying no change). " + "Saves iterations when a model is stuck. Leave unchecked for " + "comparable runs that always reach Max iterations." + ), + ) + stall_limit = st.number_input( + "Stop after N no-progress iterations", + min_value=2, max_value=20, value=5, step=1, + disabled=disabled or not stall_stop, + ) + os.environ["AGENTIGRID_STALL_LIMIT"] = str(int(stall_limit)) if stall_stop else "0" mpi_disabled = disabled or application not in ("scopflow", "sopflow") mpi_np = st.number_input( "MPI processes", @@ -798,15 +819,23 @@ def render_live_monitor(): icon = _iteration_icon(entry) obj = entry.get("objective_value") is_pflow_live = st.session_state.get("current_application") == "pflow" - if is_pflow_live: - obj_str = "analysis" if obj is not None else "FAILED" + # An entry with no objective is only "FAILED" if it was a real solve + # that did not converge. Analysis/complete/sweep/contingency/explore + # markers and discarded proposals carry no cost — they are not failures. + cs = entry.get("convergence_status") or "" + _no_solve = cs in NON_SOLVE_STATUSES or entry.get("mode") == "discarded" + if obj is not None: + obj_str = "analysis" if is_pflow_live else f"${obj:,.2f}" + elif _no_solve: + obj_str = "" # no cost, but not a failure — omit the segment else: - obj_str = f"${obj:,.2f}" if obj is not None else "FAILED" + obj_str = "FAILED" # a solve that genuinely did not converge elapsed = entry.get("sim_elapsed") time_str = f"{elapsed:.1f}s" if elapsed is not None else "—" + _cost_seg = f" — {obj_str}" if obj_str else "" label = ( f"{icon} Iteration {entry['iteration']}: " - f"{entry['description']} — {obj_str} ({time_str})" + f"{entry['description']}{_cost_seg} ({time_str})" ) with st.expander(label): @@ -934,7 +963,9 @@ def render_live_monitor(): # Progress Stats st.markdown("---") - n_iters = len(st.session_state.iteration_log) + _its = {e.get("iteration") for e in st.session_state.iteration_log} + _n_agent = sum(1 for i in _its if isinstance(i, int) and i >= 1) + n_iters = f"{_n_agent} + base case" if 0 in _its else str(_n_agent) feasible_count = sum( 1 for e in st.session_state.iteration_log if e.get("feasible") ) @@ -1326,7 +1357,7 @@ def _render_overview_tab(session): mc1, mc2, mc3, mc4, mc5 = st.columns(5) mc1.metric("Application", f"{session.application}") - mc2.metric("Iterations", f"{stats['total_iterations']}") + mc2.metric("Iterations", format_iteration_count(stats)) mc3.metric("Duration", duration_str) mc4.metric("Termination", session.termination_reason) mc5.metric("Tokens", f"{total_tokens:,}" if total_tokens > 0 else "—") diff --git a/agentigrid/launcher/config_builder.py b/agentigrid/launcher/config_builder.py index 4e13f207..a410bb61 100644 --- a/agentigrid/launcher/config_builder.py +++ b/agentigrid/launcher/config_builder.py @@ -253,7 +253,7 @@ def load_example_goals() -> list[dict[str, str]]: # Default model names per backend DEFAULT_MODELS: dict[str, str] = { - "anthropic": "claude-sonnet-4-20250514", + "anthropic": "claude-sonnet-4-6", "openai": "gpt-4o", "ollama": "qwen2.5:7b", "ollama-cloud": "qwen2.5:7b", diff --git a/agentigrid/launcher/report_generator.py b/agentigrid/launcher/report_generator.py index 8719b60f..ed9dda1b 100644 --- a/agentigrid/launcher/report_generator.py +++ b/agentigrid/launcher/report_generator.py @@ -23,7 +23,11 @@ from reportlab.pdfbase.ttfonts import TTFont from agentigrid.exago_engine.agent_loop import SearchSession -from agentigrid.exago_engine.journal import describe_contingency_scope, is_solve_iteration +from agentigrid.exago_engine.journal import ( + describe_contingency_scope, + format_iteration_count, + is_solve_iteration, +) from agentigrid.exago_parsers.opflow_results import OPFLOWResult try: @@ -599,7 +603,7 @@ def _build_executive_summary( "is not deliverable for these losses (required reserve is mis-located)." ) res_lines.extend([ - f"Total iterations: {stats['total_iterations']}", + f"Iterations: {format_iteration_count(stats)}", f"Duration: {duration.total_seconds():.0f}s", f"Termination: {session.termination_reason}", (f"Token usage: {total_tokens:,}" if total_tokens > 0 else "Token usage: N/A"), @@ -645,7 +649,7 @@ def _build_executive_summary( relief_line += f"; {unresolved_count} unresolved." cont_lines.append(relief_line) cont_lines.extend([ - f"Total iterations: {stats['total_iterations']}", + f"Iterations: {format_iteration_count(stats)}", f"Duration: {duration.total_seconds():.0f}s", f"Termination: {session.termination_reason}", (f"Token usage: {total_tokens:,}" if total_tokens > 0 else "Token usage: N/A"), @@ -683,7 +687,7 @@ def _build_executive_summary( _headline = f"Sweep complete: {n_feas} of {n_cand} candidate buses are feasible." sweep_lines = [ _headline, - f"Total iterations: {stats['total_iterations']}", + f"Iterations: {format_iteration_count(stats)}", f"Duration: {duration.total_seconds():.0f}s", f"Termination: {session.termination_reason}", f"Token usage: {total_tokens:,}" if total_tokens > 0 else "Token usage: N/A", @@ -707,7 +711,7 @@ def _build_executive_summary( 1 for e in solve_entries if e.feasibility_detail == "marginal" ) lines = [ - f"Total iterations: {stats['total_iterations']}", + f"Iterations: {format_iteration_count(stats)}", f"Feasible solutions: {solve_feasible}", f"Infeasible: {solve_infeasible}", ] @@ -1520,8 +1524,9 @@ def _build_convergence_section( ) # Auto-generated text - n = stats["total_iterations"] - text = f"The search ran for {n} iterations." + n = stats.get("llm_iterations", stats["total_iterations"]) + text = (f"The search ran for {n} iteration(s) after the base case." + if stats.get("has_base_case") else f"The search ran for {n} iteration(s).") if stats["best_objective"] is not None: base_entry = session.journal.entries[0] if session.journal.entries else None if base_entry and base_entry.objective_value and base_entry.objective_value != 0: diff --git a/agentigrid/tests/agent_loop/test_agent_loop.py b/agentigrid/tests/agent_loop/test_agent_loop.py index c7df4a30..b6f80a57 100644 --- a/agentigrid/tests/agent_loop/test_agent_loop.py +++ b/agentigrid/tests/agent_loop/test_agent_loop.py @@ -254,6 +254,152 @@ def test_invalid_json_then_valid(self, tmp_path: Path): assert len(session.journal) == 3 # base + modify + complete assert session.termination_reason == "completed" + def test_discarded_iterations_are_journaled_as_telemetry(self, tmp_path: Path): + """An unparseable action and an unknown action record no entry, but are + kept in journal.discarded_actions (validator-rejection telemetry).""" + cfg = _make_config(tmp_path, max_iterations=10) + responses = [ + _objectives_response(), # objective parser + _make_llm_response(None, raw_text="This is not JSON"), # iter 1 + _make_llm_response({"action": "dance", "reasoning": "x"}), # iter 2 + _make_llm_response({"action": "complete", "reasoning": "Done", + "findings": {"summary": "ok"}}), # iter 3 + ] + mock_backend = MockBackend(responses) + sim_result = _make_sim_result(stdout=_sample_stdout(), success=True) + with patch("agentigrid.exago_engine.agent_loop.create_backend", return_value=mock_backend), \ + patch("agentigrid.exago_engine.agent_loop.SimulationExecutor") as mock_exec_cls: + mock_executor = MagicMock() + mock_executor.run.return_value = sim_result + mock_exec_cls.return_value = mock_executor + controller = AgentLoopController(cfg) + session = controller.run(BASE_CASE, "Test discard telemetry") + + assert len(session.journal) == 2 # base + complete + disc = session.journal.discarded_actions + assert [(d["iteration"], d["kind"]) for d in disc] == [(1, "rejected"), (2, "rejected")] + assert "parse JSON" in disc[0]["feedback"] and "Unknown action" in disc[1]["feedback"] + # the rejected proposal itself is kept (raw model output) + assert disc[0]["response"] == "This is not JSON" and disc[0]["response_truncated"] is False + assert '"dance"' in disc[1]["response"] or "dance" in disc[1]["response"] + + def test_api_errors_are_not_counted_as_rejections(self, tmp_path: Path): + cfg = _make_config(tmp_path, max_iterations=10) + api_err = LLMResponse(raw_text="Anthropic API error: 400", json_data=None, json_error="x", + model="m", backend="test", prompt_tokens=None, completion_tokens=None, + api_error=True) + responses = [_objectives_response(), api_err, + _make_llm_response({"action": "complete", "reasoning": "Done", + "findings": {"summary": "ok"}})] + mock_backend = MockBackend(responses) + sim_result = _make_sim_result(stdout=_sample_stdout(), success=True) + with patch("agentigrid.exago_engine.agent_loop.create_backend", return_value=mock_backend), \ + patch("agentigrid.exago_engine.agent_loop.SimulationExecutor") as mock_exec_cls: + mock_executor = MagicMock() + mock_executor.run.return_value = sim_result + mock_exec_cls.return_value = mock_executor + session = AgentLoopController(cfg).run(BASE_CASE, "Test api error") + disc = session.journal.discarded_actions + assert [(d["iteration"], d["kind"]) for d in disc] == [(1, "api_error")] + assert "response" not in disc[0] # no model response for a failed request + + def test_truncated_output_is_journaled_as_truncated(self, tmp_path: Path): + cfg = _make_config(tmp_path, max_iterations=10) + cut = LLMResponse(raw_text="", json_data=None, json_error="no JSON", model="m", + backend="test", prompt_tokens=10, completion_tokens=4096, + stop_reason="max_tokens", content_types=["thinking"]) + responses = [_objectives_response(), cut, + _make_llm_response({"action": "complete", "reasoning": "Done", + "findings": {"summary": "ok"}})] + mock_backend = MockBackend(responses) + sim_result = _make_sim_result(stdout=_sample_stdout(), success=True) + with patch("agentigrid.exago_engine.agent_loop.create_backend", return_value=mock_backend), \ + patch("agentigrid.exago_engine.agent_loop.SimulationExecutor") as mock_exec_cls: + mock_executor = MagicMock() + mock_executor.run.return_value = sim_result + mock_exec_cls.return_value = mock_executor + session = AgentLoopController(cfg).run(BASE_CASE, "Test truncation") + d = session.journal.discarded_actions[0] + assert (d["iteration"], d["kind"], d["stop_reason"]) == (1, "truncated", "max_tokens") + assert d["content_types"] == ["thinking"] and d["response"] == "" + assert "output-token limit" in d["feedback"] + + def test_long_discarded_response_is_capped(self, tmp_path: Path): + from agentigrid.exago_engine import agent_loop as al + cfg = _make_config(tmp_path, max_iterations=10) + long_text = "x" * (al.DISCARDED_RESPONSE_MAX + 50) + responses = [_objectives_response(), _make_llm_response(None, raw_text=long_text), + _make_llm_response({"action": "complete", "reasoning": "Done", + "findings": {"summary": "ok"}})] + mock_backend = MockBackend(responses) + sim_result = _make_sim_result(stdout=_sample_stdout(), success=True) + with patch("agentigrid.exago_engine.agent_loop.create_backend", return_value=mock_backend), \ + patch("agentigrid.exago_engine.agent_loop.SimulationExecutor") as mock_exec_cls: + mock_executor = MagicMock() + mock_executor.run.return_value = sim_result + mock_exec_cls.return_value = mock_executor + session = AgentLoopController(cfg).run(BASE_CASE, "Test cap") + d = session.journal.discarded_actions[0] + assert len(d["response"]) == al.DISCARDED_RESPONSE_MAX and d["response_truncated"] is True + + def _run(self, tmp_path, responses, goal="Test", max_iterations=10, **ctl_kw): + cfg = _make_config(tmp_path, max_iterations=max_iterations) + mock_backend = MockBackend([_objectives_response()] + responses) + sim_result = _make_sim_result(stdout=_sample_stdout(), success=True) + with patch("agentigrid.exago_engine.agent_loop.create_backend", return_value=mock_backend), \ + patch("agentigrid.exago_engine.agent_loop.SimulationExecutor") as mock_exec_cls: + mock_executor = MagicMock() + mock_executor.run.return_value = sim_result + mock_exec_cls.return_value = mock_executor + return AgentLoopController(cfg, **ctl_kw).run(BASE_CASE, goal) + + _COMPLETE = {"action": "complete", "reasoning": "Done", "findings": {"summary": "ok"}} + + def test_ui_callback_never_replays_a_stale_entry(self, tmp_path: Path): + """A rejected iteration produces a 'discarded' card for that iteration, + never a re-emit of the previous entry, and the run ends without a + final duplicate of the last card.""" + calls = [] + session = self._run( + tmp_path, + [_make_llm_response(None, raw_text="not JSON"), # iter 1: rejected + _make_llm_response({"action": "dance", "reasoning": "x"}), # iter 2: rejected + _make_llm_response(self._COMPLETE)], # iter 3 + on_iteration=lambda it, entry, kind, opf: calls.append((it, kind, entry.mode)), + ) + assert [c[0] for c in calls] == [0, 1, 2, 3] + assert [c[1] for c in calls[1:3]] == ["discarded", "discarded"] + assert all(c[2] == "discarded" for c in calls[1:3]) + assert session.termination_reason == "completed" + # the synthetic cards are UI-only: the journal holds base case + complete + assert len(session.journal) == 2 + + def test_stall_breaker_is_off_by_default(self, tmp_path: Path, monkeypatch): + monkeypatch.delenv("AGENTIGRID_STALL_LIMIT", raising=False) + bad = {"action": "dance", "reasoning": "x"} + session = self._run(tmp_path, [_make_llm_response(bad)] * 3 + [_make_llm_response(self._COMPLETE)]) + assert session.termination_reason == "completed" + assert len(session.journal.discarded_actions) == 3 + + def test_stall_breaker_stops_after_n_no_progress_iterations(self, tmp_path: Path, monkeypatch): + monkeypatch.setenv("AGENTIGRID_STALL_LIMIT", "2") + bad = {"action": "dance", "reasoning": "x"} + session = self._run(tmp_path, [_make_llm_response(bad)] * 3 + [_make_llm_response(self._COMPLETE)]) + assert session.termination_reason == "stalled_no_progress" + assert [d["iteration"] for d in session.journal.discarded_actions] == [1, 2] + + def test_internal_error_in_an_action_is_discarded_and_the_run_continues(self, tmp_path: Path): + with patch.object(AgentLoopController, "_handle_analyze", side_effect=RuntimeError("boom")): + session = self._run( + tmp_path, + [_make_llm_response({"action": "analyze", "query": "max loading", "reasoning": "x"}), + _make_llm_response(self._COMPLETE)], + ) + assert session.termination_reason == "completed" + d = session.journal.discarded_actions[0] + assert (d["iteration"], d["kind"]) == (1, "internal") + assert "RuntimeError: boom" in d["feedback"] + def test_unknown_action(self, tmp_path: Path): """LLM returns unknown action, then completes.""" cfg = _make_config(tmp_path, max_iterations=10) diff --git a/agentigrid/tests/e2e/test_e2e.py b/agentigrid/tests/e2e/test_e2e.py index 05ae5f1f..2244ff43 100644 --- a/agentigrid/tests/e2e/test_e2e.py +++ b/agentigrid/tests/e2e/test_e2e.py @@ -113,7 +113,7 @@ def _make_config( data=DataConfig(data_dir=DATA_DIR), llm=LLMConfig( backend="anthropic" if _has_anthropic_key else "openai", - model="claude-sonnet-4-20250514" if _has_anthropic_key else "gpt-4o-mini", + model="claude-sonnet-4-6" if _has_anthropic_key else "gpt-4o-mini", api_key_env="ANTHROPIC_API_KEY" if _has_anthropic_key else "OPENAI_API_KEY", openai_base_url=None, ollama_host="http://localhost:11434", diff --git a/agentigrid/tests/journal/test_iteration_count.py b/agentigrid/tests/journal/test_iteration_count.py new file mode 100644 index 00000000..a78b2161 --- /dev/null +++ b/agentigrid/tests/journal/test_iteration_count.py @@ -0,0 +1,51 @@ +"""Iteration count shown to users: agent iterations plus the base case.""" + +from __future__ import annotations + +from agentigrid.exago_engine.journal import SearchJournal, format_iteration_count + +from tests.journal.test_journal import _make_entry + + +def _journal(iterations, discarded=()): + j = SearchJournal() + for i in iterations: + j.add_entry(_make_entry(i)) + j.discarded_actions = [{"iteration": i, "kind": "rejected"} for i in discarded] + return j + + +def test_base_case_is_not_counted_as_an_agent_iteration(): + stats = _journal([0, 1, 2]).summary_stats() + assert stats["total_iterations"] == 3 # unchanged: journal entries + assert stats["llm_iterations"] == 2 + assert stats["has_base_case"] is True + assert format_iteration_count(stats) == "2 + base case" + + +def test_rejected_iterations_still_count(): + # max 4: iteration 3 was rejected (no journal entry, kept in discarded_actions) + stats = _journal([0, 1, 2, 4], discarded=[3]).summary_stats() + assert stats["total_iterations"] == 4 + assert stats["llm_iterations"] == 4 + assert format_iteration_count(stats) == "4 + base case" + + +def test_several_entries_for_one_iteration_count_once(): + stats = _journal([0, 1, 1, 2]).summary_stats() + assert stats["llm_iterations"] == 2 + + +def test_no_base_case(): + stats = _journal([1, 2]).summary_stats() + assert format_iteration_count(stats) == "2" + + +def test_old_stats_without_new_keys_fall_back(): + assert format_iteration_count({"total_iterations": 5}) == "5" + + +def test_empty_journal(): + stats = SearchJournal().summary_stats() + assert stats["llm_iterations"] == 0 and stats["has_base_case"] is False + assert format_iteration_count(stats) == "0" diff --git a/agentigrid/tests/llm/test_backends.py b/agentigrid/tests/llm/test_backends.py index 2111ecfe..97ed40b4 100644 --- a/agentigrid/tests/llm/test_backends.py +++ b/agentigrid/tests/llm/test_backends.py @@ -233,7 +233,7 @@ class TestAnthropicLive: def test_simple_prompt(self): cfg = _dummy_config( backend="anthropic", - model="claude-sonnet-4-20250514", + model="claude-sonnet-4-6", api_key_env="ANTHROPIC_API_KEY", ) backend = create_backend(cfg) diff --git a/agentigrid/tests/llm/test_prompt_cache.py b/agentigrid/tests/llm/test_prompt_cache.py new file mode 100644 index 00000000..81799492 --- /dev/null +++ b/agentigrid/tests/llm/test_prompt_cache.py @@ -0,0 +1,229 @@ +"""Prompt caching (Anthropic) and per-run token metering. + +Caching must be billing-only: the model sees the same system text either way, +``prompt_tokens`` keeps meaning TOTAL input tokens, and the cache breakdown is +reported separately. The UsageMeter must count every call without altering it. +""" + +from __future__ import annotations + +import os +from types import SimpleNamespace +from unittest.mock import patch + +import pytest + +from agentigrid.backends import LLMBackend, LLMResponse, UsageMeter +from agentigrid.config import LLMConfig, load_config + +try: + import anthropic as _anthropic # noqa: F401 + _has_anthropic = True +except ImportError: + _has_anthropic = False + + +def _cfg(**over) -> LLMConfig: + base = dict( + backend="anthropic", model="claude-sonnet-4-6", api_key_env="TEST_API_KEY", + openai_base_url=None, ollama_host="http://localhost:11434", + ollama_cloud_host=None, temperature=0.3, max_tokens=256, + ) + base.update(over) + return LLMConfig(**base) + + +def _fake_message(input_tokens, creation, read, output_tokens=7, text='{"action":"complete"}'): + usage = SimpleNamespace( + input_tokens=input_tokens, + cache_creation_input_tokens=creation, + cache_read_input_tokens=read, + output_tokens=output_tokens, + ) + return SimpleNamespace(content=[SimpleNamespace(type="text", text=text)], usage=usage) + + +# --------------------------------------------------------------------------- +# Config +# --------------------------------------------------------------------------- + +def test_prompt_cache_defaults_on(): + cfg = load_config(None) + assert cfg.llm.prompt_cache is True + + +def test_prompt_cache_can_be_disabled_by_override(): + cfg = load_config(None, cli_overrides={"llm.prompt_cache": False}) + assert cfg.llm.prompt_cache is False + + +# --------------------------------------------------------------------------- +# Anthropic backend request shape + token accounting +# --------------------------------------------------------------------------- + +@pytest.mark.skipif(not _has_anthropic, reason="anthropic package not installed") +@patch.dict(os.environ, {"TEST_API_KEY": "sk-test"}) +def test_system_prompt_sent_as_cached_block(): + from agentigrid.backends.anthropic_backend import AnthropicBackend + + backend = AnthropicBackend(_cfg(prompt_cache=True)) + with patch.object(backend._client.messages, "create", + return_value=_fake_message(50, 4000, 0)) as create: + backend.complete("SYSTEM TEXT", "user text") + system = create.call_args.kwargs["system"] + assert system == [{"type": "text", "text": "SYSTEM TEXT", + "cache_control": {"type": "ephemeral"}}] + # the user turn is untouched + assert create.call_args.kwargs["messages"] == [{"role": "user", "content": "user text"}] + + +@pytest.mark.skipif(not _has_anthropic, reason="anthropic package not installed") +@patch.dict(os.environ, {"TEST_API_KEY": "sk-test"}) +def test_cache_off_sends_plain_string(): + from agentigrid.backends.anthropic_backend import AnthropicBackend + + backend = AnthropicBackend(_cfg(prompt_cache=False)) + with patch.object(backend._client.messages, "create", + return_value=_fake_message(4050, None, None)) as create: + resp = backend.complete("SYSTEM TEXT", "user text") + assert create.call_args.kwargs["system"] == "SYSTEM TEXT" + assert resp.prompt_tokens == 4050 + assert resp.cache_creation_tokens is None and resp.cache_read_tokens is None + + +@pytest.mark.skipif(not _has_anthropic, reason="anthropic package not installed") +@patch.dict(os.environ, {"TEST_API_KEY": "sk-test"}) +def test_prompt_tokens_is_total_input_with_breakdown(): + from agentigrid.backends.anthropic_backend import AnthropicBackend + + backend = AnthropicBackend(_cfg(prompt_cache=True)) + # first call writes the cache, second reads it + with patch.object(backend._client.messages, "create", + side_effect=[_fake_message(50, 4000, 0), _fake_message(60, 0, 4000)]): + first = backend.complete("S", "u1") + second = backend.complete("S", "u2") + assert (first.prompt_tokens, first.cache_creation_tokens, first.cache_read_tokens) == (4050, 4000, 0) + assert (second.prompt_tokens, second.cache_creation_tokens, second.cache_read_tokens) == (4060, 0, 4000) + assert first.json_data == {"action": "complete"} + + +# --------------------------------------------------------------------------- +# UsageMeter +# --------------------------------------------------------------------------- + +class _Stub(LLMBackend): + def __init__(self, responses): + self._responses = list(responses) + self.seen = [] + + def complete(self, system_prompt, user_prompt, temperature=None): + self.seen.append((system_prompt, user_prompt, temperature)) + return self._responses.pop(0) + + def name(self): + return "stub" + + def supports_json_mode(self): + return False + + def custom(self): + return "delegated" + + +def _resp(pt, ct, cw=None, cr=None): + return LLMResponse(raw_text="x", json_data=None, json_error=None, model="m", + backend="stub", prompt_tokens=pt, completion_tokens=ct, + cache_creation_tokens=cw, cache_read_tokens=cr) + + +def test_usage_meter_counts_all_calls_and_is_transparent(): + inner = _Stub([_resp(4050, 10, 4000, 0), _resp(4060, 12, 0, 4000), _resp(None, None)]) + meter = UsageMeter(inner) + r1 = meter.complete("S", "u1") + meter.complete("S", "u2", temperature=0.0) + meter.complete("S", "u3") + assert r1 is not None and r1.prompt_tokens == 4050 + assert inner.seen[1] == ("S", "u2", 0.0) # args passed through verbatim + assert meter.totals() == { + "calls": 3, "prompt_tokens": 8110, "completion_tokens": 22, + "cache_creation_tokens": 4000, "cache_read_tokens": 4000, + "api_errors": 0, "truncated": 0, "temperature_sent": None, + } + assert meter.name() == "stub" + assert meter.custom() == "delegated" # backend-specific attrs delegate + + +def test_journal_exports_llm_usage(tmp_path): + import json + from agentigrid.exago_engine.journal import SearchJournal + + j = SearchJournal() + j.llm_usage = {"calls": 2, "prompt_tokens": 100, "completion_tokens": 5, + "cache_creation_tokens": 80, "cache_read_tokens": 0} + out = tmp_path / "j.json" + j.export_json(out) + assert json.loads(out.read_text())["llm_usage"]["cache_creation_tokens"] == 80 + + +def test_temperature_rejected_retries_without_it_and_remembers(): + from agentigrid.backends.anthropic_backend import AnthropicBackend + backend = AnthropicBackend(_cfg(model="claude-sonnet-5")) + calls = [] + + def create(**kw): + calls.append(kw) + if "temperature" in kw: + raise RuntimeError("Error code: 400 - `temperature` is deprecated for this model.") + return _fake_message(10, 0, 0) + + with patch.object(backend._client.messages, "create", side_effect=create): + r1 = backend.complete("sys", "u1") + r2 = backend.complete("sys", "u2") + assert not r1.api_error and r1.json_data == {"action": "complete"} + assert ["temperature" in c for c in calls] == [True, False, False] # retried once, then omitted + assert backend.temperature_sent is False and r2.prompt_tokens == 10 + + +@pytest.mark.skipif(not _has_anthropic, reason="anthropic not installed") +@patch.dict(os.environ, {"TEST_API_KEY": "sk-test"}) +def test_other_api_errors_are_flagged_and_counted(): + from agentigrid.backends.anthropic_backend import AnthropicBackend + backend = AnthropicBackend(_cfg()) + meter = UsageMeter(backend) + with patch.object(backend._client.messages, "create", side_effect=RuntimeError("401 invalid x-api-key")): + r = meter.complete("sys", "u") + assert r.api_error and r.json_data is None + t = meter.totals() + assert t["calls"] == 1 and t["api_errors"] == 1 and t["prompt_tokens"] == 0 + with patch.object(backend._client.messages, "create", return_value=_fake_message(5, 0, 0)): + meter.complete("sys", "u") + assert meter.totals()["temperature_sent"] is True + + +# --------------------------------------------------------------------------- +# Output-token cap: stop_reason telemetry and AGENTIGRID_MAX_TOKENS +# --------------------------------------------------------------------------- + +@pytest.mark.skipif(not _has_anthropic, reason="anthropic not installed") +@patch.dict(os.environ, {"TEST_API_KEY": "sk-test"}) +def test_max_tokens_stop_is_reported_and_counted(): + from agentigrid.backends.anthropic_backend import AnthropicBackend + backend = AnthropicBackend(_cfg()) + meter = UsageMeter(backend) + msg = _fake_message(10, 0, 0, output_tokens=256, text="") + msg.content = [SimpleNamespace(type="thinking", thinking="...")] + msg.stop_reason = "max_tokens" + with patch.object(backend._client.messages, "create", return_value=msg): + r = meter.complete("sys", "u") + assert r.stop_reason == "max_tokens" and r.content_types == ["thinking"] + assert r.raw_text == "" and r.json_data is None and not r.api_error + assert meter.totals()["truncated"] == 1 + + +def test_max_tokens_env_override(monkeypatch): + monkeypatch.setenv("AGENTIGRID_MAX_TOKENS", "16000") + assert load_config(None).llm.max_tokens == 16000 + monkeypatch.setenv("AGENTIGRID_MAX_TOKENS", "lots") + assert load_config(None).llm.max_tokens == 4096 # ignored, default kept + monkeypatch.delenv("AGENTIGRID_MAX_TOKENS") + assert load_config(None).llm.max_tokens == 4096