refactor(evaluation): split modules - #1202
Merged
Merged
Conversation
Structural refactor. Existing tests pass unchanged (3310 passed, 6 skipped)
and pyright reports the same 8 pre-existing errors as main.
Motivation: `crud/evaluations/fast.py` had grown to 1429 lines around a
375-line `_stage3_score_and_trace` mixing v1 cosine and v2 judge scoring,
trace building, cost attachment and the run rollup in one body.
Extracted, component-wise:
crud/evaluations/
fast_results.py pure per-item S3 unit shapes, usage sums, threshold
fast_chunks.py batch_job bookkeeping + chunk queries/cleanup
fast_cosine.py pure v1 cosine scoring
fast_traces.py pure TraceData building
judge_stage.py run-level v2 judge orchestration + metric rollup
judge_prompts.py rubric text, moved out of score.py
retry.py OpenAI retry policy, was duplicated in fast + judge
services/evaluations/
iteration_state.py graph state + pure round/stop arithmetic
iteration_checkpointer.py psycopg pool + PostgresSaver
`score.py` no longer carries judge rubric text, so `judge.py` imports its
prompts directly instead of back through the score module.
LOC: fast.py 1429->1006, score.py 359->218, iteration_graph.py 612->504.
Largest function 375->115 lines.
Two deliberate behaviour deltas, both narrowing existing risk:
- `_effective_duplication_factor` skips the `get_dataset_by_id` query when
the run already carries a factor. Same value, one fewer query.
- `_cleanup_response_chunks` now guards storage resolution as well as the
deletes. It runs between the completed transition and `save_score`, so a
raise there would flip a completed run to failed *and* lose its score
unit; main only guarded the delete loop.
fast.py stays above a 400-line target: stage 1/2 IO is pinned there by
existing tests patching `app.crud.evaluations.fast.*`, and moving it would
mean editing those tests.
New tests cover the extracted pure functions and the cleanup guard only
(71 cases); no existing test file was touched.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Two reliability fixes on the v2 (is_judge_run) eval path. Behaviour changes on
purpose; no existing test edited.
R2 — score unit persisted before the completed transition
---------------------------------------------------------
`run_fast_evaluation` marked a run `completed` and only then called
`save_score`, with chunk cleanup and the Langfuse sync in between. A crash in
that window (OOM, SIGKILL, hard timeout) left the run reading `completed` with
its score unit never written, and nothing recovers it: the aggregate
short-circuits on `status == "completed"` and the cron barrier only selects
`processing`. On v2 that unit is the *only* copy of the per-row judge scores and
reasoning — no Langfuse traces are created for a judged run, so unlike v1 there
is no second copy to re-fetch and merge from.
`save_score` now runs first (stage 4), the completed transition second (stage 5),
and cleanup + Langfuse sync last as a best-effort tail. A failure now leaves the
run `processing` and visibly unfinished instead of silently scoreless. Note this
does not make it retryable — a `failed` run is not re-enqueued either — the win
is that the loss is visible rather than silent.
Two follow-on changes:
* the completed transition no longer rewrites `score`; save_score already
persisted it, and the duplicate write differed for v1 (it carried
`overall: None`). Final DB state is unchanged from before.
* `save_score` returning None (run row gone) now raises instead of completing
the run without a score.
`save_score` opens a second Session on the same row, so stage 3's write must stay
committed before it — noted in a comment, since nothing else enforces it.
R4 — in-flight and zombie guards on the iteration resume tick
-------------------------------------------------------------
`dispatch_pending_evaluation_iteration_resumes` fanned a `resume=True` graph step
out to every PROCESSING loop every tick, unconditionally. A step slower than the
5-minute tick got a second one dispatched on top of it — two workers against the
same LangGraph checkpoint thread, and a duplicate eval run or improvement job
charged for. A row whose sub-job wedged collected resumes forever, with no max age.
Adds `evaluation_iteration_run.last_dispatched_at` (migration 083, nullable, no
backfill — NULL means never dispatched, which is what a fresh loop needs) and two
guards:
* dispatch inside EVAL_ITERATION_DISPATCH_COOLDOWN_MINUTES (10, floored on
CELERY_TASK_TIME_LIMIT plus queue wait) → skip,
* still processing past EVAL_ITERATION_STALL_THRESHOLD_HOURS (72) since kickoff
→ reap via `mark_iteration_run_failed`, failure callback included.
The cooldown is a timestamp heuristic, not a lock; the ceiling and its upgrade
path are named in a `ponytail:` comment. The reap threshold is deliberately
generous because a false positive kills a live loop and POSTs a failure webhook —
the genuinely unbounded case it exists for is a loop interrupted on an eval run
wedged in `processing` (audit finding R3, not fixed here); a crashed graph step
already fails itself, and a SIGKILLed one resumes from its checkpoint.
`_mark_iteration_run_failed` is now public (`mark_iteration_run_failed`) so the
reaper can call it. The summary dict keeps exactly its two keys — skip and reap
counts go to the log, since `test_cron_iteration.py` asserts it by equality.
Verification
------------
* 3322 passed, 6 skipped; zero existing test files edited (the one touched
test file is the docstring of a test added in the previous commit, whose
stated rationale referenced the old ordering)
* new tests mutation-checked: reverting the reorder fails 5/5, removing the
cron guards fails 2/7
* pyright: 10 errors on the changed set, all pre-existing
* ruff + pre-commit clean; `docs/wiki/modules/evaluations.md` updated
Not in scope: R1 (no failure threshold on the judge stage), R3 (no healer for a
run stuck post-merge), and the v2 read path routing judged runs through Langfuse.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
|
Important Review skippedAuto reviews are limited based on label configuration. 🏷️ Required labels (at least one) (1)
Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Repository: ProjectTech4DevAI/kaapi-backend/.coderabbit.yaml Review profile: CHILL Plan: Advanced Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Resolve docs/wiki/modules/evaluations.md by keeping both sides: this branch's crud/evaluations file split and score-persist ordering, main's plain-text ai_summary rationale and prompt-improvement stop_reason guard. Renumber 083_add_last_dispatched_at_to_evaluation_iteration_run to 084 so it chains after main's 083_add_llm_call_metadata (single alembic head). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
OpenAPI changes 🟢 1 non-breaking changeTip Safe to merge from an API-contract perspective. Full changelog ·
|
| Method | Path | Change | |
|---|---|---|---|
| 🟢 | POST |
/api/v2/evaluations/iterations |
api operation id Evaluation v2-create_evaluation_iteration_v2 removed and replaced with Evaluation v2-create_evaluation_iteration |
main ↔ 00efcf97 · generated by oasdiff
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
Ayush8923
reviewed
Sep 15, 2026
Comment on lines
+584
to
+585
| response_results: list[dict[str, Any]], | ||
| embedding_results: list[dict[str, Any]] | None, |
Collaborator
There was a problem hiding this comment.
if we know the response result structure, then remove the Any type and use the proper type safety check.
Comment on lines
+621
to
+622
| response_results: list[dict[str, Any]], | ||
| embedding_results: list[dict[str, Any]] | None, |
Collaborator
There was a problem hiding this comment.
here too, just check everywhere where you have added the Any.
Comment on lines
148
to
153
| logger.info( | ||
| f"[wait_eval_node] Round scored | iteration_run_id={state['iteration_run_id']} | " | ||
| f"round_number={state['round_number']} | stop_score={stop_score} | " | ||
| f"consecutive_low_delta_rounds={consecutive_low_delta_rounds} | " | ||
| f"consecutive_low_delta_rounds={update['consecutive_low_delta_rounds']} | " | ||
| f"stop_reason={update.get('stop_reason')}" | ||
| ) |
Collaborator
There was a problem hiding this comment.
these logger info and warning not needed everywhere, add these only if we know this is breakage point.
The cron zombie reaper (mark_iteration_run_failed) and the graph's own finalize_node both write the iteration run's terminal status. The reaper already checks status == PROCESSING before writing; finalize_node did not, so an in-flight step finishing after a reap overwrote FAILED with COMPLETED and sent a second, contradictory callback. finalize_node now no-ops when the row is no longer PROCESSING. First terminal writer wins, exactly one callback goes out. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
AkhileshNegi
changed the title
Trim multiline comments to single-line essential rationale for EVAL_ITERATION_DISPATCH_COOLDOWN_MINUTES and EVAL_ITERATION_STALL_THRESHOLD_HOURS.
AkhileshNegi
marked this pull request as ready for review
September 15, 2026 13:30
AkhileshNegi
changed the title
Ayush8923
approved these changes
Sep 16, 2026
Ayush8923
left a comment
Ayush8923
left a comment
Collaborator
There was a problem hiding this comment.
good to go. but added few comments for cleanups perspective.
| from app.services.evaluations.iteration_graph import mark_iteration_run_failed | ||
|
|
||
| runs = list_processing_evaluation_iteration_runs(session=session) | ||
| now_ = now() |
Collaborator
There was a problem hiding this comment.
can we simply write now instead of the now_?
Comment on lines
+147
to
+152
| def _log_prefix(eval_run: EvaluationRun) -> str: | ||
| return ( | ||
| f"[org={eval_run.organization_id}]" | ||
| f"[project={eval_run.project_id}]" | ||
| f"[eval={eval_run.id}]" | ||
| ) |
Collaborator
There was a problem hiding this comment.
this is good, if we actually need logs then create the one helper function and use that function everywhere if needed instead of importing the logger in every file.
| delete_batch_job, | ||
| get_batch_job, | ||
| ) | ||
| from app.crud.job import get_batch_job |
Collaborator
There was a problem hiding this comment.
is not used anywhere. please check.
| session: Session, | ||
| eval_run: EvaluationRun, | ||
| job_type: str, | ||
| config: dict[str, Any], |
Collaborator
There was a problem hiding this comment.
let's try to avoid the any type.
Collaborator
There was a problem hiding this comment.
on the try safety checks issue we have remove all the any type and add the proper type safety checks.
Comment on lines
+61
to
+153
| def _create_stage_job( | ||
| *, | ||
| session: Session, | ||
| eval_run: EvaluationRun, | ||
| job_type: str, | ||
| config: dict[str, Any], | ||
| raw_output_url: str | None, | ||
| total_items: int, | ||
| ) -> BatchJob: | ||
| return create_batch_job( | ||
| session=session, | ||
| batch_job_create=BatchJobCreate( | ||
| provider="openai", | ||
| job_type=job_type, | ||
| config={"run_mode": RunModeEnum.FAST.value, **config}, | ||
| raw_output_url=raw_output_url, | ||
| total_items=total_items, | ||
| organization_id=eval_run.organization_id, | ||
| project_id=eval_run.project_id, | ||
| ), | ||
| ) | ||
|
|
||
|
|
||
| def create_response_chunk_job( | ||
| *, | ||
| session: Session, | ||
| eval_run: EvaluationRun, | ||
| chunk_index: int, | ||
| model: str | None, | ||
| results: list[dict[str, Any]], | ||
| raw_output_url: str | None, | ||
| ) -> BatchJob: | ||
| """Mark one response chunk done.""" | ||
| return _create_stage_job( | ||
| session=session, | ||
| eval_run=eval_run, | ||
| job_type=JOB_TYPE_EVALUATION_FAST_CHUNK, | ||
| config={ | ||
| "endpoint": _RESPONSES_ENDPOINT, | ||
| "model": model, | ||
| "usage": sum_usage(results, RESPONSE_USAGE_KEYS), | ||
| CHUNK_CONFIG_RUN_ID: eval_run.id, | ||
| CHUNK_CONFIG_INDEX: chunk_index, | ||
| }, | ||
| raw_output_url=raw_output_url, | ||
| total_items=len(results), | ||
| ) | ||
|
|
||
|
|
||
| def create_merged_response_job( | ||
| *, | ||
| session: Session, | ||
| eval_run: EvaluationRun, | ||
| model: str | None, | ||
| results: list[dict[str, Any]], | ||
| raw_output_url: str | None, | ||
| ) -> BatchJob: | ||
| """Mark the merged responses unit done; its id is the aggregate's retry guard.""" | ||
| return _create_stage_job( | ||
| session=session, | ||
| eval_run=eval_run, | ||
| job_type=JOB_TYPE_EVALUATION_FAST, | ||
| config={ | ||
| "endpoint": _RESPONSES_ENDPOINT, | ||
| "model": model, | ||
| "usage": sum_usage(results, RESPONSE_USAGE_KEYS), | ||
| }, | ||
| raw_output_url=raw_output_url, | ||
| total_items=len(results), | ||
| ) | ||
|
|
||
|
|
||
| def create_embedding_job( | ||
| *, | ||
| session: Session, | ||
| eval_run: EvaluationRun, | ||
| embedding_model: str, | ||
| results: list[dict[str, Any]], | ||
| raw_output_url: str | None, | ||
| ) -> BatchJob: | ||
| """Mark the embeddings unit done.""" | ||
| return _create_stage_job( | ||
| session=session, | ||
| eval_run=eval_run, | ||
| job_type=JOB_TYPE_EMBEDDING_FAST, | ||
| config={ | ||
| "endpoint": _EMBEDDINGS_ENDPOINT, | ||
| "embedding_model": embedding_model, | ||
| "usage": sum_usage(results, EMBEDDING_USAGE_KEYS), | ||
| }, | ||
| raw_output_url=raw_output_url, | ||
| total_items=len(results), | ||
| ) |
Collaborator
There was a problem hiding this comment.
from what i have seen, these functions are basically wrapping another function inside them. We could directly use/import the child function wherever it’s needed instead of creating a separate function that simply returns another function.
this would also reduce the amount of code in this file and make it a bit cleaner.
Collaborator
There was a problem hiding this comment.
and make this _create_stage_job publically accessible instead of the private function.
Comment on lines
+170
to
+173
| logger.info( | ||
| f"[delete_response_chunk_artifacts] Removed {len(chunk_jobs)} chunk " | ||
| f"artifacts | eval_run_id={eval_run_id}" | ||
| ) |
Comment on lines
+34
to
+37
|
|
||
| Stamped `last_dispatched_at` up front: kickoff enqueues the first graph step | ||
| right after this returns, and the cron's cooldown has to cover that step like | ||
| any other, or the first tick races it on the same checkpoint thread. |
Comment on lines
+275
to
+283
| if iteration_run.status != EvaluationIterationStatusEnum.PROCESSING: | ||
| # Reaper (or another terminal writer) got here first; its callback | ||
| # already went out, so a second one would contradict it. | ||
| logger.warning( | ||
| f"[finalize_node] Row already terminal, skipping | " | ||
| f"iteration_run_id={state['iteration_run_id']} | " | ||
| f"status={iteration_run.status.value}" | ||
| ) | ||
| return {} |
Collaborator
There was a problem hiding this comment.
will this actually be useful? we mostly check the data directly from the DB, and it’s very rare that we would check this warning in the logs.
If this is going to remain mostly redundant code, I think we can remove it. As far as I understand, we’re already checking the status in the DB, and here we’re just returning a null object.
If you want, I can  make it sound a bit more concise and natural for a GitHub PR comment.
- expose create_stage_job and drop the three thin wrappers around it - share build_log_prefix from core instead of repeating the f-string - trim progress-only logs in the iteration graph and chunk cleanup - shorten comments/docstrings, rename now_ to current_time Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Drop the _v2 filename suffix in favour of routes/evaluations/v2/ and docs/evaluation/v2/ so v1 and v2 files mirror each other by name. Aggregate the four v2 routers in v2/__init__.py so main.py mounts one router instead of four. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
…ived total_items - Merge v2 markers (is_judge_run, callback_url, duplication_factor) and status/total_items into the initial create_evaluation_run INSERT instead of follow-up UPDATEs - Compute total_items directly from dataset_metadata without fetching items: original_items_count * effective_duplication_factor - Drops item-list fetch from trigger path (3 DB writes + S3/Langfuse call down to 1 INSERT on happy path) - Widened create_evaluation_run and create_evaluation_run_or_409 to accept optional v2-only params with safe defaults; v1 batch path unchanged - Updated tests: fanout partitioning test now uses metadata-sized dataset; dispatch-failure test replaces removed item-fetch error scenario Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
…i-backend into refactor/evals2.0
Rename migration to avoid conflict with 084_assessment_submissions_and_result_files. Update revision chain: 084 revises 083, 085 revises 084.
Logger cleanup: - Remove 18 routine operation loggers (stages skipped/running/finished) - Keep warnings/errors for actual failure conditions - Affected: fast.py, cron.py, judge_stage.py, iteration.py Code simplification: - Delete create_stage_job wrapper from fast_chunks.py - Inline create_batch_job calls in fast.py (3 sites) - Reduces indirection, no functional change Per code review feedback from PR #1202. Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
Condense 2-line comments: - Run-level input resolution behavior - Judge call cost grouping No unused imports found (generate_run_ai_summary used at line 728). No now_/now naming issues in cron.py. Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
- Remove unused Any import from fast_chunks.py - Change question_id: Any → int | None in build_response_result - Make retry_openai_call decorator generic (ParamSpec/TypeVar) so decorated functions preserve return types instead of erasing to Any Type checking: 0 errors. Decorator genericity fixes return-type erasure on _create_response and _create_embedding. Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
Define proper types for evaluation result structures: - ResponseResult: Stage 1 response evaluation result - EmbeddingResult: Stage 2 embedding pair result Replace dict[str, Any] with typed structures in: - fast_results.py builders - fast.py function signatures Improves type safety and code clarity for API response handling. Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Issue
No linked issue. Reliability fixes came out of an audit of the v2 fast-eval / iteration path while splitting it.
Summary
Started as a structural split of
crud/evaluations/fast.py; grew to include three reliability fixes on the same code that the split made visible. Existing behaviour is otherwise unchanged.1. Module split (no behaviour change)
fast.pyhad reached 1429 lines around a 375-line_stage3_score_and_tracemixing v1 cosine and v2 judge scoring, trace building, cost attachment and the run rollup in one body. Extracted, component-wise:crud/evaluations/fast_results.pycrud/evaluations/fast_chunks.pybatch_jobbookkeeping + chunk queries/cleanupcrud/evaluations/fast_cosine.pycrud/evaluations/fast_traces.pyTraceDatabuildingcrud/evaluations/judge_stage.pycrud/evaluations/judge_prompts.pyscore.pycrud/evaluations/retry.pyservices/evaluations/iteration_state.pyservices/evaluations/iteration_checkpointer.pyPostgresSaverLOC:
fast.py1429 → 1006,score.py359 → 218,iteration_graph.py612 → 504. Largest function 375 → 115 lines.fast.pystays above a 400-line target because stage 1/2 IO is pinned there by existing tests patchingapp.crud.evaluations.fast.*.Two small deltas that narrow existing risk:
_effective_duplication_factorskips the dataset query when the run already carries a factor, and_cleanup_response_chunksnow guards storage resolution as well as the deletes.2. Score unit persisted before the
completedtransitionrun_fast_evaluationmarked a runcompletedand only then calledsave_score, with chunk cleanup and the Langfuse sync in between. A crash in that window (OOM, SIGKILL, hard timeout) left the runcompletedwith its score unit never written, and nothing recovers it: the aggregate short-circuits oncompletedand the cron barrier only selectsprocessing. On v2 that unit is the only copy of the per-row judge scores and reasoning; no Langfuse traces exist to re-fetch from.Now
save_scoreruns first, thecompletedtransition second, and cleanup + Langfuse sync last as a best-effort tail that can never fail the run. A crash leaves the runprocessingand visibly unfinished instead of silently scoreless. Follow-ons: the completed transition no longer rewritesscore(already persisted), andsave_scorereturningNoneraises instead of completing a scoreless run.3. Guarded iteration-loop resumes
dispatch_pending_evaluation_iteration_resumesfanned aresume=Truegraph step out to everyPROCESSINGloop every tick, unconditionally. A step slower than the tick got a second one dispatched on top of it: two workers against the same LangGraph checkpoint thread, and a duplicate eval run or improvement job charged for. A loop whose sub-job wedged collected resumes forever.evaluation_iteration_run.last_dispatched_at(migration084, no backfill; NULL means never dispatched). Stamped at kickoff and after every cron dispatch. Rows stamped insideEVAL_ITERATION_DISPATCH_COOLDOWN_MINUTESare skipped. Timestamp heuristic, not a lock; ceiling and upgrade path noted in aponytail:comment.PROCESSINGpastEVAL_ITERATION_STALL_THRESHOLD_HOURSsince kickoff are failed viamark_iteration_run_failed(now public), failure callback included. Threshold deliberately generous since a false positive kills a live loop and POSTs a failure webhook.finalize_nodenow no-ops when the row is no longerPROCESSING. Before, an in-flight step finishing after a reap overwroteFAILEDwithCOMPLETEDand sent a second, contradictory callback. First terminal writer wins, exactly one callback goes out.docs/wiki/modules/evaluations.mdupdated for the module split, the score-persist ordering, and both cron guards.Checklist
fastapi run --reload app/main.pyordocker compose upin the repository root and test.Notes
Tests. New files:
test_fast_results.py,test_fast_cosine.py,test_fast_traces.py,test_fast_chunk_cleanup.py,test_judge_stage.py,test_iteration_state.py(extracted pure functions),test_fast_score_durability.py(reverting the reorder fails all 5),test_cron_iteration_guards.py(cooldown + reaper). Pre-existing tests touched only to age the kickoff stamp past the cooldown and to add the already-terminalfinalize_nodecase. Mutation-checked: removing the cron guards fails their tests.Migration.
084_add_last_dispatched_at_to_evaluation_iteration_runchains after main's083_add_llm_call_metadata(single alembic head).Not in scope, from the same audit: no failure threshold on the judge stage (R1), no healer for an eval run wedged in
processingpost-merge (R3), and the v2 read path still routes judged runs through Langfuse.🤖 Generated with Claude Code