diff --git a/backend/app/services/evaluations/prompt_improvement.py b/backend/app/services/evaluations/prompt_improvement.py index 9b92d5c22..be8d93d8e 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 typing import NotRequired, TypedDict, cast from uuid import UUID import anthropic @@ -28,6 +29,8 @@ GROUND_TRUTH_SCORE_NAME, KNOWLEDGE_BASE_SCORE_NAME, PROMPT_SCORE_NAME, + TraceData, + TraceScore, ) from app.crud.jobs import JobCrud from app.models.config.config import ConfigTag @@ -54,14 +57,12 @@ # Headroom for a full prompt rewrite + JSON wrapper; too low truncates into invalid JSON. _LLM_MAX_TOKENS = 16384 -# JSON keys expected in the LLM's structured response. _LLM_KEY_INSTRUCTIONS = "improved_instructions" _LLM_KEY_RATIONALE = "rationale" # Cap the rationale so it stays a one-line summary in the commit_message. _RATIONALE_MAX_LENGTH = 200 -# Structured-output schema: guarantees the response is valid JSON with these fields. _OUTPUT_SCHEMA = { "type": "object", "properties": { @@ -134,7 +135,6 @@ def validate_improve_prompt( detail=f"evaluation_not_completed: run status is '{run.status}', must be '{_COMPLETED_STATUS}'", ) if not run.score_trace_url: - # Run exists but a precondition is unmet, so 422 rather than 404. raise HTTPException( status_code=422, detail=( @@ -143,7 +143,6 @@ def validate_improve_prompt( ), ) if run.config_id is None or run.config_version is None: - # Both FKs are nullable; without them there's no version to improve. raise HTTPException( status_code=422, detail="source_config_unavailable: run has no config_id/config_version reference", @@ -406,9 +405,9 @@ def execute_prompt_improvement( traces = load_json_from_object_store( storage=storage, url=run.score_trace_url ) - if traces is None: + if not isinstance(traces, list): raise RuntimeError( - "trace_download_failed: could not retrieve trace file from storage" + "trace_download_failed: could not retrieve trace list from storage" ) improved_instructions, rationale = _draft_improved_prompt( @@ -484,8 +483,6 @@ def execute_prompt_improvement( ) raise except Exception as exc: - # HTTPException (from re-validation) carries the clean detail; everything - # else falls back to str(exc). error_message = getattr(exc, "detail", None) or str(exc) logger.error( f"[execute_prompt_improvement] Failed | job_id={job_id} " @@ -565,7 +562,6 @@ def _call_prompt_drafting_llm(*, user_message_text: str) -> tuple[str, str]: per-cause log lines are kept so operators can still triage the fault. """ if not settings.ANTHROPIC_API_KEY: - # Missing platform key is a server-side misconfiguration, not an upstream fault. raise RuntimeError( "prompt_generation_failed: the platform Anthropic key " "(ANTHROPIC_API_KEY) is not configured" @@ -646,11 +642,83 @@ def _call_prompt_drafting_llm(*, user_message_text: str) -> tuple[str, str]: ) from exc +class TraceScoreSummary(TypedDict): + """A TraceScore without the UI-only fields (data_type, verdict).""" + + name: str + value: float | str + comment: NotRequired[str] + unscoreable: NotRequired[bool] + + +class TraceAttempt(TypedDict): + """One run of a question: its answer and scores.""" + + llm_answer: str + scores: list[TraceScoreSummary] + + +class TraceQuestion(TypedDict): + """A dataset question with one attempt per duplication-factor repeat.""" + + question_id: int | str | None + question: str + ground_truth_answer: str + attempts: list[TraceAttempt] + + +def _trace_score_summary(score: TraceScore) -> TraceScoreSummary: + return cast( + TraceScoreSummary, + {k: v for k, v in score.items() if k not in ("data_type", "verdict")}, + ) + + +def _format_traces_for_brief(traces: list[TraceData]) -> list[TraceQuestion]: + """Group repeated attempts per question and keep only what the brief reads. + + A 500-row Hindi judge run blew past Claude's 1M-token context: every repeat + (duplication factor) re-shipped question + ground truth, and trace_id / + data_type / verdict / guardrail bookkeeping rode along for nothing. Rows the + guardrails rewrote substitute `input_to_llm` / `output_from_llm`, since the + prompt under repair acts on what the LLM actually saw and produced, not on + the user-facing text. Not `crud.evaluations.core.group_traces_by_question_id`: + that is the export shape (keeps trace_ids/category, scores as parallel lists) + and rejects v1 traces that have no question_id. + """ + grouped: dict[int | str | tuple[str, str], TraceQuestion] = {} + for trace in traces: + question = trace.get("input_to_llm") or trace.get("question", "") + ground_truth = trace.get("ground_truth_answer", "") + question_id = trace.get("question_id") + # v1 traces may lack question_id; the text pair is the next-best identity. + key = question_id if question_id is not None else (question, ground_truth) + entry = grouped.setdefault( + key, + { + "question_id": question_id, + "question": question, + "ground_truth_answer": ground_truth, + "attempts": [], + }, + ) + entry["attempts"].append( + { + "llm_answer": trace.get("output_from_llm") + or trace.get("llm_answer", ""), + "scores": [ + _trace_score_summary(score) for score in trace.get("scores", []) + ], + } + ) + return list(grouped.values()) + + def _draft_improved_prompt( *, current_instructions: str, config_params: dict, - traces: list | dict, + traces: list[TraceData], is_judge_run: bool = False, ) -> tuple[str, str]: """Rewrite the prompt from a run's score traces. @@ -662,9 +730,13 @@ def _draft_improved_prompt( """ if is_judge_run: trace_description = ( - " from an LLM-as-judge run. Each trace has the fields: `question`, " - "`ground_truth_answer`, `llm_answer`, and `scores`. `scores` is a list of " - "metric objects; each has `name`, `value`, and `comment`, where:\n" + " from an LLM-as-judge run. Each entry is one dataset question with the " + "fields: `question_id`, `question`, `ground_truth_answer`, and `attempts`. " + "`attempts` is a list, one per time the question was run, each with " + "`llm_answer` and `scores`; a wide spread of scores across a question's " + "attempts means the prompt yields inconsistent behaviour on it. `scores` " + "is a list of metric objects; each has `name`, `value`, and `comment`, " + "where:\n" f"- `name` is one of `{GROUND_TRUTH_SCORE_NAME}`, `{PROMPT_SCORE_NAME}`, " f"or `{KNOWLEDGE_BASE_SCORE_NAME}`.\n" "- `value` is the metric's score, an integer from 0 (worst) to 5 (best). Metrics that " @@ -706,9 +778,10 @@ def _draft_improved_prompt( ] else: trace_description = ( - ". Each trace has the fields: `question`, `ground_truth_answer`, " - "`llm_answer`, `category`, and `scores` (a list of scoring objects with " - "`name`, `value`, and `unscoreable`)." + ". Each entry is one dataset question with the fields: `question`, " + "`ground_truth_answer`, and `attempts`, a list, one per time the " + "question was run, each with `llm_answer` and `scores` (a list of " + "scoring objects with `name`, `value`, and `unscoreable`)." ) task_steps = [ "1. Identify the answers that performed poorly — those with low scores or " @@ -730,11 +803,14 @@ def _draft_improved_prompt( f"sentence (≤ {_RATIONALE_MAX_LENGTH} characters): what you changed and why." ) target_config = _target_config_from_params(config_params) + traces_json = json.dumps( + _format_traces_for_brief(traces), ensure_ascii=False, separators=(",", ":") + ) 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"## Evaluation traces\n```\n{traces_json}\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_v2.py b/backend/app/tests/api/routes/test_improve_prompt_v2.py index c83c5d9d4..a4fc3f08b 100644 --- a/backend/app/tests/api/routes/test_improve_prompt_v2.py +++ b/backend/app/tests/api/routes/test_improve_prompt_v2.py @@ -42,6 +42,7 @@ from app.models.job import Job, JobStatus, JobType from app.services.evaluations.prompt_improvement import ( AI_GENERATED_MARKER, + _format_traces_for_brief, execute_prompt_improvement, ) from app.services.llm.providers.claude import STOP_REASON_COMPLETE @@ -671,3 +672,209 @@ def test_redelivery_of_success_judge_job_resends_v2_callback_without_new_version assert payload["success"] is True assert payload["data"]["recommendation_type"] == "prompt" assert payload["data"]["config_version"]["version"] == first_version + + +def _score(name: str, value: int | str, comment: str) -> dict: + return { + "name": name, + "value": value, + "data_type": "NUMERIC", + "comment": comment, + "verdict": "Good", + } + + +class TestFormatTracesForBrief: + """Pure-function checks on the trace projection fed to the drafting LLM.""" + + def test_groups_repeats_per_question_and_drops_bookkeeping(self) -> None: + traces = [ + { + "trace_id": "t1", + "question": "Q", + "llm_answer": "A1", + "ground_truth_answer": "G", + "question_id": 7, + "category": "default", + "guardrail": None, + "input_to_llm": None, + "output_from_llm": None, + "scores": [_score(GROUND_TRUTH_SCORE_NAME, 1, "wrong")], + }, + { + "trace_id": "t2", + "question": "Q", + "llm_answer": "A2", + "ground_truth_answer": "G", + "question_id": 7, + "guardrail": None, + "input_to_llm": None, + "output_from_llm": None, + "scores": [ + _score(GROUND_TRUTH_SCORE_NAME, 5, "right"), + { + "name": KNOWLEDGE_BASE_SCORE_NAME, + "value": "N/A", + "data_type": "CATEGORICAL", + "comment": "no KB", + "unscoreable": True, + }, + ], + }, + ] + + compact = _format_traces_for_brief(traces) + + assert compact == [ + { + "question_id": 7, + "question": "Q", + "ground_truth_answer": "G", + "attempts": [ + { + "llm_answer": "A1", + "scores": [ + { + "name": GROUND_TRUTH_SCORE_NAME, + "value": 1, + "comment": "wrong", + } + ], + }, + { + "llm_answer": "A2", + "scores": [ + { + "name": GROUND_TRUTH_SCORE_NAME, + "value": 5, + "comment": "right", + }, + { + "name": KNOWLEDGE_BASE_SCORE_NAME, + "value": "N/A", + "comment": "no KB", + "unscoreable": True, + }, + ], + }, + ], + } + ] + + def test_guardrail_rewritten_rows_use_what_the_llm_saw(self) -> None: + traces = [ + { + "trace_id": "t1", + "question": "user text with PII", + "llm_answer": "answer shown to user", + "ground_truth_answer": "G", + "question_id": 1, + "guardrail": "applied", + "input_to_llm": "user text masked", + "output_from_llm": "raw model answer", + "scores": [], + }, + { + "trace_id": "t2", + "question": "Q2", + "llm_answer": "A2", + "ground_truth_answer": "G2", + "question_id": 2, + "guardrail": "applied", + "input_to_llm": "Q2 masked", + "output_from_llm": None, + "scores": [], + }, + ] + + compact = _format_traces_for_brief(traces) + + assert compact[0]["question"] == "user text masked" + assert compact[0]["attempts"][0]["llm_answer"] == "raw model answer" + # Only the input side was rewritten; the answer falls back to llm_answer. + assert compact[1]["question"] == "Q2 masked" + assert compact[1]["attempts"][0]["llm_answer"] == "A2" + for entry in compact: + assert "guardrail" not in entry + assert "input_to_llm" not in entry + assert "output_from_llm" not in entry + assert "trace_id" not in entry + + def test_v1_traces_without_question_id_group_by_text(self) -> None: + traces = [ + { + "trace_id": "t1", + "question": "Q", + "llm_answer": "A1", + "ground_truth_answer": "G", + "category": "Geo", + "scores": [{"name": "cosine_similarity", "value": 0.3}], + }, + { + "trace_id": "t2", + "question": "Q", + "llm_answer": "A2", + "ground_truth_answer": "G", + "category": "Geo", + "scores": [{"name": "cosine_similarity", "value": 0.9}], + }, + { + "trace_id": "t3", + "question": "Other", + "llm_answer": "A3", + "ground_truth_answer": "G3", + "scores": [], + }, + ] + + compact = _format_traces_for_brief(traces) + + assert [entry["question"] for entry in compact] == ["Q", "Other"] + assert compact[0]["question_id"] is None + assert [a["llm_answer"] for a in compact[0]["attempts"]] == ["A1", "A2"] + assert "category" not in compact[0] + + +class TestV2BriefEncoding: + """The drafting message must ship Indic text raw, not \\uXXXX-escaped.""" + + def _pending_job(self, db: Session, project_id: int) -> Job: + return JobCrud(session=db).create( + job_type=JobType.PROMPT_IMPROVEMENT, + trace_id="N/A", + project_id=project_id, + ) + + def test_indic_text_is_not_ascii_escaped_in_brief( + self, + db: Session, + auth: TestAuthContext, + judge_run: EvaluationRun, + ) -> None: + hindi_question = "भारत की राजधानी क्या है?" + hindi_answer = "नई दिल्ली" + traces = [ + { + **_JUDGE_TRACES[0], + "question": hindi_question, + "llm_answer": hindi_answer, + "ground_truth_answer": hindi_answer, + } + ] + job = self._pending_job(db, auth.project_id) + + with _worker_env(db, traces=traces) as env: + execute_prompt_improvement( + project_id=auth.project_id, + job_id=str(job.id), + organization_id=auth.organization_id, + evaluation_id=judge_run.id, + callback_url=_CALLBACK_URL, + ) + + message = _llm_user_message(env.llm) + assert hindi_question in message + assert hindi_answer in message + assert "\\u0" not in message + assert '"attempts":[{"llm_answer":' in message + assert '"trace_id"' not in message diff --git a/docs/wiki/modules/evaluations.md b/docs/wiki/modules/evaluations.md index c39f05de9..630c34134 100644 --- a/docs/wiki/modules/evaluations.md +++ b/docs/wiki/modules/evaluations.md @@ -38,7 +38,7 @@ v2 judge field on `EvaluationRun`: `is_judge_run` (bool marker gating native jud ## Async - Provider batches polled by cron (`crud/evaluations/cron.py`); no long-lived Celery task per run. - 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. The aggregate persists the score unit (`save_score` → traces to S3 behind `score_trace_url`, summary + overall to the DB) **before** flipping the run to `completed`: on v2 that unit is the only copy of the per-row judge scores, so a crash between the two must leave the run `processing`, never `completed` with its scores gone. Chunk-artifact cleanup and the v1 Langfuse score sync run after completion as a best-effort tail that may never fail the run. -- 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. `_call_prompt_drafting_llm` keeps `json_schema` structured output (two fields, `rationale` length-capped) but raises `prompt_generation_failed` when `stop_reason != STOP_REASON_COMPLETE`, so a `max_tokens`-truncated draft fails the job instead of minting a half-written system prompt. Celery redelivery of a `SUCCESS` job re-sends the callback without re-running the LLM. +- 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. `_call_prompt_drafting_llm` keeps `json_schema` structured output (two fields, `rationale` length-capped) but raises `prompt_generation_failed` when `stop_reason != STOP_REASON_COMPLETE`, so a `max_tokens`-truncated draft fails the job instead of minting a half-written system prompt. Before the call, `_format_traces_for_brief` projects the score traces to what the brief actually reads (`TraceQuestion` / `TraceAttempt` / `TraceScoreSummary` TypedDicts in the same module) — one entry per question (`question_id`, `question`, `ground_truth_answer`) with repeated runs nested as `attempts[{llm_answer, scores[{name, value, comment, unscoreable}]}]`, dropping `trace_id`/`data_type`/`verdict`/`category`/guardrail bookkeeping; a guardrail-rewritten row substitutes `input_to_llm`/`output_from_llm` for `question`/`llm_answer` — and the JSON is dumped with `ensure_ascii=False` (escaping Devanagari inflates its token count several-fold; a 500-row Hindi run previously exceeded Claude's 1M-token context). 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. - 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 `