From 3988e7bb0cdcead7c16a51c0866ff5fe5116559d Mon Sep 17 00:00:00 2001 From: AkhileshNegi Date: Tue, 8 Sep 2026 23:45:56 +0530 Subject: [PATCH 1/4] fix(evaluation): cap prompt-improvement trace payload MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A prompt-improvement run dumped the whole score-trace file into the drafting request, so a large judged run overflowed the provider's 1M input-token ceiling and failed the job with an HTTP 400: prompt_generation_failed: Anthropic returned HTTP 400 prompt is too long: 1121367 tokens > 1000000 maximum Each trace is now projected down to the fields the brief actually names, long text fields are clamped, and rows are filled to a character budget worst-scoring-first — the brief tells the model to focus on the failing rows, so those are the ones a cap has to keep. The brief states when rows were dropped and the count is logged. Co-Authored-By: Claude Opus 5 (1M context) --- .../evaluations/prompt_improvement.py | 108 +++++++++++++++++- .../tests/api/routes/test_improve_prompt.py | 86 ++++++++++++++ docs/wiki/modules/evaluations.md | 1 + 3 files changed, 193 insertions(+), 2 deletions(-) diff --git a/backend/app/services/evaluations/prompt_improvement.py b/backend/app/services/evaluations/prompt_improvement.py index bf4519ed9..873331c16 100644 --- a/backend/app/services/evaluations/prompt_improvement.py +++ b/backend/app/services/evaluations/prompt_improvement.py @@ -25,6 +25,7 @@ from app.crud.config.version import ConfigVersionCrud from app.crud.evaluations.core import get_evaluation_run_by_id from app.crud.evaluations.score import ( + COSINE_SCORE_NAME, GROUND_TRUTH_SCORE_NAME, KNOWLEDGE_BASE_SCORE_NAME, PROMPT_SCORE_NAME, @@ -50,6 +51,18 @@ # Headroom for a full prompt rewrite + JSON wrapper; too low truncates into invalid JSON. _LLM_MAX_TOKENS = 16384 +# Trace-payload budget, in characters rather than tokens: there is no tokenizer +# in-tree, and Indic-script Q&A tokenizes several times worse than English, so any +# chars-per-token divisor is a tuning knob, not a fact. Deliberately far below the +# provider's 1M-token input ceiling — a rewrite brief does not read better with more +# rows, and a near-ceiling Opus call per run is not worth paying for. +_TRACE_PAYLOAD_MAX_CHARS = 400_000 +_TRACE_FIELD_MAX_CHARS = 2_000 +_TRUNCATION_MARKER = " …[truncated]" + +# Sorts unscoreable rows (value = "N/A") last, behind every real score. +_UNSCOREABLE_SORT_VALUE = float("inf") + # JSON keys expected in the LLM's structured response. _LLM_KEY_INSTRUCTIONS = "improved_instructions" _LLM_KEY_RATIONALE = "rationale" @@ -630,6 +643,87 @@ def _call_prompt_drafting_llm(*, user_message_text: str) -> tuple[str, str]: ) from exc +def _truncate(text: object) -> str: + """Clamp one trace field; long judge rationales are the main source of bloat.""" + value = "" if text is None else str(text) + if len(value) <= _TRACE_FIELD_MAX_CHARS: + return value + return value[:_TRACE_FIELD_MAX_CHARS] + _TRUNCATION_MARKER + + +def _project_trace(trace: dict, *, is_judge_run: bool) -> dict: + """Keep only the fields the brief tells the model to read. + + `trace_id` / `question_id` are referenced nowhere in the prompt, and the v1 + brief never mentions the judge `comment`, so both are dropped. + """ + projected: dict = { + "question": _truncate(trace.get("question")), + "ground_truth_answer": _truncate(trace.get("ground_truth_answer")), + "llm_answer": _truncate(trace.get("llm_answer")), + "scores": [ + { + "name": score.get("name"), + "value": score.get("value"), + **({"unscoreable": True} if score.get("unscoreable") else {}), + **( + {"comment": _truncate(score.get("comment"))} if is_judge_run else {} + ), + } + for score in trace.get("scores") or [] + ], + } + if not is_judge_run and trace.get("category"): + projected["category"] = trace["category"] + return projected + + +def _primary_score(trace: dict, *, score_name: str) -> float: + """The metric the brief calls the primary signal; +inf when it is unusable.""" + for score in trace.get("scores") or []: + if score.get("name") != score_name or score.get("unscoreable"): + continue + value = score.get("value") + if isinstance(value, bool) or not isinstance(value, (int, float)): + continue + return float(value) + return _UNSCOREABLE_SORT_VALUE + + +def _select_traces( + traces: list | dict, *, is_judge_run: bool +) -> tuple[list[dict], int]: + """Project, truncate, and cap the traces to a char budget, worst-scoring first. + + Returns (kept, total). Worst-first both selects and orders: the brief tells the + model to focus on the low-scoring rows, so those are the ones a cap must keep. + """ + rows = traces if isinstance(traces, list) else traces.get("traces") or [] + score_name = GROUND_TRUTH_SCORE_NAME if is_judge_run else COSINE_SCORE_NAME + ranked = sorted( + (row for row in rows if isinstance(row, dict)), + key=lambda row: _primary_score(row, score_name=score_name), + ) + + kept: list[dict] = [] + used = 0 + for row in ranked: + projected = _project_trace(row, is_judge_run=is_judge_run) + size = len(json.dumps(projected, ensure_ascii=False)) + if kept and used + size > _TRACE_PAYLOAD_MAX_CHARS: + break + kept.append(projected) + used += size + + if len(kept) < len(rows): + logger.warning( + f"[_select_traces] Trimmed traces to fit the prompt budget | " + f"kept={len(kept)} total={len(rows)} chars={used} " + f"is_judge_run={is_judge_run}" + ) + return kept, len(rows) + + def _draft_improved_prompt( *, current_instructions: str, @@ -714,11 +808,21 @@ def _draft_improved_prompt( f"sentence (≤ {_RATIONALE_MAX_LENGTH} characters): what you changed and why." ) target_config = _target_config_from_params(config_params) + selected_traces, total_traces = _select_traces(traces, is_judge_run=is_judge_run) + trace_scope = ( + "" + if len(selected_traces) == total_traces + else ( + f" You are seeing the {len(selected_traces)} lowest-scoring traces of " + f"{total_traces}; the rest scored higher and are omitted." + ) + ) user_message_text = ( "You are a prompt engineer. Below is a JSON array of evaluation traces" - f"{trace_description}\n\n" - f"## Evaluation traces\n```\n{json.dumps(traces)}\n```\n\n" + f"{trace_description}{trace_scope}\n\n" + "## Evaluation traces\n```\n" + f"{json.dumps(selected_traces, ensure_ascii=False)}\n```\n\n" f"## Current system prompt\n```\n{current_instructions}\n```\n\n" "## Target configuration (read-only — do NOT change any of these)\n" f"```\n{json.dumps(target_config)}\n```\n\n" diff --git a/backend/app/tests/api/routes/test_improve_prompt.py b/backend/app/tests/api/routes/test_improve_prompt.py index 1f9b91723..c6b0499b1 100644 --- a/backend/app/tests/api/routes/test_improve_prompt.py +++ b/backend/app/tests/api/routes/test_improve_prompt.py @@ -42,16 +42,20 @@ from app.core.config import settings from app.crud.config.version import ConfigVersionCrud +from app.crud.evaluations.score import GROUND_TRUTH_SCORE_NAME from app.crud.jobs import JobCrud from app.models import EvaluationDataset, EvaluationRun from app.models.config.config import ConfigTag from app.models.job import Job, JobStatus, JobType from app.services.evaluations.prompt_improvement import ( + _UNSCOREABLE_SORT_VALUE, _UPSTREAM_ERROR_DETAIL_MAX_LENGTH, AI_GENERATED_MARKER, COMMIT_MESSAGE_MAX_LENGTH, _anthropic_error_detail, _call_prompt_drafting_llm, + _draft_improved_prompt, + _primary_score, execute_prompt_improvement, start_prompt_improvement_job, validate_improve_prompt, @@ -1039,3 +1043,85 @@ def test_missing_platform_key_is_flagged_as_misconfiguration( assert "ANTHROPIC_API_KEY" in str(raised.value) assert "anthropic_response" not in str(raised.value) + + +class TestTraceBudget: + """Oversized runs must be trimmed to fit the model's input window, keeping the + rows the brief actually asks the model to fix.""" + + @staticmethod + def _judge_trace(question_id: int, ground_truth_value: float | str) -> dict: + unscoreable = ground_truth_value == "N/A" + return { + "trace_id": f"t{question_id}", + "question_id": question_id, + "question": f"Q{question_id} " + "\u0915" * 3000, + "ground_truth_answer": "A" * 5000, + "llm_answer": "B" * 5000, + "scores": [ + { + "name": GROUND_TRUTH_SCORE_NAME, + "value": ground_truth_value, + "data_type": "NUMERIC", + "comment": "C" * 8000, + **({"unscoreable": True} if unscoreable else {}), + } + ], + } + + @staticmethod + def _capture_brief(traces: list) -> str: + captured: dict[str, str] = {} + + def fake_call(*, user_message_text: str) -> tuple[str, str]: + captured["text"] = user_message_text + return "improved", "why" + + with patch(f"{_SERVICE}._call_prompt_drafting_llm", side_effect=fake_call): + assert _draft_improved_prompt( + current_instructions="be helpful", + config_params={"model": "gpt-4o"}, + traces=traces, + is_judge_run=True, + ) == ("improved", "why") + return captured["text"] + + def test_oversized_traces_are_capped_worst_first(self) -> None: + # The uniquely BEST row is first in input order and the uniquely WORST is + # last, so an unsorted (or reversed) selection fails both assertions. + # Untrimmed this payload is ~1.3M chars. + traces = [ + self._judge_trace(1000, 5.0), + *(self._judge_trace(i, 3.0) for i in range(1, 60)), + self._judge_trace(999, "N/A"), + self._judge_trace(0, 0.0), + ] + + text = self._capture_brief(traces) + + assert len(text) < 450_000 + assert "Q0 " in text + assert "Q1000 " not in text + # Fields the brief never names are not shipped. + assert "trace_id" not in text + assert f"lowest-scoring traces of {len(traces)}" in text + + def test_small_run_is_sent_whole_without_a_scope_note(self) -> None: + text = self._capture_brief([self._judge_trace(1, 5.0)]) + + assert "lowest-scoring traces of" not in text + + def test_unscoreable_score_sorts_behind_every_real_score(self) -> None: + """`value` is the string "N/A" on unscoreable rows; mixed str/float must not + raise, and those rows must rank last.""" + unscoreable = self._judge_trace(1, "N/A") + best = self._judge_trace(2, 5.0) + + assert _primary_score( + unscoreable, score_name=GROUND_TRUTH_SCORE_NAME + ) > _primary_score(best, score_name=GROUND_TRUTH_SCORE_NAME) + # A missing metric is treated the same way. + assert ( + _primary_score({"scores": []}, score_name=GROUND_TRUTH_SCORE_NAME) + == _UNSCOREABLE_SORT_VALUE + ) diff --git a/docs/wiki/modules/evaluations.md b/docs/wiki/modules/evaluations.md index 3e6f72c1d..c6cb1ffd9 100644 --- a/docs/wiki/modules/evaluations.md +++ b/docs/wiki/modules/evaluations.md @@ -40,6 +40,7 @@ v2 judge field on `EvaluationRun`: `is_judge_run` (bool marker gating native jud - Fast text runs (v1 cosine + v2 judge) fan out `ceil(total_items / EVAL_FAST_CHUNK_SIZE)` `run_evaluation_fast_chunk` tasks (responses only), then a cron barrier (`dispatch_fast_evaluation_barriers`) enqueues one `run_evaluation_fast_aggregate` once every chunk has a `raw_output_url`. The aggregate merges chunks, then (v2) judges **every** row in that single task — so the judge pool is sized by its own `EVAL_JUDGE_CONCURRENCY` (not the response stage's `EVAL_FAST_API_CONCURRENCY`) to clear the max dataset (`EVAL_FAST_MAX_UNIQUE_ROWS` × `duplication_factor`) under the aggregate's `CELERY_TASK_SOFT_TIME_LIMIT`. No judge fan-out / second barrier. - Prompt improvement is job-based with callback delivery: `POST /evaluations/{id}/improve-prompt` validates preconditions, enqueues a `Job` (`JobType.PROMPT_IMPROVEMENT`, `models/job.py`) run by Celery task `run_prompt_improvement` (`celery/tasks/job_execution.py`), and returns `202` with an `LLMJobImmediatePublic` handle. On finish the worker POSTs a single best-effort callback to the caller-supplied `callback_url` (SSRF-guarded via `validate_callback_url`): an `APIResponse[PromptImprovementJobPublic]` (`models/evaluation.py`) carrying the new `ConfigVersion` on success or `error_message` on failure (for an Anthropic fault `error_message` carries the provider's own response body, appended by `_anthropic_error_detail` in `services/evaluations/prompt_improvement.py`). The `ConfigVersion` is persisted regardless of callback outcome. Celery redelivery of a `SUCCESS` job re-sends the callback without re-running the LLM. - v2 prompt iteration reuses the same job/Celery/config-version machinery (`services/evaluations/prompt_improvement.py`) and branches on `run.is_judge_run`: the v2 route calls `start_prompt_improvement_job(..., require_judge_run=True)` (a non-judge run → `422 not_a_judge_run`), and the worker drafts from the three-metric judge trace (`_draft_improved_prompt(is_judge_run=True)`, using each metric's score + reasoning) and delivers a `PromptRecommendationJobPublic` callback carrying `recommendation_type` (`Literal["prompt"]`, `models/evaluation.py`; widens to a union when knowledge-base / model recommendations land). Non-judge (v1) runs keep the default `_draft_improved_prompt` brief + `PromptImprovementJobPublic` path unchanged. +- Both prompt-improvement briefs cap the trace payload before the Anthropic call (`_select_traces` in `services/evaluations/prompt_improvement.py`): each trace is projected down to the fields the brief names, long text fields (notably the judge `comment`) are clamped, and rows are then filled to a character budget worst-scoring-first — a large run used to exceed the provider's 1M-token input limit and fail the job with `prompt_generation_failed: ... HTTP 400`. When rows are dropped the brief says so, and the count is logged. `crud/evaluations/summary.py` still sends every trace whole (its own `ponytail:` note) and has the same latent ceiling. - Evaluation iteration loop (`POST /evaluations/iterations`) chains fast-eval + v2 prompt improvement into a self-driving cycle via LangGraph (`services/evaluations/iteration_graph.py`): `start_eval_node` → `wait_eval_node` → (conditional) `start_improve_node` → `wait_improve_node` → loops back to `start_eval_node`, or → `finalize_node`. Stop-score = mean(`Adherence to Ground Truth`, `Adherence to Prompt`) from `EvaluationRun.score["summary_scores"]` (`compute_round_scores`); `Adherence to Knowledge Base` is recorded per round for visibility only, never gates stopping. Stops on 3 consecutive rounds with ` Date: Tue, 8 Sep 2026 23:52:40 +0530 Subject: [PATCH 2/4] fix(evaluation): degrade repeats before rows when the brief is too large MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The character cap alone bounds the payload by dropping whole rows, which loses questions the model never gets to see. A judged run repeats every question `duplication_factor` times, and the 5th repeat of a question carries far less signal than a question that is missing entirely. The brief is now measured with `messages.count_tokens` — same model and output schema as the real call, since both are billed as input — and degraded until it fits: first one repeat per question at a time, from the run's own duplication factor down to a floor of 3 (below that the repeats no longer show whether the judge is stable), and only then whole rows, worst-scoring-first against a budget rescaled from the measured count. Which repeats survive is decided by file order, not score: repeats exist to measure the judge's spread on one question, so dropping the worst of them would erase the signal the judge brief is told to read. A trace with no `question_id` is keyed uniquely and never grouped, so older datasets uploaded without ids are not collapsed into a single group. Counting is a network call from a Celery worker, so a failure there is non-fatal: the flat character cap applies on every attempt regardless. Co-Authored-By: Claude Opus 5 (1M context) --- .../evaluations/prompt_improvement.py | 223 +++++++++++++++--- .../tests/api/routes/test_improve_prompt.py | 61 ++++- docs/wiki/modules/evaluations.md | 2 +- 3 files changed, 250 insertions(+), 36 deletions(-) diff --git a/backend/app/services/evaluations/prompt_improvement.py b/backend/app/services/evaluations/prompt_improvement.py index 873331c16..30b172c48 100644 --- a/backend/app/services/evaluations/prompt_improvement.py +++ b/backend/app/services/evaluations/prompt_improvement.py @@ -8,6 +8,7 @@ import copy import json import logging +from collections import Counter from uuid import UUID import anthropic @@ -51,15 +52,29 @@ # Headroom for a full prompt rewrite + JSON wrapper; too low truncates into invalid JSON. _LLM_MAX_TOKENS = 16384 -# Trace-payload budget, in characters rather than tokens: there is no tokenizer -# in-tree, and Indic-script Q&A tokenizes several times worse than English, so any -# chars-per-token divisor is a tuning knob, not a fact. Deliberately far below the -# provider's 1M-token input ceiling — a rewrite brief does not read better with more -# rows, and a near-ceiling Opus call per run is not worth paying for. +# Input-token ceiling for the drafting request, measured with `count_tokens` +# rather than estimated: the model's window is 1M and Indic-script Q&A tokenizes +# several times worse than English, so any chars-per-token divisor guesses wrong +# in the direction that fails the job. The gap to 1M leaves room for the schema +# and the rewrite itself. +_INPUT_TOKEN_BUDGET = 900_000 +# Shrink factor applied when the measured count has to be converted back into a +# character budget; the relationship is only approximately linear. +_TOKEN_BUDGET_SAFETY = 0.9 + +# Coarse character budget, applied on every attempt. It is what bounds the payload +# when `count_tokens` is unavailable, and it keeps a near-ceiling Opus call from +# being the normal case — a rewrite brief does not read better with more rows. _TRACE_PAYLOAD_MAX_CHARS = 400_000 _TRACE_FIELD_MAX_CHARS = 2_000 _TRUNCATION_MARKER = " …[truncated]" +# Repeats of the same question are the first thing dropped when the brief is too +# large: cutting the 5th, then the 4th, keeps every question represented, whereas +# dropping rows loses questions outright. Below this floor the repeats no longer +# show whether the judge is stable, so row-dropping takes over instead. +_MIN_REPEATS_PER_QUESTION = 3 + # Sorts unscoreable rows (value = "N/A") last, behind every real score. _UNSCOREABLE_SORT_VALUE = float("inf") @@ -678,6 +693,50 @@ def _project_trace(trace: dict, *, is_judge_run: bool) -> dict: return projected +def _group_key(index: int, trace: dict) -> object: + """Repeats of one question share its `question_id`. + + A trace without one is keyed uniquely so it is never treated as a repeat — + older datasets uploaded without question ids would otherwise collapse into a + single group and be cut down to a handful of rows. + """ + question_id = trace.get("question_id") + if question_id or question_id == 0: + return question_id + return f"__row_{index}" + + +def _max_repeats_observed(rows: list[dict]) -> int: + counts = Counter(_group_key(index, row) for index, row in enumerate(rows)) + return max(counts.values(), default=0) + + +def _repeat_ladder(rows: list[dict]) -> list[int]: + """Repeat limits to try, from the run's own duplication factor down to the floor.""" + observed = _max_repeats_observed(rows) + if observed <= _MIN_REPEATS_PER_QUESTION: + return [observed] + return list(range(observed, _MIN_REPEATS_PER_QUESTION - 1, -1)) + + +def _limit_repeats(rows: list[dict], *, max_repeats: int) -> list[dict]: + """Keep at most `max_repeats` traces per question, in file order. + + Order, not score, decides which repeats survive: repeats exist to show how + stable the judge is on one question, so dropping the worst of them would erase + the very spread the judge brief is told to read. + """ + kept: list[dict] = [] + seen: dict[object, int] = {} + for index, row in enumerate(rows): + key = _group_key(index, row) + if seen.get(key, 0) >= max_repeats: + continue + seen[key] = seen.get(key, 0) + 1 + kept.append(row) + return kept + + def _primary_score(trace: dict, *, score_name: str) -> float: """The metric the brief calls the primary signal; +inf when it is unusable.""" for score in trace.get("scores") or []: @@ -691,18 +750,25 @@ def _primary_score(trace: dict, *, score_name: str) -> float: def _select_traces( - traces: list | dict, *, is_judge_run: bool + rows: list[dict], + *, + is_judge_run: bool, + max_repeats: int | None = None, + max_chars: int = _TRACE_PAYLOAD_MAX_CHARS, ) -> tuple[list[dict], int]: - """Project, truncate, and cap the traces to a char budget, worst-scoring first. + """Project, truncate, and cap the traces, worst-scoring first. - Returns (kept, total). Worst-first both selects and orders: the brief tells the - model to focus on the low-scoring rows, so those are the ones a cap must keep. + Returns (kept, total) where total counts the whole run, so the brief can say + how much it is not seeing. Worst-first both selects and orders: the brief tells + the model to focus on the low-scoring rows, so those are the ones a cap must + keep. """ - rows = traces if isinstance(traces, list) else traces.get("traces") or [] score_name = GROUND_TRUTH_SCORE_NAME if is_judge_run else COSINE_SCORE_NAME + candidates = ( + rows if max_repeats is None else _limit_repeats(rows, max_repeats=max_repeats) + ) ranked = sorted( - (row for row in rows if isinstance(row, dict)), - key=lambda row: _primary_score(row, score_name=score_name), + candidates, key=lambda row: _primary_score(row, score_name=score_name) ) kept: list[dict] = [] @@ -710,7 +776,7 @@ def _select_traces( for row in ranked: projected = _project_trace(row, is_judge_run=is_judge_run) size = len(json.dumps(projected, ensure_ascii=False)) - if kept and used + size > _TRACE_PAYLOAD_MAX_CHARS: + if kept and used + size > max_chars: break kept.append(projected) used += size @@ -719,11 +785,70 @@ def _select_traces( logger.warning( f"[_select_traces] Trimmed traces to fit the prompt budget | " f"kept={len(kept)} total={len(rows)} chars={used} " + f"max_repeats={max_repeats} max_chars={max_chars} " f"is_judge_run={is_judge_run}" ) return kept, len(rows) +def _count_input_tokens(*, user_message_text: str) -> int | None: + """Exact input-token count for the drafting request, or None when unavailable. + + Counted against the same model and output schema the real call uses, since both + are billed as input. A failure here must not fail the job — callers fall back to + the character budget, which bounds the payload on its own. + """ + if not settings.ANTHROPIC_API_KEY: + return None + try: + client = ClaudeProvider.create_client({"api_key": settings.ANTHROPIC_API_KEY}) + counted = client.messages.count_tokens( + model=settings.PROMPT_IMPROVEMENT_MODEL, + messages=[{"role": "user", "content": user_message_text}], + output_config={"format": {"type": "json_schema", "schema": _OUTPUT_SCHEMA}}, + ) + # A malformed / stubbed response must not crash the worker; the character + # budget already bounds the payload without a count. + return counted.input_tokens if isinstance(counted.input_tokens, int) else None + except Exception as exc: + logger.warning( + f"[_count_input_tokens] Count unavailable, falling back to the " + f"character budget | {exc}" + ) + return None + + +def _build_user_message( + *, + trace_description: str, + task_steps: list[str], + current_instructions: str, + target_config: dict, + selected_traces: list[dict], + total_traces: int, +) -> str: + """Assemble the drafting brief; rebuilt once per attempt as the payload shrinks.""" + trace_scope = ( + "" + if len(selected_traces) == total_traces + else ( + f" You are seeing {len(selected_traces)} of {total_traces} traces — " + "the lowest-scoring rows, and at most a few repeats per question; the " + "rest are omitted." + ) + ) + return ( + "You are a prompt engineer. Below is a JSON array of evaluation traces" + f"{trace_description}{trace_scope}\n\n" + "## Evaluation traces\n```\n" + f"{json.dumps(selected_traces, ensure_ascii=False)}\n```\n\n" + f"## Current system prompt\n```\n{current_instructions}\n```\n\n" + "## Target configuration (read-only — do NOT change any of these)\n" + f"```\n{json.dumps(target_config)}\n```\n\n" + "## Task\n" + "".join(task_steps) + ) + + def _draft_improved_prompt( *, current_instructions: str, @@ -808,25 +933,61 @@ def _draft_improved_prompt( f"sentence (≤ {_RATIONALE_MAX_LENGTH} characters): what you changed and why." ) target_config = _target_config_from_params(config_params) - selected_traces, total_traces = _select_traces(traces, is_judge_run=is_judge_run) - trace_scope = ( - "" - if len(selected_traces) == total_traces - else ( - f" You are seeing the {len(selected_traces)} lowest-scoring traces of " - f"{total_traces}; the rest scored higher and are omitted." + raw_rows = traces if isinstance(traces, list) else traces.get("traces") or [] + rows = [row for row in raw_rows if isinstance(row, dict)] + + def brief(selected: list[dict], total: int) -> str: + return _build_user_message( + trace_description=trace_description, + task_steps=task_steps, + current_instructions=current_instructions, + target_config=target_config, + selected_traces=selected, + total_traces=total, ) - ) - user_message_text = ( - "You are a prompt engineer. Below is a JSON array of evaluation traces" - f"{trace_description}{trace_scope}\n\n" - "## Evaluation traces\n```\n" - f"{json.dumps(selected_traces, ensure_ascii=False)}\n```\n\n" - f"## Current system prompt\n```\n{current_instructions}\n```\n\n" - "## Target configuration (read-only — do NOT change any of these)\n" - f"```\n{json.dumps(target_config)}\n```\n\n" - "## Task\n" + "".join(task_steps) - ) + # Degrade repeats before rows: the 5th repeat of a question carries far less + # signal than a question the model never sees at all. + user_message_text = "" + counted: int | None = None + for max_repeats in _repeat_ladder(rows): + selected_traces, total_traces = _select_traces( + rows, is_judge_run=is_judge_run, max_repeats=max_repeats + ) + user_message_text = brief(selected_traces, total_traces) + counted = _count_input_tokens(user_message_text=user_message_text) + if counted is None or counted <= _INPUT_TOKEN_BUDGET: + return _call_prompt_drafting_llm(user_message_text=user_message_text) + logger.warning( + f"[_draft_improved_prompt] Over the input budget, dropping a repeat | " + f"max_repeats={max_repeats} tokens={counted} " + f"budget={_INPUT_TOKEN_BUDGET}" + ) + + # The repeat floor is not a get-out: at 3 repeats a long enough run is still + # over, so stop cutting repeats and cut rows instead, converting the measured + # count back into a character budget. + if counted: + scaled_chars = max( + _TRACE_FIELD_MAX_CHARS, + int( + _TRACE_PAYLOAD_MAX_CHARS + * _TOKEN_BUDGET_SAFETY + * _INPUT_TOKEN_BUDGET + / counted + ), + ) + logger.warning( + f"[_draft_improved_prompt] Repeat floor reached, dropping rows | " + f"max_repeats={_MIN_REPEATS_PER_QUESTION} tokens={counted} " + f"scaled_chars={scaled_chars}" + ) + selected_traces, total_traces = _select_traces( + rows, + is_judge_run=is_judge_run, + max_repeats=_MIN_REPEATS_PER_QUESTION, + max_chars=scaled_chars, + ) + user_message_text = brief(selected_traces, total_traces) return _call_prompt_drafting_llm(user_message_text=user_message_text) diff --git a/backend/app/tests/api/routes/test_improve_prompt.py b/backend/app/tests/api/routes/test_improve_prompt.py index c6b0499b1..385881dc6 100644 --- a/backend/app/tests/api/routes/test_improve_prompt.py +++ b/backend/app/tests/api/routes/test_improve_prompt.py @@ -48,6 +48,7 @@ from app.models.config.config import ConfigTag from app.models.job import Job, JobStatus, JobType from app.services.evaluations.prompt_improvement import ( + _MIN_REPEATS_PER_QUESTION, _UNSCOREABLE_SORT_VALUE, _UPSTREAM_ERROR_DETAIL_MAX_LENGTH, AI_GENERATED_MARKER, @@ -113,6 +114,9 @@ def _make_fake_claude_client(text_content: str | None = None) -> MagicMock: client = MagicMock() client.messages.create.return_value = response + # The drafting path counts input tokens before it calls; a real int keeps the + # repeat-degradation ladder on its first rung for these fixtures. + client.messages.count_tokens.return_value = MagicMock(input_tokens=1_000) return client @@ -1070,14 +1074,30 @@ def _judge_trace(question_id: int, ground_truth_value: float | str) -> dict: } @staticmethod - def _capture_brief(traces: list) -> str: + def _capture_brief(traces: list, *, token_counts: list[int] | None = None) -> str: + """Return the brief the drafting call would have received. + + ``token_counts`` feeds one measured input-token count per ladder rung; the + last value repeats once the list is exhausted. Under the budget on the first + rung by default, so the ladder stays put unless a test says otherwise. + """ + counts = list(token_counts or [1_000]) captured: dict[str, str] = {} + def fake_count(*, user_message_text: str) -> int: + return counts.pop(0) if len(counts) > 1 else counts[0] + def fake_call(*, user_message_text: str) -> tuple[str, str]: captured["text"] = user_message_text return "improved", "why" - with patch(f"{_SERVICE}._call_prompt_drafting_llm", side_effect=fake_call): + with ExitStack() as stack: + stack.enter_context( + patch(f"{_SERVICE}._count_input_tokens", side_effect=fake_count) + ) + stack.enter_context( + patch(f"{_SERVICE}._call_prompt_drafting_llm", side_effect=fake_call) + ) assert _draft_improved_prompt( current_instructions="be helpful", config_params={"model": "gpt-4o"}, @@ -1104,12 +1124,45 @@ def test_oversized_traces_are_capped_worst_first(self) -> None: assert "Q1000 " not in text # Fields the brief never names are not shipped. assert "trace_id" not in text - assert f"lowest-scoring traces of {len(traces)}" in text + assert f"of {len(traces)} traces" in text def test_small_run_is_sent_whole_without_a_scope_note(self) -> None: text = self._capture_brief([self._judge_trace(1, 5.0)]) - assert "lowest-scoring traces of" not in text + assert " traces —" not in text + + @staticmethod + def _repeated_run(questions: int, repeats: int) -> list[dict]: + """A judged run of `questions` questions scored `repeats` times each.""" + return [ + TestTraceBudget._judge_trace(question_id, 2.0) + for question_id in range(questions) + for _ in range(repeats) + ] + + def test_repeats_are_dropped_one_at_a_time_until_the_brief_fits(self) -> None: + traces = self._repeated_run(questions=4, repeats=5) + + # Over budget at 5 and 4 repeats per question, under it at 3. + text = self._capture_brief(traces, token_counts=[1_100_000, 950_000, 800_000]) + + # Every question is still represented, three times each — no question was + # dropped to make room. + for question_id in range(4): + assert text.count(f"Q{question_id} ") == 3 + + def test_row_dropping_takes_over_at_the_repeat_floor(self) -> None: + # Sized so the flat character budget alone would keep every row: only the + # floor's rescaled budget can drop any, which is what this asserts. + questions = 9 + traces = self._repeated_run(questions=questions, repeats=5) + + # Still far over budget after the ladder bottoms out at 3 repeats. + text = self._capture_brief(traces, token_counts=[2_800_000]) + + kept = sum(text.count(f"Q{question_id} ") for question_id in range(questions)) + assert 0 < kept < questions * _MIN_REPEATS_PER_QUESTION + assert f"of {len(traces)} traces" in text def test_unscoreable_score_sorts_behind_every_real_score(self) -> None: """`value` is the string "N/A" on unscoreable rows; mixed str/float must not diff --git a/docs/wiki/modules/evaluations.md b/docs/wiki/modules/evaluations.md index c6cb1ffd9..10522983b 100644 --- a/docs/wiki/modules/evaluations.md +++ b/docs/wiki/modules/evaluations.md @@ -40,7 +40,7 @@ v2 judge field on `EvaluationRun`: `is_judge_run` (bool marker gating native jud - Fast text runs (v1 cosine + v2 judge) fan out `ceil(total_items / EVAL_FAST_CHUNK_SIZE)` `run_evaluation_fast_chunk` tasks (responses only), then a cron barrier (`dispatch_fast_evaluation_barriers`) enqueues one `run_evaluation_fast_aggregate` once every chunk has a `raw_output_url`. The aggregate merges chunks, then (v2) judges **every** row in that single task — so the judge pool is sized by its own `EVAL_JUDGE_CONCURRENCY` (not the response stage's `EVAL_FAST_API_CONCURRENCY`) to clear the max dataset (`EVAL_FAST_MAX_UNIQUE_ROWS` × `duplication_factor`) under the aggregate's `CELERY_TASK_SOFT_TIME_LIMIT`. No judge fan-out / second barrier. - Prompt improvement is job-based with callback delivery: `POST /evaluations/{id}/improve-prompt` validates preconditions, enqueues a `Job` (`JobType.PROMPT_IMPROVEMENT`, `models/job.py`) run by Celery task `run_prompt_improvement` (`celery/tasks/job_execution.py`), and returns `202` with an `LLMJobImmediatePublic` handle. On finish the worker POSTs a single best-effort callback to the caller-supplied `callback_url` (SSRF-guarded via `validate_callback_url`): an `APIResponse[PromptImprovementJobPublic]` (`models/evaluation.py`) carrying the new `ConfigVersion` on success or `error_message` on failure (for an Anthropic fault `error_message` carries the provider's own response body, appended by `_anthropic_error_detail` in `services/evaluations/prompt_improvement.py`). The `ConfigVersion` is persisted regardless of callback outcome. Celery redelivery of a `SUCCESS` job re-sends the callback without re-running the LLM. - v2 prompt iteration reuses the same job/Celery/config-version machinery (`services/evaluations/prompt_improvement.py`) and branches on `run.is_judge_run`: the v2 route calls `start_prompt_improvement_job(..., require_judge_run=True)` (a non-judge run → `422 not_a_judge_run`), and the worker drafts from the three-metric judge trace (`_draft_improved_prompt(is_judge_run=True)`, using each metric's score + reasoning) and delivers a `PromptRecommendationJobPublic` callback carrying `recommendation_type` (`Literal["prompt"]`, `models/evaluation.py`; widens to a union when knowledge-base / model recommendations land). Non-judge (v1) runs keep the default `_draft_improved_prompt` brief + `PromptImprovementJobPublic` path unchanged. -- Both prompt-improvement briefs cap the trace payload before the Anthropic call (`_select_traces` in `services/evaluations/prompt_improvement.py`): each trace is projected down to the fields the brief names, long text fields (notably the judge `comment`) are clamped, and rows are then filled to a character budget worst-scoring-first — a large run used to exceed the provider's 1M-token input limit and fail the job with `prompt_generation_failed: ... HTTP 400`. When rows are dropped the brief says so, and the count is logged. `crud/evaluations/summary.py` still sends every trace whole (its own `ponytail:` note) and has the same latent ceiling. +- Both prompt-improvement briefs are size-bounded before the Anthropic call (`services/evaluations/prompt_improvement.py`) — a large run used to exceed the provider's 1M-token input limit and fail the job with `prompt_generation_failed: ... HTTP 400`. Every trace is projected down to the fields the brief names and its long text fields (notably the judge `comment`) are clamped; then the brief is measured with `messages.count_tokens` (`_count_input_tokens`, same model + output schema as the real call) and degraded until it fits `_INPUT_TOKEN_BUDGET`, in two stages: first `_repeat_ladder` drops one repeat per question at a time, from the run's duplication factor down to `_MIN_REPEATS_PER_QUESTION` (3) — every question stays represented; only below that floor does `_select_traces` start dropping whole rows, worst-scoring-first against a budget rescaled from the measured count. A `count_tokens` failure is non-fatal: the flat `_TRACE_PAYLOAD_MAX_CHARS` cap applies on every attempt regardless. The brief states how many traces it is not seeing, and each degradation step is logged. `crud/evaluations/summary.py` still sends every trace whole (its own `ponytail:` note) and has the same latent ceiling. - Evaluation iteration loop (`POST /evaluations/iterations`) chains fast-eval + v2 prompt improvement into a self-driving cycle via LangGraph (`services/evaluations/iteration_graph.py`): `start_eval_node` → `wait_eval_node` → (conditional) `start_improve_node` → `wait_improve_node` → loops back to `start_eval_node`, or → `finalize_node`. Stop-score = mean(`Adherence to Ground Truth`, `Adherence to Prompt`) from `EvaluationRun.score["summary_scores"]` (`compute_round_scores`); `Adherence to Knowledge Base` is recorded per round for visibility only, never gates stopping. Stops on 3 consecutive rounds with ` Date: Tue, 8 Sep 2026 23:55:58 +0530 Subject: [PATCH 3/4] fix(evaluation): skip token counting for a brief that cannot overflow MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Counting is a network call from a worker with a 300s soft time limit, and the ladder can make up to four of them. A brief under 200K characters cannot reach the 900K-token budget even at a pessimistic three tokens per character, so counting it is a wasted round trip — and that is where most runs land, leaving the path at its previous single call. Also covers the case the repeat ladder cannot help with: v1 traces from datasets uploaded without question ids are keyed uniquely and never grouped, so the rescaled character budget is the only thing that can bring an oversized brief down. Co-Authored-By: Claude Opus 5 (1M context) --- .../evaluations/prompt_improvement.py | 16 +++++++++--- .../tests/api/routes/test_improve_prompt.py | 25 +++++++++++++++++++ docs/wiki/modules/evaluations.md | 2 +- 3 files changed, 38 insertions(+), 5 deletions(-) diff --git a/backend/app/services/evaluations/prompt_improvement.py b/backend/app/services/evaluations/prompt_improvement.py index 30b172c48..cfb58fbca 100644 --- a/backend/app/services/evaluations/prompt_improvement.py +++ b/backend/app/services/evaluations/prompt_improvement.py @@ -63,9 +63,14 @@ _TOKEN_BUDGET_SAFETY = 0.9 # Coarse character budget, applied on every attempt. It is what bounds the payload -# when `count_tokens` is unavailable, and it keeps a near-ceiling Opus call from -# being the normal case — a rewrite brief does not read better with more rows. +# when `count_tokens` is unavailable or skipped, and it is what keeps the ordinary +# run cheap — the token budget above is only the hard never-400 ceiling, not a +# spend target. _TRACE_PAYLOAD_MAX_CHARS = 400_000 +# Below this the brief cannot breach the token budget even at a pessimistic three +# tokens per character, so counting it would just be a wasted round trip inside a +# worker with a soft time limit. Most runs land here. +_COUNT_TOKENS_ABOVE_CHARS = 200_000 _TRACE_FIELD_MAX_CHARS = 2_000 _TRUNCATION_MARKER = " …[truncated]" @@ -795,11 +800,14 @@ def _count_input_tokens(*, user_message_text: str) -> int | None: """Exact input-token count for the drafting request, or None when unavailable. Counted against the same model and output schema the real call uses, since both - are billed as input. A failure here must not fail the job — callers fall back to - the character budget, which bounds the payload on its own. + are billed as input. Returns None — meaning "treat as fitting" — for a brief too + small to be at risk, and for any failure: this is a network call from a Celery + worker, and the character budget bounds the payload on its own. """ if not settings.ANTHROPIC_API_KEY: return None + if len(user_message_text) <= _COUNT_TOKENS_ABOVE_CHARS: + return None try: client = ClaudeProvider.create_client({"api_key": settings.ANTHROPIC_API_KEY}) counted = client.messages.count_tokens( diff --git a/backend/app/tests/api/routes/test_improve_prompt.py b/backend/app/tests/api/routes/test_improve_prompt.py index 385881dc6..615580a57 100644 --- a/backend/app/tests/api/routes/test_improve_prompt.py +++ b/backend/app/tests/api/routes/test_improve_prompt.py @@ -55,6 +55,7 @@ COMMIT_MESSAGE_MAX_LENGTH, _anthropic_error_detail, _call_prompt_drafting_llm, + _count_input_tokens, _draft_improved_prompt, _primary_score, execute_prompt_improvement, @@ -1164,6 +1165,30 @@ def test_row_dropping_takes_over_at_the_repeat_floor(self) -> None: assert 0 < kept < questions * _MIN_REPEATS_PER_QUESTION assert f"of {len(traces)} traces" in text + def test_a_small_brief_is_never_sent_for_counting( + self, anthropic_creds: None + ) -> None: + """No round trip for a brief that cannot breach the budget.""" + with patch(f"{_SERVICE}.ClaudeProvider.create_client") as create_client: + assert _count_input_tokens(user_message_text="rewrite this") is None + create_client.assert_not_called() + + def test_traces_without_question_ids_still_get_trimmed(self) -> None: + """v1 traces from older datasets can carry no `question_id`. + + Those must never be grouped as repeats of one another, so the repeat ladder + is a no-op for them and the rescaled character budget is the only thing that + can bring an oversized brief down. + """ + traces = [self._judge_trace(i, 2.0) for i in range(20)] + for trace in traces: + del trace["question_id"] + + text = self._capture_brief(traces, token_counts=[2_800_000]) + + kept = sum(text.count(f"Q{i} ") for i in range(20)) + assert 0 < kept < 20 + def test_unscoreable_score_sorts_behind_every_real_score(self) -> None: """`value` is the string "N/A" on unscoreable rows; mixed str/float must not raise, and those rows must rank last.""" diff --git a/docs/wiki/modules/evaluations.md b/docs/wiki/modules/evaluations.md index 10522983b..50bf2a1aa 100644 --- a/docs/wiki/modules/evaluations.md +++ b/docs/wiki/modules/evaluations.md @@ -40,7 +40,7 @@ v2 judge field on `EvaluationRun`: `is_judge_run` (bool marker gating native jud - Fast text runs (v1 cosine + v2 judge) fan out `ceil(total_items / EVAL_FAST_CHUNK_SIZE)` `run_evaluation_fast_chunk` tasks (responses only), then a cron barrier (`dispatch_fast_evaluation_barriers`) enqueues one `run_evaluation_fast_aggregate` once every chunk has a `raw_output_url`. The aggregate merges chunks, then (v2) judges **every** row in that single task — so the judge pool is sized by its own `EVAL_JUDGE_CONCURRENCY` (not the response stage's `EVAL_FAST_API_CONCURRENCY`) to clear the max dataset (`EVAL_FAST_MAX_UNIQUE_ROWS` × `duplication_factor`) under the aggregate's `CELERY_TASK_SOFT_TIME_LIMIT`. No judge fan-out / second barrier. - Prompt improvement is job-based with callback delivery: `POST /evaluations/{id}/improve-prompt` validates preconditions, enqueues a `Job` (`JobType.PROMPT_IMPROVEMENT`, `models/job.py`) run by Celery task `run_prompt_improvement` (`celery/tasks/job_execution.py`), and returns `202` with an `LLMJobImmediatePublic` handle. On finish the worker POSTs a single best-effort callback to the caller-supplied `callback_url` (SSRF-guarded via `validate_callback_url`): an `APIResponse[PromptImprovementJobPublic]` (`models/evaluation.py`) carrying the new `ConfigVersion` on success or `error_message` on failure (for an Anthropic fault `error_message` carries the provider's own response body, appended by `_anthropic_error_detail` in `services/evaluations/prompt_improvement.py`). The `ConfigVersion` is persisted regardless of callback outcome. Celery redelivery of a `SUCCESS` job re-sends the callback without re-running the LLM. - v2 prompt iteration reuses the same job/Celery/config-version machinery (`services/evaluations/prompt_improvement.py`) and branches on `run.is_judge_run`: the v2 route calls `start_prompt_improvement_job(..., require_judge_run=True)` (a non-judge run → `422 not_a_judge_run`), and the worker drafts from the three-metric judge trace (`_draft_improved_prompt(is_judge_run=True)`, using each metric's score + reasoning) and delivers a `PromptRecommendationJobPublic` callback carrying `recommendation_type` (`Literal["prompt"]`, `models/evaluation.py`; widens to a union when knowledge-base / model recommendations land). Non-judge (v1) runs keep the default `_draft_improved_prompt` brief + `PromptImprovementJobPublic` path unchanged. -- Both prompt-improvement briefs are size-bounded before the Anthropic call (`services/evaluations/prompt_improvement.py`) — a large run used to exceed the provider's 1M-token input limit and fail the job with `prompt_generation_failed: ... HTTP 400`. Every trace is projected down to the fields the brief names and its long text fields (notably the judge `comment`) are clamped; then the brief is measured with `messages.count_tokens` (`_count_input_tokens`, same model + output schema as the real call) and degraded until it fits `_INPUT_TOKEN_BUDGET`, in two stages: first `_repeat_ladder` drops one repeat per question at a time, from the run's duplication factor down to `_MIN_REPEATS_PER_QUESTION` (3) — every question stays represented; only below that floor does `_select_traces` start dropping whole rows, worst-scoring-first against a budget rescaled from the measured count. A `count_tokens` failure is non-fatal: the flat `_TRACE_PAYLOAD_MAX_CHARS` cap applies on every attempt regardless. The brief states how many traces it is not seeing, and each degradation step is logged. `crud/evaluations/summary.py` still sends every trace whole (its own `ponytail:` note) and has the same latent ceiling. +- Both prompt-improvement briefs are size-bounded before the Anthropic call (`services/evaluations/prompt_improvement.py`) — a large run used to exceed the provider's 1M-token input limit and fail the job with `prompt_generation_failed: ... HTTP 400`. Every trace is projected down to the fields the brief names and its long text fields (notably the judge `comment`) are clamped; then the brief is measured with `messages.count_tokens` (`_count_input_tokens`, same model + output schema as the real call) and degraded until it fits `_INPUT_TOKEN_BUDGET`, in two stages: first `_repeat_ladder` drops one repeat per question at a time, from the run's duplication factor down to `_MIN_REPEATS_PER_QUESTION` (3) — every question stays represented; only below that floor does `_select_traces` start dropping whole rows, worst-scoring-first against a budget rescaled from the measured count. Counting is skipped entirely for a brief under `_COUNT_TOKENS_ABOVE_CHARS` (it cannot breach the budget even at a pessimistic three tokens per character), and a `count_tokens` failure is non-fatal: the flat `_TRACE_PAYLOAD_MAX_CHARS` cap applies on every attempt regardless. The brief states how many traces it is not seeing, and each degradation step is logged. `crud/evaluations/summary.py` still sends every trace whole (its own `ponytail:` note) and has the same latent ceiling. - Evaluation iteration loop (`POST /evaluations/iterations`) chains fast-eval + v2 prompt improvement into a self-driving cycle via LangGraph (`services/evaluations/iteration_graph.py`): `start_eval_node` → `wait_eval_node` → (conditional) `start_improve_node` → `wait_improve_node` → loops back to `start_eval_node`, or → `finalize_node`. Stop-score = mean(`Adherence to Ground Truth`, `Adherence to Prompt`) from `EvaluationRun.score["summary_scores"]` (`compute_round_scores`); `Adherence to Knowledge Base` is recorded per round for visibility only, never gates stopping. Stops on 3 consecutive rounds with ` Date: Wed, 9 Sep 2026 13:03:41 +0530 Subject: [PATCH 4/4] cleanups --- .../evaluations/prompt_improvement.py | 255 ++++-------------- .../tests/api/routes/test_improve_prompt.py | 87 ++---- docs/wiki/modules/evaluations.md | 2 +- 3 files changed, 82 insertions(+), 262 deletions(-) diff --git a/backend/app/services/evaluations/prompt_improvement.py b/backend/app/services/evaluations/prompt_improvement.py index cfb58fbca..e351c7a03 100644 --- a/backend/app/services/evaluations/prompt_improvement.py +++ b/backend/app/services/evaluations/prompt_improvement.py @@ -52,32 +52,16 @@ # Headroom for a full prompt rewrite + JSON wrapper; too low truncates into invalid JSON. _LLM_MAX_TOKENS = 16384 -# Input-token ceiling for the drafting request, measured with `count_tokens` -# rather than estimated: the model's window is 1M and Indic-script Q&A tokenizes -# several times worse than English, so any chars-per-token divisor guesses wrong -# in the direction that fails the job. The gap to 1M leaves room for the schema -# and the rewrite itself. -_INPUT_TOKEN_BUDGET = 900_000 -# Shrink factor applied when the measured count has to be converted back into a -# character budget; the relationship is only approximately linear. -_TOKEN_BUDGET_SAFETY = 0.9 - -# Coarse character budget, applied on every attempt. It is what bounds the payload -# when `count_tokens` is unavailable or skipped, and it is what keeps the ordinary -# run cheap — the token budget above is only the hard never-400 ceiling, not a -# spend target. -_TRACE_PAYLOAD_MAX_CHARS = 400_000 -# Below this the brief cannot breach the token budget even at a pessimistic three -# tokens per character, so counting it would just be a wasted round trip inside a -# worker with a soft time limit. Most runs land here. -_COUNT_TOKENS_ABOVE_CHARS = 200_000 +# Byte budget for the projected trace payload. A byte-level BPE never emits more +# than one token per UTF-8 byte, so this cannot breach the provider's 1M input +# window; budgeting bytes rather than characters is also what charges a script +# that encodes wide (Devanagari at 3 bytes/char) for what it actually costs. +_TRACE_PAYLOAD_MAX_BYTES = 400_000 _TRACE_FIELD_MAX_CHARS = 2_000 _TRUNCATION_MARKER = " …[truncated]" -# Repeats of the same question are the first thing dropped when the brief is too -# large: cutting the 5th, then the 4th, keeps every question represented, whereas -# dropping rows loses questions outright. Below this floor the repeats no longer -# show whether the judge is stable, so row-dropping takes over instead. +# Repeats of one question are cut before whole rows: a question the model never +# sees at all is a bigger loss than the 5th repeat of one it does. _MIN_REPEATS_PER_QUESTION = 3 # Sorts unscoreable rows (value = "N/A") last, behind every real score. @@ -672,11 +656,7 @@ def _truncate(text: object) -> str: def _project_trace(trace: dict, *, is_judge_run: bool) -> dict: - """Keep only the fields the brief tells the model to read. - - `trace_id` / `question_id` are referenced nowhere in the prompt, and the v1 - brief never mentions the judge `comment`, so both are dropped. - """ + """Keep only the fields the brief tells the model to read.""" projected: dict = { "question": _truncate(trace.get("question")), "ground_truth_answer": _truncate(trace.get("ground_truth_answer")), @@ -698,46 +678,20 @@ def _project_trace(trace: dict, *, is_judge_run: bool) -> dict: return projected -def _group_key(index: int, trace: dict) -> object: - """Repeats of one question share its `question_id`. - - A trace without one is keyed uniquely so it is never treated as a repeat — - older datasets uploaded without question ids would otherwise collapse into a - single group and be cut down to a handful of rows. - """ - question_id = trace.get("question_id") - if question_id or question_id == 0: - return question_id - return f"__row_{index}" - - -def _max_repeats_observed(rows: list[dict]) -> int: - counts = Counter(_group_key(index, row) for index, row in enumerate(rows)) - return max(counts.values(), default=0) - - -def _repeat_ladder(rows: list[dict]) -> list[int]: - """Repeat limits to try, from the run's own duplication factor down to the floor.""" - observed = _max_repeats_observed(rows) - if observed <= _MIN_REPEATS_PER_QUESTION: - return [observed] - return list(range(observed, _MIN_REPEATS_PER_QUESTION - 1, -1)) - - -def _limit_repeats(rows: list[dict], *, max_repeats: int) -> list[dict]: - """Keep at most `max_repeats` traces per question, in file order. +def _limit_repeats(rows: list[dict]) -> list[dict]: + """Keep at most `_MIN_REPEATS_PER_QUESTION` traces per question, in file order. - Order, not score, decides which repeats survive: repeats exist to show how - stable the judge is on one question, so dropping the worst of them would erase - the very spread the judge brief is told to read. + A trace with no `question_id` is keyed uniquely, so datasets uploaded before + ids existed are not collapsed into one group and cut down to a handful of rows. """ kept: list[dict] = [] - seen: dict[object, int] = {} + seen: Counter[object] = Counter() for index, row in enumerate(rows): - key = _group_key(index, row) - if seen.get(key, 0) >= max_repeats: + question_id = row.get("question_id") + key = question_id if question_id or question_id == 0 else f"__row_{index}" + if seen[key] >= _MIN_REPEATS_PER_QUESTION: continue - seen[key] = seen.get(key, 0) + 1 + seen[key] += 1 kept.append(row) return kept @@ -754,107 +708,36 @@ def _primary_score(trace: dict, *, score_name: str) -> float: return _UNSCOREABLE_SORT_VALUE -def _select_traces( - rows: list[dict], - *, - is_judge_run: bool, - max_repeats: int | None = None, - max_chars: int = _TRACE_PAYLOAD_MAX_CHARS, -) -> tuple[list[dict], int]: - """Project, truncate, and cap the traces, worst-scoring first. - - Returns (kept, total) where total counts the whole run, so the brief can say - how much it is not seeing. Worst-first both selects and orders: the brief tells - the model to focus on the low-scoring rows, so those are the ones a cap must - keep. - """ +def _pack_traces(rows: list[dict], *, is_judge_run: bool) -> list[dict]: + """Project rows into the byte budget, worst-scoring first.""" score_name = GROUND_TRUTH_SCORE_NAME if is_judge_run else COSINE_SCORE_NAME - candidates = ( - rows if max_repeats is None else _limit_repeats(rows, max_repeats=max_repeats) - ) - ranked = sorted( - candidates, key=lambda row: _primary_score(row, score_name=score_name) - ) - kept: list[dict] = [] used = 0 - for row in ranked: + for row in sorted(rows, key=lambda r: _primary_score(r, score_name=score_name)): projected = _project_trace(row, is_judge_run=is_judge_run) - size = len(json.dumps(projected, ensure_ascii=False)) - if kept and used + size > max_chars: + size = len(json.dumps(projected, ensure_ascii=False).encode()) + if kept and used + size > _TRACE_PAYLOAD_MAX_BYTES: break kept.append(projected) used += size - - if len(kept) < len(rows): - logger.warning( - f"[_select_traces] Trimmed traces to fit the prompt budget | " - f"kept={len(kept)} total={len(rows)} chars={used} " - f"max_repeats={max_repeats} max_chars={max_chars} " - f"is_judge_run={is_judge_run}" - ) - return kept, len(rows) + return kept -def _count_input_tokens(*, user_message_text: str) -> int | None: - """Exact input-token count for the drafting request, or None when unavailable. +def _select_traces(rows: list[dict], *, is_judge_run: bool) -> tuple[list[dict], int]: + """Fit the run's traces into the byte budget; returns (kept, total). - Counted against the same model and output schema the real call uses, since both - are billed as input. Returns None — meaning "treat as fitting" — for a brief too - small to be at risk, and for any failure: this is a network call from a Celery - worker, and the character budget bounds the payload on its own. + Worst-first both selects and orders — the brief tells the model to focus on the + failing rows, so those are the ones a cap has to keep. Repeats are only capped + when the whole run does not fit. """ - if not settings.ANTHROPIC_API_KEY: - return None - if len(user_message_text) <= _COUNT_TOKENS_ABOVE_CHARS: - return None - try: - client = ClaudeProvider.create_client({"api_key": settings.ANTHROPIC_API_KEY}) - counted = client.messages.count_tokens( - model=settings.PROMPT_IMPROVEMENT_MODEL, - messages=[{"role": "user", "content": user_message_text}], - output_config={"format": {"type": "json_schema", "schema": _OUTPUT_SCHEMA}}, - ) - # A malformed / stubbed response must not crash the worker; the character - # budget already bounds the payload without a count. - return counted.input_tokens if isinstance(counted.input_tokens, int) else None - except Exception as exc: + kept = _pack_traces(rows, is_judge_run=is_judge_run) + if len(kept) < len(rows): + kept = _pack_traces(_limit_repeats(rows), is_judge_run=is_judge_run) logger.warning( - f"[_count_input_tokens] Count unavailable, falling back to the " - f"character budget | {exc}" - ) - return None - - -def _build_user_message( - *, - trace_description: str, - task_steps: list[str], - current_instructions: str, - target_config: dict, - selected_traces: list[dict], - total_traces: int, -) -> str: - """Assemble the drafting brief; rebuilt once per attempt as the payload shrinks.""" - trace_scope = ( - "" - if len(selected_traces) == total_traces - else ( - f" You are seeing {len(selected_traces)} of {total_traces} traces — " - "the lowest-scoring rows, and at most a few repeats per question; the " - "rest are omitted." + f"[_select_traces] Trimmed traces to fit the prompt budget | " + f"kept={len(kept)} total={len(rows)} is_judge_run={is_judge_run}" ) - ) - return ( - "You are a prompt engineer. Below is a JSON array of evaluation traces" - f"{trace_description}{trace_scope}\n\n" - "## Evaluation traces\n```\n" - f"{json.dumps(selected_traces, ensure_ascii=False)}\n```\n\n" - f"## Current system prompt\n```\n{current_instructions}\n```\n\n" - "## Target configuration (read-only — do NOT change any of these)\n" - f"```\n{json.dumps(target_config)}\n```\n\n" - "## Task\n" + "".join(task_steps) - ) + return kept, len(rows) def _draft_improved_prompt( @@ -943,59 +826,25 @@ def _draft_improved_prompt( target_config = _target_config_from_params(config_params) raw_rows = traces if isinstance(traces, list) else traces.get("traces") or [] rows = [row for row in raw_rows if isinstance(row, dict)] - - def brief(selected: list[dict], total: int) -> str: - return _build_user_message( - trace_description=trace_description, - task_steps=task_steps, - current_instructions=current_instructions, - target_config=target_config, - selected_traces=selected, - total_traces=total, - ) - - # Degrade repeats before rows: the 5th repeat of a question carries far less - # signal than a question the model never sees at all. - user_message_text = "" - counted: int | None = None - for max_repeats in _repeat_ladder(rows): - selected_traces, total_traces = _select_traces( - rows, is_judge_run=is_judge_run, max_repeats=max_repeats - ) - user_message_text = brief(selected_traces, total_traces) - counted = _count_input_tokens(user_message_text=user_message_text) - if counted is None or counted <= _INPUT_TOKEN_BUDGET: - return _call_prompt_drafting_llm(user_message_text=user_message_text) - logger.warning( - f"[_draft_improved_prompt] Over the input budget, dropping a repeat | " - f"max_repeats={max_repeats} tokens={counted} " - f"budget={_INPUT_TOKEN_BUDGET}" + selected_traces, total_traces = _select_traces(rows, is_judge_run=is_judge_run) + trace_scope = ( + "" + if len(selected_traces) == total_traces + else ( + f" You are seeing {len(selected_traces)} of {total_traces} traces — the " + "lowest-scoring rows; the rest are omitted." ) + ) - # The repeat floor is not a get-out: at 3 repeats a long enough run is still - # over, so stop cutting repeats and cut rows instead, converting the measured - # count back into a character budget. - if counted: - scaled_chars = max( - _TRACE_FIELD_MAX_CHARS, - int( - _TRACE_PAYLOAD_MAX_CHARS - * _TOKEN_BUDGET_SAFETY - * _INPUT_TOKEN_BUDGET - / counted - ), - ) - logger.warning( - f"[_draft_improved_prompt] Repeat floor reached, dropping rows | " - f"max_repeats={_MIN_REPEATS_PER_QUESTION} tokens={counted} " - f"scaled_chars={scaled_chars}" - ) - selected_traces, total_traces = _select_traces( - rows, - is_judge_run=is_judge_run, - max_repeats=_MIN_REPEATS_PER_QUESTION, - max_chars=scaled_chars, - ) - user_message_text = brief(selected_traces, total_traces) + user_message_text = ( + "You are a prompt engineer. Below is a JSON array of evaluation traces" + f"{trace_description}{trace_scope}\n\n" + "## Evaluation traces\n```\n" + f"{json.dumps(selected_traces, ensure_ascii=False)}\n```\n\n" + f"## Current system prompt\n```\n{current_instructions}\n```\n\n" + "## Target configuration (read-only — do NOT change any of these)\n" + f"```\n{json.dumps(target_config)}\n```\n\n" + "## Task\n" + "".join(task_steps) + ) return _call_prompt_drafting_llm(user_message_text=user_message_text) diff --git a/backend/app/tests/api/routes/test_improve_prompt.py b/backend/app/tests/api/routes/test_improve_prompt.py index 615580a57..587863cbe 100644 --- a/backend/app/tests/api/routes/test_improve_prompt.py +++ b/backend/app/tests/api/routes/test_improve_prompt.py @@ -55,7 +55,6 @@ COMMIT_MESSAGE_MAX_LENGTH, _anthropic_error_detail, _call_prompt_drafting_llm, - _count_input_tokens, _draft_improved_prompt, _primary_score, execute_prompt_improvement, @@ -115,9 +114,6 @@ def _make_fake_claude_client(text_content: str | None = None) -> MagicMock: client = MagicMock() client.messages.create.return_value = response - # The drafting path counts input tokens before it calls; a real int keeps the - # repeat-degradation ladder on its first rung for these fixtures. - client.messages.count_tokens.return_value = MagicMock(input_tokens=1_000) return client @@ -1060,7 +1056,7 @@ def _judge_trace(question_id: int, ground_truth_value: float | str) -> dict: return { "trace_id": f"t{question_id}", "question_id": question_id, - "question": f"Q{question_id} " + "\u0915" * 3000, + "question": f"Q{question_id} " + "क" * 3000, "ground_truth_answer": "A" * 5000, "llm_answer": "B" * 5000, "scores": [ @@ -1075,30 +1071,24 @@ def _judge_trace(question_id: int, ground_truth_value: float | str) -> dict: } @staticmethod - def _capture_brief(traces: list, *, token_counts: list[int] | None = None) -> str: - """Return the brief the drafting call would have received. + def _repeated_run(questions: int, repeats: int) -> list[dict]: + """A judged run of `questions` questions scored `repeats` times each.""" + return [ + TestTraceBudget._judge_trace(question_id, 2.0) + for question_id in range(questions) + for _ in range(repeats) + ] - ``token_counts`` feeds one measured input-token count per ladder rung; the - last value repeats once the list is exhausted. Under the budget on the first - rung by default, so the ladder stays put unless a test says otherwise. - """ - counts = list(token_counts or [1_000]) + @staticmethod + def _capture_brief(traces: list) -> str: + """Return the brief the drafting call would have received.""" captured: dict[str, str] = {} - def fake_count(*, user_message_text: str) -> int: - return counts.pop(0) if len(counts) > 1 else counts[0] - def fake_call(*, user_message_text: str) -> tuple[str, str]: captured["text"] = user_message_text return "improved", "why" - with ExitStack() as stack: - stack.enter_context( - patch(f"{_SERVICE}._count_input_tokens", side_effect=fake_count) - ) - stack.enter_context( - patch(f"{_SERVICE}._call_prompt_drafting_llm", side_effect=fake_call) - ) + with patch(f"{_SERVICE}._call_prompt_drafting_llm", side_effect=fake_call): assert _draft_improved_prompt( current_instructions="be helpful", config_params={"model": "gpt-4o"}, @@ -1132,62 +1122,43 @@ def test_small_run_is_sent_whole_without_a_scope_note(self) -> None: assert " traces —" not in text - @staticmethod - def _repeated_run(questions: int, repeats: int) -> list[dict]: - """A judged run of `questions` questions scored `repeats` times each.""" - return [ - TestTraceBudget._judge_trace(question_id, 2.0) - for question_id in range(questions) - for _ in range(repeats) - ] - - def test_repeats_are_dropped_one_at_a_time_until_the_brief_fits(self) -> None: - traces = self._repeated_run(questions=4, repeats=5) + def test_repeats_are_capped_before_any_question_is_dropped(self) -> None: + # Over budget at 5 repeats per question, under it at the floor of 3. + questions = 8 + traces = self._repeated_run(questions=questions, repeats=5) - # Over budget at 5 and 4 repeats per question, under it at 3. - text = self._capture_brief(traces, token_counts=[1_100_000, 950_000, 800_000]) + text = self._capture_brief(traces) # Every question is still represented, three times each — no question was # dropped to make room. - for question_id in range(4): - assert text.count(f"Q{question_id} ") == 3 + for question_id in range(questions): + assert text.count(f"Q{question_id} ") == _MIN_REPEATS_PER_QUESTION - def test_row_dropping_takes_over_at_the_repeat_floor(self) -> None: - # Sized so the flat character budget alone would keep every row: only the - # floor's rescaled budget can drop any, which is what this asserts. - questions = 9 + def test_rows_are_dropped_when_the_repeat_floor_is_not_enough(self) -> None: + questions = 15 traces = self._repeated_run(questions=questions, repeats=5) - # Still far over budget after the ladder bottoms out at 3 repeats. - text = self._capture_brief(traces, token_counts=[2_800_000]) + text = self._capture_brief(traces) kept = sum(text.count(f"Q{question_id} ") for question_id in range(questions)) assert 0 < kept < questions * _MIN_REPEATS_PER_QUESTION assert f"of {len(traces)} traces" in text - def test_a_small_brief_is_never_sent_for_counting( - self, anthropic_creds: None - ) -> None: - """No round trip for a brief that cannot breach the budget.""" - with patch(f"{_SERVICE}.ClaudeProvider.create_client") as create_client: - assert _count_input_tokens(user_message_text="rewrite this") is None - create_client.assert_not_called() - def test_traces_without_question_ids_still_get_trimmed(self) -> None: """v1 traces from older datasets can carry no `question_id`. - Those must never be grouped as repeats of one another, so the repeat ladder - is a no-op for them and the rescaled character budget is the only thing that - can bring an oversized brief down. + Those must never be grouped as repeats of one another, so capping repeats is + a no-op for them and the byte budget is the only thing that trims the brief. """ - traces = [self._judge_trace(i, 2.0) for i in range(20)] + questions = 40 + traces = [self._judge_trace(i, 2.0) for i in range(questions)] for trace in traces: del trace["question_id"] - text = self._capture_brief(traces, token_counts=[2_800_000]) + text = self._capture_brief(traces) - kept = sum(text.count(f"Q{i} ") for i in range(20)) - assert 0 < kept < 20 + kept = sum(text.count(f"Q{i} ") for i in range(questions)) + assert 0 < kept < questions def test_unscoreable_score_sorts_behind_every_real_score(self) -> None: """`value` is the string "N/A" on unscoreable rows; mixed str/float must not diff --git a/docs/wiki/modules/evaluations.md b/docs/wiki/modules/evaluations.md index 50bf2a1aa..3419169dd 100644 --- a/docs/wiki/modules/evaluations.md +++ b/docs/wiki/modules/evaluations.md @@ -40,7 +40,7 @@ v2 judge field on `EvaluationRun`: `is_judge_run` (bool marker gating native jud - Fast text runs (v1 cosine + v2 judge) fan out `ceil(total_items / EVAL_FAST_CHUNK_SIZE)` `run_evaluation_fast_chunk` tasks (responses only), then a cron barrier (`dispatch_fast_evaluation_barriers`) enqueues one `run_evaluation_fast_aggregate` once every chunk has a `raw_output_url`. The aggregate merges chunks, then (v2) judges **every** row in that single task — so the judge pool is sized by its own `EVAL_JUDGE_CONCURRENCY` (not the response stage's `EVAL_FAST_API_CONCURRENCY`) to clear the max dataset (`EVAL_FAST_MAX_UNIQUE_ROWS` × `duplication_factor`) under the aggregate's `CELERY_TASK_SOFT_TIME_LIMIT`. No judge fan-out / second barrier. - Prompt improvement is job-based with callback delivery: `POST /evaluations/{id}/improve-prompt` validates preconditions, enqueues a `Job` (`JobType.PROMPT_IMPROVEMENT`, `models/job.py`) run by Celery task `run_prompt_improvement` (`celery/tasks/job_execution.py`), and returns `202` with an `LLMJobImmediatePublic` handle. On finish the worker POSTs a single best-effort callback to the caller-supplied `callback_url` (SSRF-guarded via `validate_callback_url`): an `APIResponse[PromptImprovementJobPublic]` (`models/evaluation.py`) carrying the new `ConfigVersion` on success or `error_message` on failure (for an Anthropic fault `error_message` carries the provider's own response body, appended by `_anthropic_error_detail` in `services/evaluations/prompt_improvement.py`). The `ConfigVersion` is persisted regardless of callback outcome. Celery redelivery of a `SUCCESS` job re-sends the callback without re-running the LLM. - v2 prompt iteration reuses the same job/Celery/config-version machinery (`services/evaluations/prompt_improvement.py`) and branches on `run.is_judge_run`: the v2 route calls `start_prompt_improvement_job(..., require_judge_run=True)` (a non-judge run → `422 not_a_judge_run`), and the worker drafts from the three-metric judge trace (`_draft_improved_prompt(is_judge_run=True)`, using each metric's score + reasoning) and delivers a `PromptRecommendationJobPublic` callback carrying `recommendation_type` (`Literal["prompt"]`, `models/evaluation.py`; widens to a union when knowledge-base / model recommendations land). Non-judge (v1) runs keep the default `_draft_improved_prompt` brief + `PromptImprovementJobPublic` path unchanged. -- Both prompt-improvement briefs are size-bounded before the Anthropic call (`services/evaluations/prompt_improvement.py`) — a large run used to exceed the provider's 1M-token input limit and fail the job with `prompt_generation_failed: ... HTTP 400`. Every trace is projected down to the fields the brief names and its long text fields (notably the judge `comment`) are clamped; then the brief is measured with `messages.count_tokens` (`_count_input_tokens`, same model + output schema as the real call) and degraded until it fits `_INPUT_TOKEN_BUDGET`, in two stages: first `_repeat_ladder` drops one repeat per question at a time, from the run's duplication factor down to `_MIN_REPEATS_PER_QUESTION` (3) — every question stays represented; only below that floor does `_select_traces` start dropping whole rows, worst-scoring-first against a budget rescaled from the measured count. Counting is skipped entirely for a brief under `_COUNT_TOKENS_ABOVE_CHARS` (it cannot breach the budget even at a pessimistic three tokens per character), and a `count_tokens` failure is non-fatal: the flat `_TRACE_PAYLOAD_MAX_CHARS` cap applies on every attempt regardless. The brief states how many traces it is not seeing, and each degradation step is logged. `crud/evaluations/summary.py` still sends every trace whole (its own `ponytail:` note) and has the same latent ceiling. +- Both prompt-improvement briefs are size-bounded before the Anthropic call (`services/evaluations/prompt_improvement.py`) — a large run used to exceed the provider's 1M-token input limit and fail the job with `prompt_generation_failed: ... HTTP 400`. Every trace is projected down to the fields the brief names and its long text fields (notably the judge `comment`) are clamped; `_select_traces` then fills `_TRACE_PAYLOAD_MAX_BYTES` worst-scoring-first (the brief tells the model to focus on the failing rows, so those are what a cap keeps), and only when the whole run does not fit does `_limit_repeats` cap repeats of a question at `_MIN_REPEATS_PER_QUESTION` (3) — a question the model never sees is a bigger loss than its 5th repeat, and a trace with no `question_id` is keyed uniquely so pre-id datasets are never collapsed into one group. The budget is bytes rather than characters because a byte-level BPE never emits more than one token per UTF-8 byte: the bound needs no `count_tokens` round trip, and a wide-encoding script (Devanagari at 3 bytes/char) is charged for what it actually costs. The brief states how many traces it is not seeing, and trimming is logged. `crud/evaluations/summary.py` still sends every trace whole (its own `ponytail:` note) and has the same latent ceiling. - Evaluation iteration loop (`POST /evaluations/iterations`) chains fast-eval + v2 prompt improvement into a self-driving cycle via LangGraph (`services/evaluations/iteration_graph.py`): `start_eval_node` → `wait_eval_node` → (conditional) `start_improve_node` → `wait_improve_node` → loops back to `start_eval_node`, or → `finalize_node`. Stop-score = mean(`Adherence to Ground Truth`, `Adherence to Prompt`) from `EvaluationRun.score["summary_scores"]` (`compute_round_scores`); `Adherence to Knowledge Base` is recorded per round for visibility only, never gates stopping. Stops on 3 consecutive rounds with `