3232from concurrent .futures import ThreadPoolExecutor , as_completed
3333from dataclasses import dataclass , field
3434from typing import Any
35+ from uuid import uuid4
3536
3637import openai
3738from langfuse import Langfuse
7677)
7778from app .crud .evaluations .fast_results import (
7879 EMBEDDING_USAGE_KEYS ,
80+ GUARDRAIL_APPLIED ,
81+ GUARDRAIL_METADATA_KEYS ,
82+ INPUT_GUARDRAIL_METADATA_KEY ,
83+ OUTPUT_GUARDRAIL_METADATA_KEY ,
7984 RESPONSE_USAGE_KEYS ,
8085 EmbeddingResult ,
8186 ResponseResult ,
8287 build_embedding_failure ,
8388 build_response_result ,
8489 extract_usage ,
8590 is_failure_threshold_breached ,
91+ is_guardrail_blocked ,
8692 parse_embedding_pair ,
8793 sum_usage ,
8894)
98104 create_langfuse_dataset_run ,
99105 update_traces_with_cosine_scores ,
100106)
101- from app .crud .evaluations .response_parsing import extract_response_text
102- from app .crud .evaluations .retry import retry_openai_call
107+ from app .crud .evaluations .response_parsing import (
108+ extract_file_search_chunks ,
109+ field_value ,
110+ )
111+ from app .crud .evaluations .retry import retry_llm_call , retry_openai_call
103112from app .crud .evaluations .score import (
104113 JUDGE_FAILED_REASON ,
105114 EvaluationScore ,
112121from app .crud .job import create_batch_job , get_batch_job
113122from app .models import EvaluationRun , EvaluationRunUpdate
114123from app .models .batch_job import BatchJob , BatchJobCreate
115- from app .models .llm .request import TextLLMParams
116- from app .services .llm .mappers import map_kaapi_to_openai_params
117- from app .services .response .response import get_file_search_results
124+ from app .models .llm .request import (
125+ ConfigBlob ,
126+ LLMCallConfig ,
127+ QueryParams ,
128+ TextContent ,
129+ TextInput ,
130+ )
131+ from app .models .llm .response import TextOutput
132+ from app .services .llm .chain .types import BlockResult , GuardrailOutcomeEnum
133+ from app .services .llm .jobs import execute_llm_call
118134
119135logger = logging .getLogger (__name__ )
120136
121137_retry_openai_call = retry_openai_call (logger )
122-
123-
124- @_retry_openai_call
125- def _create_response (openai_client : OpenAI , params : dict [str , Any ]) -> Any :
126- return openai_client .responses .create (** params )
138+ _retry_llm_call = retry_llm_call (logger )
127139
128140
129141@_retry_openai_call
@@ -151,17 +163,62 @@ def _run_in_pool(
151163 return results
152164
153165
154- def _responses_call_for_item (
166+ @_retry_llm_call
167+ def _execute_llm_call_for_question (
155168 * ,
156- openai_client : OpenAI ,
157- base_params : dict [str , Any ],
169+ config : LLMCallConfig ,
170+ question : str ,
171+ project_id : int ,
172+ organization_id : int ,
173+ ) -> BlockResult :
174+ """Generate one answer through the same `/llm/call` path production runs."""
175+ return execute_llm_call (
176+ config = config ,
177+ # Fresh query per attempt: execute_llm_call mutates it in place.
178+ query = QueryParams (input = TextInput (content = TextContent (value = question ))),
179+ job_id = uuid4 (),
180+ project_id = project_id ,
181+ organization_id = organization_id ,
182+ request_metadata = None ,
183+ langfuse_credentials = None ,
184+ # Raw response carries the file_search hits the knowledge_base metric scores.
185+ include_provider_raw_response = True ,
186+ include_guardrail_metadata = True ,
187+ record_call = False ,
188+ )
189+
190+
191+ def _response_text (result : BlockResult ) -> str :
192+ """Generated text; empty when blocked or output wasn't text."""
193+ output = result .response .response .output if result .response else None
194+ return output .content .value if isinstance (output , TextOutput ) else ""
195+
196+
197+ def _guardrail_fields (result : BlockResult ) -> tuple [str | None , str | None , str | None ]:
198+ """`(guardrail, input_to_llm, output_from_llm)` for a scored row."""
199+ # Rephrased rows never reached the LLM; blocked rows carry no metadata.
200+ if result .guardrail_outcome == GuardrailOutcomeEnum .BLOCKED :
201+ return f"{ GuardrailOutcomeEnum .BLOCKED } : { result .error } " , None , None
202+ if result .guardrail_outcome == GuardrailOutcomeEnum .REPHRASED :
203+ return GuardrailOutcomeEnum .REPHRASED , None , None
204+
205+ metadata = result .metadata or {}
206+ guardrail = GUARDRAIL_APPLIED if metadata .keys () & GUARDRAIL_METADATA_KEYS else None
207+ return (
208+ guardrail ,
209+ (metadata .get (INPUT_GUARDRAIL_METADATA_KEY ) or {}).get ("input_to_llm" ),
210+ (metadata .get (OUTPUT_GUARDRAIL_METADATA_KEY ) or {}).get ("output_from_llm" ),
211+ )
212+
213+
214+ def _llm_call_for_item (
215+ * ,
216+ config : LLMCallConfig ,
217+ project_id : int ,
218+ organization_id : int ,
158219 item : dict [str , Any ],
159220) -> ResponseResult :
160- """Run one Responses call for a dataset item, in the batch path's per-item shape.
161-
162- `base_params` is the question-independent OpenAI body produced once by
163- `map_kaapi_to_openai_params`; only `input` varies per item.
164- """
221+ """Generate one item's answer. Never raises: that would abort the whole chunk."""
165222 item_id = item ["id" ]
166223 question = item ["input" ].get ("question" , "" ) if item .get ("input" ) else ""
167224 ground_truth = (
@@ -183,27 +240,51 @@ def failed_result(generated_output: str) -> ResponseResult:
183240 return failed_result ("ERROR: missing question in dataset item" )
184241
185242 try :
186- response = _create_response (openai_client , {** base_params , "input" : question })
187- except openai .OpenAIError as exc :
243+ result = _execute_llm_call_for_question (
244+ config = config ,
245+ question = question ,
246+ project_id = project_id ,
247+ organization_id = organization_id ,
248+ )
249+ except Exception as exc :
188250 logger .warning (
189- f"[_responses_call_for_item] Item failed | item_id={ item_id } | error={ exc } "
251+ f"[_llm_call_for_item] Item failed | item_id={ item_id } | error={ exc } " ,
252+ exc_info = True ,
190253 )
191254 return failed_result (f"ERROR: { exc } " )
192255
256+ # A guardrail block also carries an error, but is a scored outcome, not a failure.
257+ if (
258+ result .error is not None
259+ and result .guardrail_outcome != GuardrailOutcomeEnum .BLOCKED
260+ ):
261+ logger .warning (
262+ f"[_llm_call_for_item] Item failed | item_id={ item_id } | "
263+ f"error={ result .error } "
264+ )
265+ return failed_result (f"ERROR: { result .error } " )
266+
267+ guardrail , input_to_llm , output_from_llm = _guardrail_fields (result )
268+
269+ # Tokens are billed even on an output block, so usage is read off every outcome.
193270 return build_response_result (
194271 item_id = item_id ,
195272 question = question ,
196273 ground_truth = ground_truth ,
197274 question_id = question_id ,
198- generated_output = extract_response_text (response ),
199- response_id = getattr (response , "id" , None ),
200- usage = extract_usage (getattr (response , "usage" , None ), RESPONSE_USAGE_KEYS ),
275+ # Empty for a blocked row: it has no response.
276+ generated_output = _response_text (result ),
277+ response_id = (
278+ result .response .response .provider_response_id if result .response else None
279+ ),
280+ usage = extract_usage (result .usage , RESPONSE_USAGE_KEYS ),
201281 failed = False ,
202- # Plain dicts (not FileResultChunk) so the unit stays JSON-serializable for S3.
203- retrieved_chunks = [
204- {"score" : c .score , "text" : c .text , "filename" : c .filename }
205- for c in get_file_search_results (response )
206- ],
282+ guardrail = guardrail ,
283+ input_to_llm = input_to_llm ,
284+ output_from_llm = output_from_llm ,
285+ retrieved_chunks = extract_file_search_chunks (
286+ result .response .provider_raw_response if result .response else None
287+ ),
207288 )
208289
209290
@@ -317,9 +398,8 @@ def _cleanup_response_chunks(*, session: Session, eval_run: EvaluationRun) -> No
317398def run_response_chunk (
318399 * ,
319400 session : Session ,
320- openai_client : OpenAI ,
321401 eval_run : EvaluationRun ,
322- config : TextLLMParams ,
402+ config_blob : ConfigBlob ,
323403 dataset_items_slice : list [dict [str , Any ]],
324404 chunk_index : int ,
325405) -> None :
@@ -339,20 +419,22 @@ def run_response_chunk(
339419 if existing and existing .raw_output_url :
340420 return
341421
342- base_params , mapper_warnings = map_kaapi_to_openai_params (
343- session = session , kaapi_params = config
344- )
422+ # Resolved blob passed ad-hoc skips a per-row config fetch; shared read-only.
423+ config = LLMCallConfig (blob = config_blob )
424+ # Read off the session-bound row here: the workers must not touch it in threads.
425+ project_id = eval_run .project_id
426+ organization_id = eval_run .organization_id
345427
346- # Ask OpenAI to return the file_search hits so knowledge_base can judge them.
347- # tool_choice stays at the model default (auto) — consistent with normal calls;
348- # a row where the model doesn't query the KB is scored N/A, not forced to search.
349- if any (t .get ("type" ) == "file_search" for t in base_params .get ("tools" , [])):
350- base_params ["include" ] = ["file_search_call.results" ]
428+ # Native params are a dict, Kaapi params a typed model; field_value reads either.
429+ model = field_value (config_blob .completion .params , "model" )
351430
352431 results = _run_in_pool (
353432 items = dataset_items_slice ,
354- worker = lambda item : _responses_call_for_item (
355- openai_client = openai_client , base_params = base_params , item = item
433+ worker = lambda item : _llm_call_for_item (
434+ config = config ,
435+ project_id = project_id ,
436+ organization_id = organization_id ,
437+ item = item ,
356438 ),
357439 max_workers = settings .EVAL_FAST_API_CONCURRENCY ,
358440 )
@@ -372,7 +454,7 @@ def run_response_chunk(
372454 config = {
373455 "run_mode" : "fast" ,
374456 "endpoint" : RESPONSES_ENDPOINT ,
375- "model" : config . model ,
457+ "model" : model ,
376458 "usage" : sum_usage (results , RESPONSE_USAGE_KEYS ),
377459 CHUNK_CONFIG_RUN_ID : eval_run .id ,
378460 CHUNK_CONFIG_INDEX : chunk_index ,
@@ -476,8 +558,12 @@ def _stage2_embeddings(
476558 if cached is not None :
477559 return eval_run , cached
478560
479- # Only embed items that succeeded in Stage 1.
480- embed_candidates = [r for r in response_results if not r .get ("failed" )]
561+ # Blocked rows are unscoreable, not failed; embedding them trips the threshold.
562+ embed_candidates = [
563+ r
564+ for r in response_results
565+ if not r .get ("failed" ) and not is_guardrail_blocked (r )
566+ ]
481567
482568 embedding_results = _run_in_pool (
483569 items = embed_candidates ,
0 commit comments