From 33465c606019f2eba6dabe0feff6d639f161d18e Mon Sep 17 00:00:00 2001 From: Ayush8923 <80516839+Ayush8923@users.noreply.github.com> Date: Fri, 24 Jul 2026 20:03:16 +0530 Subject: [PATCH 1/2] feat(evals): v2 prompt iteration flow --- .../api/docs/evaluation/improve_prompt_v2.md | 29 + backend/app/api/main.py | 4 + .../evaluations/prompt_improvement_v2.py | 66 ++ backend/app/models/evaluation.py | 27 + .../evaluations/prompt_improvement.py | 219 ++++-- .../api/routes/test_improve_prompt_v2.py | 670 ++++++++++++++++++ docs/wiki/modules/evaluations.md | 2 + 7 files changed, 966 insertions(+), 51 deletions(-) create mode 100644 backend/app/api/docs/evaluation/improve_prompt_v2.md create mode 100644 backend/app/api/routes/evaluations/prompt_improvement_v2.py create mode 100644 backend/app/tests/api/routes/test_improve_prompt_v2.py diff --git a/backend/app/api/docs/evaluation/improve_prompt_v2.md b/backend/app/api/docs/evaluation/improve_prompt_v2.md new file mode 100644 index 000000000..060820f6c --- /dev/null +++ b/backend/app/api/docs/evaluation/improve_prompt_v2.md @@ -0,0 +1,29 @@ +Enqueue a v2 prompt-recommendation job for the configuration evaluated by a judged run. + +Unlike v1 (which reads cosine similarity + correctness), this consumes the native +three-metric judge results — **Adherence to Ground Truth**, **Adherence to Prompt**, +and **Adherence to Knowledge Base** — each carrying a 0–1 score and the judge's +reasoning. The worker reads both the score and the reasoning per metric, focuses the +rewrite on rows where Adherence to Prompt / Ground Truth are low, and changes only +`completion.params.instructions`; model, knowledge base, and all other settings are +preserved. Low Adherence to Knowledge Base is treated as a retrieval/KB gap rather +than a prompt fault. + +The run must be a **judged v2 run** (`is_judge_run = true`), `status = completed`, +and have a `score_trace_url`. The request body is identical to v1 — `callback_url` +only. The LLM round-trip runs on a background worker, so this returns `202 Accepted` +with a `job_id` immediately. + +The success/failure result is POSTed to `callback_url` as an `APIResponse` envelope +whose `data` is a `PromptRecommendationJobPublic`. It adds `recommendation_type`, +which is `"prompt"` for now; knowledge-base and model recommendation types are +deferred to a later phase but the field keeps the callback API extensible. + +**Validation errors (returned synchronously before the job is created):** +- `404 evaluation_not_found` — no run with this id in the caller's project. +- `409 evaluation_not_completed` — run is not yet completed. +- `409 source_config_unavailable` — the run's config or config_version has been deleted. +- `422 traces_not_available` — the run has no `score_trace_url`. +- `422 not_a_judge_run` — the run is not a judged (v2) run. +- `422 invalid_callback_url` — `callback_url` is missing, not `https://`, or resolves to a blocked (private/loopback/metadata) address. +- `500 prompt_improvement_enqueue_failed` — the job could not be queued. diff --git a/backend/app/api/main.py b/backend/app/api/main.py index a36ebcef1..9e96e7fc4 100644 --- a/backend/app/api/main.py +++ b/backend/app/api/main.py @@ -41,6 +41,9 @@ router as evaluations_dataset_v2_router, ) from app.api.routes.evaluations.evaluation_v2 import router as evaluations_v2_router +from app.api.routes.evaluations.prompt_improvement_v2 import ( + router as evaluations_prompt_improvement_v2_router, +) api_router = APIRouter() api_router.include_router(analytics.router) @@ -85,3 +88,4 @@ api_v2_router = APIRouter() api_v2_router.include_router(evaluations_v2_router) api_v2_router.include_router(evaluations_dataset_v2_router) +api_v2_router.include_router(evaluations_prompt_improvement_v2_router) diff --git a/backend/app/api/routes/evaluations/prompt_improvement_v2.py b/backend/app/api/routes/evaluations/prompt_improvement_v2.py new file mode 100644 index 000000000..bf450e236 --- /dev/null +++ b/backend/app/api/routes/evaluations/prompt_improvement_v2.py @@ -0,0 +1,66 @@ +"""v2 prompt-improvement trigger — consumes three-metric judge results. + +Mirrors the v1 improve-prompt route but requires a judged (v2) run and returns a +typed prompt recommendation. The v1 route is left unchanged. +""" + +import logging + +from fastapi import APIRouter, Depends, HTTPException + +from app.api.deps import AuthContextDep, SessionDep +from app.api.permissions import Permission, require_permission +from app.models.evaluation import ImprovePromptRequest +from app.models.llm.response import LLMJobImmediatePublic +from app.services.evaluations import start_prompt_improvement_job +from app.utils import APIResponse, load_description, validate_callback_url + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/evaluations", tags=["Evaluation v2"]) + + +@router.post( + "/{evaluation_id}/improve-prompt", + description=load_description("evaluation/improve_prompt_v2.md"), + response_model=APIResponse[LLMJobImmediatePublic], + status_code=202, + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def improve_evaluation_prompt_v2( + evaluation_id: int, + session: SessionDep, + auth_context: AuthContextDep, + request: ImprovePromptRequest, +) -> APIResponse[LLMJobImmediatePublic]: + """Enqueue a v2 prompt-recommendation job for a judged evaluation run.""" + logger.info( + f"[improve_evaluation_prompt_v2] Starting | evaluation_id={evaluation_id} " + f"org_id={auth_context.organization_.id} project_id={auth_context.project_.id}" + ) + + # No global ValueError handler exists, so map the SSRF guard's ValueError to + # a 422 here rather than letting it surface as a 500. + try: + validate_callback_url(str(request.callback_url)) + except ValueError as exc: + raise HTTPException(status_code=422, detail=f"invalid_callback_url: {exc}") + + job = start_prompt_improvement_job( + session=session, + evaluation_id=evaluation_id, + organization_id=auth_context.organization_.id, + project_id=auth_context.project_.id, + callback_url=str(request.callback_url), + require_judge_run=True, + ) + + return APIResponse.success_response( + data=LLMJobImmediatePublic( + job_id=job.id, + status=job.status.value, + message="Prompt recommendation is running; the result will be delivered to your callback_url.", + job_inserted_at=job.inserted_at, + job_updated_at=job.updated_at, + ) + ) diff --git a/backend/app/models/evaluation.py b/backend/app/models/evaluation.py index 3e165e487..3d3f8b1e8 100644 --- a/backend/app/models/evaluation.py +++ b/backend/app/models/evaluation.py @@ -583,3 +583,30 @@ class PromptImprovementJobPublic(SQLModel): status: str config_version: ConfigVersionPublic | None = None error_message: str | None = None + + +class RecommendationTypeEnum(str, Enum): + """Kind of recommendation a v2 improvement job returns. + + Only prompt rewrites ship now; KNOWLEDGE_BASE and MODEL are deferred to a + later phase but named here so the callback API stays extensible. + """ + + PROMPT = "prompt" + # KNOWLEDGE_BASE = "knowledge_base" # deferred + # MODEL = "model" # deferred + + +class PromptRecommendationJobPublic(SQLModel): + """v2 callback payload: a typed recommendation from a judged (v2) run. + + Distinct from the v1 PromptImprovementJobPublic so the v1 callback shape + stays byte-for-byte unchanged; adds recommendation_type as the extensibility + hook for future KB/model recommendations. + """ + + job_id: UUID + status: str + recommendation_type: RecommendationTypeEnum = RecommendationTypeEnum.PROMPT + config_version: ConfigVersionPublic | None = None + error_message: str | None = None diff --git a/backend/app/services/evaluations/prompt_improvement.py b/backend/app/services/evaluations/prompt_improvement.py index eb7cfadef..970281faa 100644 --- a/backend/app/services/evaluations/prompt_improvement.py +++ b/backend/app/services/evaluations/prompt_improvement.py @@ -28,6 +28,11 @@ from app.core.storage_utils import load_json_from_object_store from app.crud.config.version import ConfigVersionCrud from app.crud.evaluations.core import get_evaluation_run_by_id +from app.crud.evaluations.score import ( + GROUND_TRUTH_SCORE_NAME, + KNOWLEDGE_BASE_SCORE_NAME, + PROMPT_SCORE_NAME, +) from app.crud.jobs import JobCrud from app.models.config.config import ConfigTag from app.models.config.version import ( @@ -35,7 +40,12 @@ ConfigVersionPublic, ConfigVersionUpdate, ) -from app.models.evaluation import EvaluationRun, PromptImprovementJobPublic +from app.models.evaluation import ( + EvaluationRun, + PromptImprovementJobPublic, + PromptRecommendationJobPublic, + RecommendationTypeEnum, +) from app.models.job import Job, JobStatus, JobType, JobUpdate from app.services.llm.providers.claude import ClaudeProvider from app.utils import APIResponse, get_webhook_secret, send_callback @@ -103,12 +113,16 @@ def validate_improve_prompt( evaluation_id: int, organization_id: int, project_id: int, + require_judge_run: bool = False, ) -> EvaluationRun: """Run the cheap DB precondition checks for prompt improvement. Raises HTTPException for every domain failure so the request-side caller returns a real 4xx before any job is enqueued. No LLM call, no trace download. Returns the run for reuse by the worker path. + + When require_judge_run is True (v2 callers), the run must be a judged run; + v1 callers leave it False and are unaffected. """ run = get_evaluation_run_by_id( session=session, @@ -148,6 +162,12 @@ def validate_improve_prompt( detail="source_config_unavailable: the run's config or config_version is missing or soft-deleted", ) + if require_judge_run and not run.is_judge_run: + raise HTTPException( + status_code=422, + detail="not_a_judge_run: v2 prompt improvement requires a judged (v2) evaluation run", + ) + return run @@ -158,16 +178,19 @@ def start_prompt_improvement_job( organization_id: int, project_id: int, callback_url: str, + require_judge_run: bool = False, ) -> Job: """Validate preconditions, create a job row, and enqueue the worker task. Returns the created Job immediately; the result is delivered to callback_url. + v2 callers pass require_judge_run=True to reject non-judged runs up front. """ validate_improve_prompt( session=session, evaluation_id=evaluation_id, organization_id=organization_id, project_id=project_id, + require_judge_run=require_judge_run, ) trace_id = correlation_id.get() or "N/A" @@ -220,19 +243,31 @@ def _build_improve_prompt_payload( job_id: UUID, config_version: ConfigVersionPublic | None, error_message: str | None, + is_judge_run: bool = False, ) -> dict: - """Build the callback body: PromptImprovementJobPublic inside an APIResponse. + """Build the callback body: the job-result model inside an APIResponse. - Pre-dump the inner model to JSON so no UUID/datetime survives into the dict - send_callback serialises. + Judge (v2) runs emit PromptRecommendationJobPublic (adds recommendation_type); + v1 runs emit PromptImprovementJobPublic byte-for-byte unchanged. Pre-dump the + inner model to JSON so no UUID/datetime survives into the dict send_callback + serialises. """ status = JobStatus.FAILED if error_message else JobStatus.SUCCESS - job_public = PromptImprovementJobPublic( - job_id=job_id, - status=status.value, - config_version=config_version, - error_message=error_message, - ).model_dump(mode="json") + if is_judge_run: + job_public = PromptRecommendationJobPublic( + job_id=job_id, + status=status.value, + recommendation_type=RecommendationTypeEnum.PROMPT, + config_version=config_version, + error_message=error_message, + ).model_dump(mode="json") + else: + job_public = PromptImprovementJobPublic( + job_id=job_id, + status=status.value, + config_version=config_version, + error_message=error_message, + ).model_dump(mode="json") envelope = ( APIResponse.failure_response(error=error_message, data=job_public) if error_message @@ -319,6 +354,14 @@ def execute_prompt_improvement( f"[execute_prompt_improvement] Redelivery of completed job, skipping | " f"job_id={job_id}" ) + # Reload the run so the redelivered callback keeps the same (v1 vs v2) + # shape as the original — is_judge_run decides which model is emitted. + redelivery_run = get_evaluation_run_by_id( + session=session, + evaluation_id=evaluation_id, + organization_id=organization_id, + project_id=project_id, + ) # At-least-once delivery: the first callback may have failed before the # worker died, so re-send. The client may receive a duplicate callback. _send_improve_prompt_callback( @@ -333,6 +376,7 @@ def execute_prompt_improvement( version_number=(existing.meta or {}).get("version"), ), error_message=None, + is_judge_run=bool(redelivery_run and redelivery_run.is_judge_run), ), project_id=project_id, organization_id=organization_id, @@ -343,6 +387,10 @@ def execute_prompt_improvement( job_uuid, JobUpdate(status=JobStatus.PROCESSING, task_id=task_id) ) + # Set before the try so the failure/timeout callbacks (which may fire + # before `run` is bound) still emit the correct v1/v2 payload shape. + is_judge_run = False + try: # Re-validate defensively: the run may have changed between enqueue and pickup. run = validate_improve_prompt( @@ -351,6 +399,7 @@ def execute_prompt_improvement( organization_id=organization_id, project_id=project_id, ) + is_judge_run = bool(run.is_judge_run) version = _resolve_source_version( session=session, run=run, project_id=project_id ) @@ -372,7 +421,10 @@ def execute_prompt_improvement( "trace_download_failed: could not retrieve trace file from storage" ) - improved_instructions, rationale = _draft_improved_prompt( + draft = ( + _draft_improved_prompt_v2 if is_judge_run else _draft_improved_prompt + ) + improved_instructions, rationale = draft( current_instructions=current_instructions, config_params=params, traces=traces, @@ -437,6 +489,7 @@ def execute_prompt_improvement( job_id=job_uuid, config_version=None, error_message=error_message, + is_judge_run=is_judge_run, ), project_id=project_id, organization_id=organization_id, @@ -461,6 +514,7 @@ def execute_prompt_improvement( job_id=job_uuid, config_version=None, error_message=error_message, + is_judge_run=is_judge_run, ), project_id=project_id, organization_id=organization_id, @@ -478,6 +532,7 @@ def execute_prompt_improvement( job_id=job_uuid, config_version=new_version_public, error_message=None, + is_judge_run=is_judge_run, ), project_id=project_id, organization_id=organization_id, @@ -485,14 +540,21 @@ def execute_prompt_improvement( return result -def _draft_improved_prompt( - *, - current_instructions: str, - config_params: dict, - traces: list | dict, -) -> tuple[str, str]: - """Call Claude with the current prompt + score traces and return - (improved_instructions, rationale). +def _target_config_from_params(config_params: dict) -> dict: + """Read-only config context shown to the model. + + instructions is rendered separately; knowledge_base_ids are opaque ids the + model can't act on, so both are stripped. + """ + excluded_keys = {"instructions", "knowledge_base_ids"} + return { + key: value for key, value in config_params.items() if key not in excluded_keys + } + + +def _call_prompt_drafting_llm(*, user_message_text: str) -> tuple[str, str]: + """Run the Anthropic structured-output call shared by both draft variants and + return (improved_instructions, rationale). Uses structured outputs so the first text block is guaranteed-valid JSON. Runs inside a Celery worker, so failures raise plain exceptions (no client @@ -508,32 +570,6 @@ def _draft_improved_prompt( client = ClaudeProvider.create_client({"api_key": settings.ANTHROPIC_API_KEY}) - # instructions is shown above already; knowledge_base_ids are opaque ids the model can't use. - excluded_keys = {"instructions", "knowledge_base_ids"} - target_config = { - key: value for key, value in config_params.items() if key not in excluded_keys - } - - user_message_text = ( - "You are a prompt engineer. Below is a JSON array of evaluation traces. " - "Each trace has the fields: `question`, `ground_truth_answer`, " - "`llm_answer`, `category`, and `scores` (a list of scoring objects with " - "`name`, `value`, and `unscoreable`).\n\n" - f"## Evaluation traces\n```\n{json.dumps(traces)}\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" - "1. Identify the answers that performed poorly — those with low scores or " - "where `llm_answer` diverges significantly from `ground_truth_answer`.\n" - "2. Rewrite the system prompt to improve those low-performing answers while " - "keeping what already works well.\n" - "3. Change ONLY the prompt text — do not alter the model, knowledge base, " - "or any other configuration.\n" - f"4. Return `{_LLM_KEY_RATIONALE}` as ONE concise sentence " - f"(≤ {_RATIONALE_MAX_LENGTH} characters): what you changed and why." - ) - try: response = client.messages.create( model=settings.PROMPT_IMPROVEMENT_MODEL, @@ -547,7 +583,7 @@ def _draft_improved_prompt( except anthropic.AuthenticationError: logger.warning( - "[_draft_improved_prompt] [ANTHROPIC] Authentication failed " + "[_call_prompt_drafting_llm] [ANTHROPIC] Authentication failed " "(code: 401): Verify the ANTHROPIC_API_KEY is " "valid, not expired, and configured correctly.", exc_info=True, @@ -559,7 +595,7 @@ def _draft_improved_prompt( except anthropic.RateLimitError: logger.warning( - "[_draft_improved_prompt] [ANTHROPIC] Rate limit exceeded " + "[_call_prompt_drafting_llm] [ANTHROPIC] Rate limit exceeded " "(code: 429): Hit Anthropic rate/quota — wait ≥1 min and retry.", exc_info=True, ) @@ -571,7 +607,7 @@ def _draft_improved_prompt( except anthropic.APITimeoutError: # Must come before APIConnectionError — APITimeoutError is a subclass. logger.error( - "[_draft_improved_prompt] [KAAPI] Anthropic request timed out " + "[_call_prompt_drafting_llm] [KAAPI] Anthropic request timed out " "(code: APITimeoutError): retry with a smaller payload.", exc_info=True, ) @@ -582,7 +618,7 @@ def _draft_improved_prompt( except anthropic.APIConnectionError: logger.error( - "[_draft_improved_prompt] [KAAPI] Anthropic connection failed " + "[_call_prompt_drafting_llm] [KAAPI] Anthropic connection failed " "(code: APIConnectionError): network or DNS issue reaching Anthropic.", exc_info=True, ) @@ -596,7 +632,7 @@ def _draft_improved_prompt( # 5xx is provider-side (alert-worthy); 4xx is caller's fault (noise if alerted) log = logger.error if status and status >= 500 else logger.warning log( - f"[_draft_improved_prompt] [ANTHROPIC] API status error " + f"[_call_prompt_drafting_llm] [ANTHROPIC] API status error " f"(code: {status}): {exc.message}.", exc_info=True, ) @@ -607,7 +643,7 @@ def _draft_improved_prompt( except Exception as exc: logger.error( - f"[_draft_improved_prompt] [KAAPI] Unexpected error during LLM call " + f"[_call_prompt_drafting_llm] [KAAPI] Unexpected error during LLM call " f"(code: {type(exc).__name__}): not raised by the Anthropic SDK — " f"likely a Kaapi-side failure. Contact Kaapi if persistent.", exc_info=True, @@ -616,3 +652,84 @@ def _draft_improved_prompt( "prompt_generation_failed: unexpected error during prompt generation — " "contact Kaapi if persistent" ) + + +def _draft_improved_prompt( + *, + current_instructions: str, + config_params: dict, + traces: list | dict, +) -> tuple[str, str]: + """v1 draft: rewrite the prompt from cosine/correctness score traces.""" + target_config = _target_config_from_params(config_params) + + user_message_text = ( + "You are a prompt engineer. Below is a JSON array of evaluation traces. " + "Each trace has the fields: `question`, `ground_truth_answer`, " + "`llm_answer`, `category`, and `scores` (a list of scoring objects with " + "`name`, `value`, and `unscoreable`).\n\n" + f"## Evaluation traces\n```\n{json.dumps(traces)}\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" + "1. Identify the answers that performed poorly — those with low scores or " + "where `llm_answer` diverges significantly from `ground_truth_answer`.\n" + "2. Rewrite the system prompt to improve those low-performing answers while " + "keeping what already works well.\n" + "3. Change ONLY the prompt text — do not alter the model, knowledge base, " + "or any other configuration.\n" + f"4. Return `{_LLM_KEY_RATIONALE}` as ONE concise sentence " + f"(≤ {_RATIONALE_MAX_LENGTH} characters): what you changed and why." + ) + + return _call_prompt_drafting_llm(user_message_text=user_message_text) + + +def _draft_improved_prompt_v2( + *, + current_instructions: str, + config_params: dict, + traces: list | dict, +) -> tuple[str, str]: + """v2 draft: rewrite the prompt from the three-metric judge results. + + Each trace's `scores` list holds the three Adherence-to-* metrics, each with a + 0-1 `value` (score) and a `comment` (the judge's reasoning). The model must read + both. Low Adherence to Knowledge Base is a retrieval/KB gap, not a prompt fault, + so the model is told not to chase grounding by editing the prompt. + """ + target_config = _target_config_from_params(config_params) + + user_message_text = ( + "You are a prompt engineer. Below is a JSON array of evaluation traces 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" + 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 from 0 (worst) to 1 (best). Metrics that " + 'could not be scored appear with `value` = "N/A" and `unscoreable` = true; ' + "ignore those.\n" + "- `comment` is the judge's reasoning for that score — read it, not just the " + "number, to understand *why* a row failed.\n\n" + f"## Evaluation traces\n```\n{json.dumps(traces)}\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" + f"1. Focus on rows where `{PROMPT_SCORE_NAME}` or `{GROUND_TRUTH_SCORE_NAME}` " + "is low. Use BOTH the `value` and the `comment` of each metric: the score " + "tells you how bad it is, the reasoning tells you what went wrong.\n" + "2. Rewrite the system prompt to fix those failures while keeping what already " + "works well.\n" + f"3. A low `{KNOWLEDGE_BASE_SCORE_NAME}` usually reflects a retrieval / " + "knowledge-base gap, NOT a prompt problem — do not try to fix grounding by " + "editing the prompt. You may add a light instruction to avoid unsupported " + "claims, but do not attempt to change the model, knowledge base, or config.\n" + "4. Change ONLY the prompt text.\n" + f"5. Return `{_LLM_KEY_RATIONALE}` as ONE concise sentence " + f"(≤ {_RATIONALE_MAX_LENGTH} characters): what you changed and why." + ) + + return _call_prompt_drafting_llm(user_message_text=user_message_text) diff --git a/backend/app/tests/api/routes/test_improve_prompt_v2.py b/backend/app/tests/api/routes/test_improve_prompt_v2.py new file mode 100644 index 000000000..20f891034 --- /dev/null +++ b/backend/app/tests/api/routes/test_improve_prompt_v2.py @@ -0,0 +1,670 @@ +"""Tests for the v2 prompt-improvement feature (judged runs, three-metric judge). + +The v2 endpoint is: + - POST /api/v2/evaluations/{id}/improve-prompt → 202 + a job handle. Requires a + JSON body {"callback_url": "https://..."} and a *judged* (is_judge_run) run; + calls start_prompt_improvement_job(..., require_judge_run=True). + - execute_prompt_improvement (shared with v1) → branches on run.is_judge_run: + a judge run drafts via _draft_improved_prompt_v2 and delivers a + PromptRecommendationJobPublic callback; a non-judge run keeps the v1 shape. + +HTTP boundaries mocked (patched as bound in the modules under test): +- ClaudeProvider.create_client (fake Anthropic client) +- get_cloud_storage / load_json_from_object_store (v2 judge traces) +- start_prompt_improvement (the Celery enqueue helper) — never touch a broker +- send_callback + get_webhook_secret — never make real outbound HTTP +- validate_callback_url at the v2 route call site +- Session (the worker opens its own Session(engine); redirected at the db fixture) + +DB is real (transactional db fixture; rolls back after each test). +""" + +import contextlib +import json +from contextlib import ExitStack +from dataclasses import dataclass +from typing import Any, Iterator +from unittest.mock import MagicMock, patch + +import pytest +from sqlmodel import Session + +from app.core.config import settings +from app.crud.config.version import ConfigVersionCrud +from app.crud.evaluations.score import ( + GROUND_TRUTH_SCORE_NAME, + KNOWLEDGE_BASE_SCORE_NAME, + PROMPT_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 ( + AI_GENERATED_MARKER, + execute_prompt_improvement, +) +from app.tests.utils.auth import TestAuthContext +from app.tests.utils.test_data import create_test_evaluation_dataset +from app.tests.utils.utils import random_lower_string + +_SERVICE = "app.services.evaluations.prompt_improvement" +_ROUTE_VALIDATE = ( + "app.api.routes.evaluations.prompt_improvement_v2.validate_callback_url" +) +POST_URL = f"{settings.API_V2_STR}/evaluations/{{evaluation_id}}/improve-prompt" + +_CALLBACK_URL = "https://example.com/callback" + +_IMPROVED_INSTRUCTIONS = "You are an improved assistant. Answer precisely." +_RATIONALE = "Tightened language adherence and grounding to fix weak rows." + +# The judge's reasoning strings — asserted verbatim in the v2 LLM message, so kept +# as named constants and reused in the trace fixture (single independent source). +_GT_COMMENT = "missed key fact X" +_PROMPT_COMMENT = "answered in wrong language" +_KB_COMMENT = "Knowledge base not queried." + +# v2 judge traces: three Adherence-to-* metrics, KB unscoreable (value "N/A"). +_JUDGE_TRACES: list[dict] = [ + { + "trace_id": "t1", + "question": "Q", + "llm_answer": "A", + "ground_truth_answer": "G", + "question_id": "q1", + "scores": [ + { + "name": GROUND_TRUTH_SCORE_NAME, + "value": 0.4, + "data_type": "NUMERIC", + "comment": _GT_COMMENT, + }, + { + "name": PROMPT_SCORE_NAME, + "value": 0.3, + "data_type": "NUMERIC", + "comment": _PROMPT_COMMENT, + }, + { + "name": KNOWLEDGE_BASE_SCORE_NAME, + "value": "N/A", + "data_type": "CATEGORICAL", + "comment": _KB_COMMENT, + "unscoreable": True, + }, + ], + } +] + +_AUTO_URL = object() # "generate a valid score_trace_url from the run id after commit" + + +def _llm_json( + improved_instructions: str = _IMPROVED_INSTRUCTIONS, + rationale: str = _RATIONALE, +) -> str: + return json.dumps( + {"improved_instructions": improved_instructions, "rationale": rationale} + ) + + +def _make_fake_claude_client(text_content: str | None = None) -> MagicMock: + if text_content is None: + text_content = _llm_json() + + content_block = MagicMock() + content_block.type = "text" + content_block.text = text_content + + response = MagicMock() + response.content = [content_block] + response.id = "msg_test_id" + + client = MagicMock() + client.messages.create.return_value = response + return client + + +@dataclass +class WorkerEnv: + """Handles yielded by ``_worker_env``: the LLM boundary and the callback sink.""" + + llm: MagicMock + callback: MagicMock + + +@contextlib.contextmanager +def _worker_env( + db: Session, + *, + draft_v2: MagicMock | None = None, + claude_client: MagicMock | None = None, + traces: Any = _JUDGE_TRACES, +) -> Iterator[WorkerEnv]: + """Redirect the worker's Session(engine) at the test db and mock its boundaries. + + Default: the fake Claude client, so the real _draft_improved_prompt_v2 runs and + its user message is inspectable. Pass ``draft_v2`` to stub the v2 draft wholesale + (used to assert it is NOT re-called on redelivery). ``env.callback`` is the + patched send_callback — no real HTTP ever. + """ + session_cm = MagicMock() + session_cm.__enter__.return_value = db + session_cm.__exit__.return_value = None + + with ExitStack() as stack: + stack.enter_context(patch(f"{_SERVICE}.Session", return_value=session_cm)) + stack.enter_context( + patch(f"{_SERVICE}.get_cloud_storage", return_value=MagicMock()) + ) + stack.enter_context( + patch(f"{_SERVICE}.load_json_from_object_store", return_value=traces) + ) + stack.enter_context(patch(f"{_SERVICE}.get_webhook_secret", return_value=None)) + callback = stack.enter_context(patch(f"{_SERVICE}.send_callback")) + + if draft_v2 is not None: + stack.enter_context( + patch(f"{_SERVICE}._draft_improved_prompt_v2", draft_v2) + ) + yield WorkerEnv(llm=draft_v2, callback=callback) + else: + if claude_client is None: + claude_client = _make_fake_claude_client() + stack.enter_context( + patch( + f"{_SERVICE}.ClaudeProvider.create_client", + return_value=claude_client, + ) + ) + yield WorkerEnv(llm=claude_client, callback=callback) + + +def _callback_payload(callback: MagicMock) -> dict: + """Single send_callback invocation → the APIResponse envelope it POSTed.""" + assert callback.call_count == 1 + return callback.call_args.args[1] + + +def _llm_user_message(claude_client: MagicMock) -> str: + """The single user-message text passed to the mocked Anthropic create().""" + messages = claude_client.messages.create.call_args.kwargs["messages"] + return messages[0]["content"] + + +def _make_config_with_instructions( + db: Session, + project_id: int, + instructions: str = "You are a helpful assistant.", +) -> Any: + from app.crud.config import ConfigCrud + from app.models.config.config import ConfigCreate + from app.models.llm import KaapiCompletionConfig + from app.models.llm.request import ConfigBlob + + config_blob = ConfigBlob( + completion=KaapiCompletionConfig( + provider="openai", + type="text", + params={ + "model": "gpt-4o", + "temperature": 0.5, + "instructions": instructions, + "knowledge_base_ids": ["vs_abc123"], + }, + ) + ) + config_create = ConfigCreate( + name=f"test-config-{random_lower_string()}", + description="Test configuration for improve-prompt v2", + config_blob=config_blob, + commit_message="Initial version", + tag=ConfigTag.DEFAULT, + ) + config, _ = ConfigCrud(session=db, project_id=project_id).create_or_raise( + config_create + ) + return config + + +def _make_run( + db: Session, + config_id: Any, + config_version: int | None, + organization_id: int, + project_id: int, + dataset_id: int, + *, + is_judge_run: bool | None, + status: str = "completed", + run_name: str | None = None, + score_trace_url: Any = _AUTO_URL, +) -> EvaluationRun: + if run_name is None: + run_name = f"run-{random_lower_string()}" + + initial_url: str | None = None if score_trace_url is _AUTO_URL else score_trace_url + + run = EvaluationRun( + run_name=run_name, + dataset_name=f"ds-{random_lower_string()}", + dataset_id=dataset_id, + config_id=config_id, + config_version=config_version, + status=status, + total_items=1, + type="text", + organization_id=organization_id, + project_id=project_id, + score_trace_url=initial_url, + is_judge_run=is_judge_run, + ) + db.add(run) + db.commit() + db.refresh(run) + + if score_trace_url is _AUTO_URL: + run.score_trace_url = ( + f"s3://test-bucket/evaluations/score/{run.id}/traces_{run.id}.json" + ) + db.add(run) + db.commit() + db.refresh(run) + + return run + + +@pytest.fixture +def auth(user_api_key: TestAuthContext) -> TestAuthContext: + return user_api_key + + +@pytest.fixture +def headers(user_api_key_header: dict[str, str]) -> dict[str, str]: + return user_api_key_header + + +@pytest.fixture +def dataset(db: Session, auth: TestAuthContext) -> EvaluationDataset: + return create_test_evaluation_dataset( + db=db, + organization_id=auth.organization_id, + project_id=auth.project_id, + ) + + +@pytest.fixture +def config_with_instructions(db: Session, auth: TestAuthContext) -> Any: + return _make_config_with_instructions( + db=db, + project_id=auth.project_id, + instructions="You are a helpful assistant. Answer clearly.", + ) + + +@pytest.fixture +def anthropic_creds(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + settings, "ANTHROPIC_API_KEY", "sk-ant-test-" + random_lower_string() + ) + + +@pytest.fixture +def judge_run( + db: Session, + auth: TestAuthContext, + dataset: EvaluationDataset, + config_with_instructions: Any, + anthropic_creds: None, +) -> EvaluationRun: + return _make_run( + db=db, + config_id=config_with_instructions.id, + config_version=1, + organization_id=auth.organization_id, + project_id=auth.project_id, + dataset_id=dataset.id, + is_judge_run=True, + ) + + +@pytest.fixture +def non_judge_run( + db: Session, + auth: TestAuthContext, + dataset: EvaluationDataset, + config_with_instructions: Any, + anthropic_creds: None, +) -> EvaluationRun: + return _make_run( + db=db, + config_id=config_with_instructions.id, + config_version=1, + organization_id=auth.organization_id, + project_id=auth.project_id, + dataset_id=dataset.id, + is_judge_run=False, + ) + + +class TestV2Route: + """POST /api/v2/.../improve-prompt: judged-run gate + request-side validation.""" + + def test_judge_run_returns_202_with_job_handle_and_enqueues( + self, + client: Any, + headers: dict[str, str], + db: Session, + auth: TestAuthContext, + judge_run: EvaluationRun, + ) -> None: + with patch(_ROUTE_VALIDATE), patch( + f"{_SERVICE}.start_prompt_improvement", return_value="task-v2-1" + ) as enqueue: + resp = client.post( + POST_URL.format(evaluation_id=judge_run.id), + headers=headers, + json={"callback_url": _CALLBACK_URL}, + ) + + assert resp.status_code == 202, resp.text + body = resp.json()["data"] + assert body["status"] == JobStatus.PENDING.value + job_id = body["job_id"] + + job = JobCrud(session=db).get(job_id=job_id, project_id=auth.project_id) + assert job is not None + assert job.job_type == JobType.PROMPT_IMPROVEMENT + + enqueue.assert_called_once() + kwargs = enqueue.call_args.kwargs + assert kwargs["job_id"] == job_id + assert kwargs["evaluation_id"] == judge_run.id + assert kwargs["callback_url"] == _CALLBACK_URL + + def test_non_judge_run_returns_422_not_a_judge_run( + self, + client: Any, + headers: dict[str, str], + non_judge_run: EvaluationRun, + ) -> None: + with patch(_ROUTE_VALIDATE), patch( + f"{_SERVICE}.start_prompt_improvement" + ) as enqueue: + resp = client.post( + POST_URL.format(evaluation_id=non_judge_run.id), + headers=headers, + json={"callback_url": _CALLBACK_URL}, + ) + + assert resp.status_code == 422, resp.text + assert "not_a_judge_run" in resp.json()["error"], resp.text + enqueue.assert_not_called() + + def test_missing_run_returns_404( + self, + client: Any, + headers: dict[str, str], + ) -> None: + with patch(_ROUTE_VALIDATE), patch( + f"{_SERVICE}.start_prompt_improvement" + ) as enqueue: + resp = client.post( + POST_URL.format(evaluation_id=9999999), + headers=headers, + json={"callback_url": _CALLBACK_URL}, + ) + assert resp.status_code == 404, resp.text + assert "evaluation_not_found" in resp.json()["error"], resp.text + enqueue.assert_not_called() + + def test_non_completed_run_returns_409( + self, + client: Any, + headers: dict[str, str], + db: Session, + auth: TestAuthContext, + dataset: EvaluationDataset, + config_with_instructions: Any, + ) -> None: + run = _make_run( + db=db, + config_id=config_with_instructions.id, + config_version=1, + organization_id=auth.organization_id, + project_id=auth.project_id, + dataset_id=dataset.id, + is_judge_run=True, + status="processing", + ) + with patch(_ROUTE_VALIDATE), patch( + f"{_SERVICE}.start_prompt_improvement" + ) as enqueue: + resp = client.post( + POST_URL.format(evaluation_id=run.id), + headers=headers, + json={"callback_url": _CALLBACK_URL}, + ) + assert resp.status_code == 409, resp.text + assert "evaluation_not_completed" in resp.json()["error"], resp.text + enqueue.assert_not_called() + + def test_missing_trace_url_returns_422( + self, + client: Any, + headers: dict[str, str], + db: Session, + auth: TestAuthContext, + dataset: EvaluationDataset, + config_with_instructions: Any, + ) -> None: + run = _make_run( + db=db, + config_id=config_with_instructions.id, + config_version=1, + organization_id=auth.organization_id, + project_id=auth.project_id, + dataset_id=dataset.id, + is_judge_run=True, + score_trace_url=None, + ) + with patch(_ROUTE_VALIDATE), patch( + f"{_SERVICE}.start_prompt_improvement" + ) as enqueue: + resp = client.post( + POST_URL.format(evaluation_id=run.id), + headers=headers, + json={"callback_url": _CALLBACK_URL}, + ) + assert resp.status_code == 422, resp.text + assert "traces_not_available" in resp.json()["error"], resp.text + enqueue.assert_not_called() + + @pytest.mark.parametrize( + "callback_url", + [ + "http://example.com/callback", # non-HTTPS scheme + "https://127.0.0.1/callback", # loopback (SSRF) + ], + ) + def test_invalid_callback_url_rejected_by_ssrf_guard( + self, + callback_url: str, + client: Any, + headers: dict[str, str], + judge_run: EvaluationRun, + ) -> None: + with patch(f"{_SERVICE}.start_prompt_improvement") as enqueue: + resp = client.post( + POST_URL.format(evaluation_id=judge_run.id), + headers=headers, + json={"callback_url": callback_url}, + ) + assert resp.status_code == 422, resp.text + assert "invalid_callback_url" in resp.json()["error"], resp.text + enqueue.assert_not_called() + + +class TestV2Worker: + """execute_prompt_improvement branches on is_judge_run: v2 draft + v2 callback.""" + + 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_judge_run_mints_version_and_sends_recommendation_callback( + self, + db: Session, + auth: TestAuthContext, + judge_run: EvaluationRun, + ) -> None: + job = self._pending_job(db, auth.project_id) + + with _worker_env(db) as env: + result = 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, + ) + + assert result["success"] is True + + v2 = ConfigVersionCrud( + session=db, config_id=judge_run.config_id, project_id=auth.project_id + ).read_one(version_number=2) + assert v2 is not None + assert ( + v2.config_blob["completion"]["params"]["instructions"] + == _IMPROVED_INSTRUCTIONS + ) + assert v2.commit_message.startswith(AI_GENERATED_MARKER) + + payload = _callback_payload(env.callback) + assert payload["success"] is True + assert payload["data"]["status"] == JobStatus.SUCCESS.value + assert payload["data"]["recommendation_type"] == "prompt" + assert payload["data"]["config_version"]["version"] == 2 + + def test_v2_message_carries_metric_reasoning_and_ignores_na( + self, + db: Session, + auth: TestAuthContext, + judge_run: EvaluationRun, + ) -> None: + job = self._pending_job(db, auth.project_id) + + with _worker_env(db) 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) + + for metric_name in ( + GROUND_TRUTH_SCORE_NAME, + PROMPT_SCORE_NAME, + KNOWLEDGE_BASE_SCORE_NAME, + ): + assert metric_name in message + + assert _GT_COMMENT in message + assert _PROMPT_COMMENT in message + + # The unscoreable KB metric must be presented as ignore-worthy, not a real + # score: the prompt tells the model to skip value="N/A" / unscoreable metrics. + assert '"N/A"' in message + assert "unscoreable" in message + assert "ignore those" in message + + def test_non_judge_run_uses_v1_shape_no_recommendation_type( + self, + db: Session, + auth: TestAuthContext, + non_judge_run: EvaluationRun, + ) -> None: + job = self._pending_job(db, auth.project_id) + + # v1 traces carry cosine scores; feed those so the non-judge path is realistic. + v1_traces = [ + { + "trace_id": "t1", + "question": "Q", + "llm_answer": "A", + "ground_truth_answer": "G", + "category": "Geo", + "scores": [ + {"name": "cosine_similarity", "value": 0.3, "unscoreable": False} + ], + } + ] + with _worker_env(db, traces=v1_traces) as env: + execute_prompt_improvement( + project_id=auth.project_id, + job_id=str(job.id), + organization_id=auth.organization_id, + evaluation_id=non_judge_run.id, + callback_url=_CALLBACK_URL, + ) + + payload = _callback_payload(env.callback) + assert payload["success"] is True + assert "recommendation_type" not in payload["data"] + + # v1 draft message does not surface the three-metric judge names. + message = _llm_user_message(env.llm) + assert PROMPT_SCORE_NAME not in message + assert GROUND_TRUTH_SCORE_NAME not in message + + def test_redelivery_of_success_judge_job_resends_v2_callback_without_new_version( + self, + db: Session, + auth: TestAuthContext, + judge_run: EvaluationRun, + ) -> None: + job = self._pending_job(db, auth.project_id) + + with _worker_env(db) 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, + ) + assert env.llm.messages.create.call_count == 1 + + crud = ConfigVersionCrud( + session=db, config_id=judge_run.config_id, project_id=auth.project_id + ) + first_version = db.get(Job, job.id).meta["version"] + assert crud.read_one(version_number=first_version + 1) is None + + draft = MagicMock(side_effect=AssertionError("v2 draft must not be re-called")) + with _worker_env(db, draft_v2=draft) as env: + result = 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, + ) + + draft.assert_not_called() + assert result["success"] is True + assert result["version"] == first_version + assert crud.read_one(version_number=first_version + 1) is None + + payload = _callback_payload(env.callback) + assert payload["success"] is True + assert payload["data"]["recommendation_type"] == "prompt" + assert payload["data"]["config_version"]["version"] == first_version diff --git a/docs/wiki/modules/evaluations.md b/docs/wiki/modules/evaluations.md index 0e7446cef..4e4484356 100644 --- a/docs/wiki/modules/evaluations.md +++ b/docs/wiki/modules/evaluations.md @@ -9,6 +9,7 @@ All paths relative to `backend/app/`. - `api/routes/evaluations/dataset.py`, `api/routes/evaluations/evaluation.py` — text datasets + runs (v1, `/api/v1`) - `api/routes/evaluations/evaluation_v2.py` — `POST /api/v2/evaluations`, replica of v1 run trigger + native ground-truth LLM judge (Langfuse-free); mounted under `settings.API_V2_STR` - `api/routes/evaluations/dataset_v2.py` — `POST /api/v2/evaluations/datasets`, Langfuse-free dataset upload; stores only the original CSV in S3 and records `duplication_factor` as metadata (rows expanded ×factor at run time, not physically duplicated) +- `api/routes/evaluations/prompt_improvement_v2.py` — `POST /api/v2/evaluations/{evaluation_id}/improve-prompt`, prompt iteration off the three-metric judge results (requires an `is_judge_run` run); same body as v1, returns a recommendation of type `prompt` - `api/routes/stt_evaluations/`, `api/routes/tts_evaluations/` — STT/TTS - `api/routes/cron.py` — batch polling trigger @@ -33,6 +34,7 @@ v2 judge field on `EvaluationRun`: `is_judge_run` (bool marker gating native jud - 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. - 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. 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_v2`, using each metric's score + reasoning) and delivers a `PromptRecommendationJobPublic` callback carrying `recommendation_type` (`RecommendationTypeEnum`, `models/evaluation.py`; only `PROMPT` today, knowledge-base / model recommendations deferred). Non-judge (v1) runs keep the `_draft_improved_prompt` + `PromptImprovementJobPublic` path unchanged. ## External - OpenAI Batch + embeddings, Gemini Batch, Langfuse (per-trace scores + summary), object storage (datasets/audio), kaapi-frontend console (results + annotation). From dccc7a3fc602bc14234c14c91146a43baa991b06 Mon Sep 17 00:00:00 2001 From: Ayush8923 <80516839+Ayush8923@users.noreply.github.com> Date: Fri, 24 Jul 2026 21:18:07 +0530 Subject: [PATCH 2/2] fix(evals): cleanups --- .../app/api/routes/evaluations/prompt_improvement_v2.py | 7 ------- 1 file changed, 7 deletions(-) diff --git a/backend/app/api/routes/evaluations/prompt_improvement_v2.py b/backend/app/api/routes/evaluations/prompt_improvement_v2.py index bf450e236..e65f72acb 100644 --- a/backend/app/api/routes/evaluations/prompt_improvement_v2.py +++ b/backend/app/api/routes/evaluations/prompt_improvement_v2.py @@ -34,13 +34,6 @@ def improve_evaluation_prompt_v2( request: ImprovePromptRequest, ) -> APIResponse[LLMJobImmediatePublic]: """Enqueue a v2 prompt-recommendation job for a judged evaluation run.""" - logger.info( - f"[improve_evaluation_prompt_v2] Starting | evaluation_id={evaluation_id} " - f"org_id={auth_context.organization_.id} project_id={auth_context.project_.id}" - ) - - # No global ValueError handler exists, so map the SSRF guard's ValueError to - # a 422 here rather than letting it surface as a 500. try: validate_callback_url(str(request.callback_url)) except ValueError as exc: