2020
2121import assemblyai as aai
2222from rich .console import RenderableType
23+ from rich .markup import escape
2324
2425from aai_cli .app .context import AppState
2526from aai_cli .commands .evaluate import _data as eval_data
2627from aai_cli .core import client , jsonshape , wer
28+ from aai_cli .core import llm as gateway
2729from aai_cli .core .errors import CLIError , NotAuthenticated
2830from aai_cli .ui import output
2931
@@ -50,6 +52,31 @@ class EvalOptions:
5052 speech_model : EvalSpeechModel | None
5153 language_code : str | None
5254 concurrency : int
55+ llm_prompt : list [str ] | None
56+ llm_reduce : list [str ] | None
57+ model : str
58+ max_tokens : int
59+
60+ def llm_options (self ) -> _LlmOptions :
61+ """The ``--llm`` / ``--llm-reduce`` chain settings as plain data."""
62+ return _LlmOptions (
63+ prompts = list (self .llm_prompt or []),
64+ reduce_prompts = list (self .llm_reduce or []),
65+ model = self .model ,
66+ max_tokens = self .max_tokens ,
67+ )
68+
69+
70+ @dataclass (frozen = True )
71+ class _LlmOptions :
72+ """The post-transcription LLM-Gateway transform: the per-item ``--llm`` chain
73+ (a *map*) and the across-items ``--llm-reduce`` chain (a *reduce*), plus the
74+ gateway model + token budget both run under."""
75+
76+ prompts : list [str ]
77+ reduce_prompts : list [str ]
78+ model : str
79+ max_tokens : int
5380
5481
5582def _pct (value : object ) -> str :
@@ -75,11 +102,16 @@ def _percentile(values: list[float], q: float) -> float:
75102
76103@dataclass (frozen = True )
77104class _ItemResult :
78- """One scored row: the emitted dict plus the score and latency kept for pooling."""
105+ """One scored row: the emitted dict plus the score and latency kept for pooling.
106+
107+ ``hypothesis`` is the transcript text (``None`` for a failed row) — kept so the
108+ optional ``--llm`` map / ``--llm-reduce`` reduce can run over it after scoring.
109+ """
79110
80111 row : dict [str , object ]
81112 words : wer .Score | None
82113 latency : float
114+ hypothesis : str | None = None
83115
84116
85117def _failed_result (item : eval_data .EvalItem , err : CLIError , latency : float ) -> _ItemResult :
@@ -94,15 +126,16 @@ def _failed_result(item: eval_data.EvalItem, err: CLIError, latency: float) -> _
94126def _score_item (
95127 item : eval_data .EvalItem , transcript : aai .Transcript , latency : float
96128) -> _ItemResult :
97- words = wer .score (item .reference , str (transcript .text or "" ))
129+ hypothesis = str (transcript .text or "" )
130+ words = wer .score (item .reference , hypothesis )
98131 row : dict [str , object ] = {
99132 "item" : item .item_id ,
100133 "words" : words .words ,
101134 "errors" : words .errors ,
102135 "wer" : words .wer ,
103136 "latency" : latency ,
104137 }
105- return _ItemResult (row = row , words = words , latency = latency )
138+ return _ItemResult (row = row , words = words , latency = latency , hypothesis = hypothesis )
106139
107140
108141def _pooled_metrics (results : list [_ItemResult ]) -> dict [str , object ]:
@@ -204,6 +237,87 @@ def _transcripts(
204237 )
205238
206239
240+ def _run_llm_map (
241+ api_key : str ,
242+ results : list [_ItemResult ],
243+ llm_opts : _LlmOptions ,
244+ * ,
245+ json_mode : bool ,
246+ quiet : bool ,
247+ ) -> None :
248+ """Run the ``--llm`` chain over each transcribed row and attach it under ``llm``.
249+
250+ A *map*: the chain runs over the row's transcript text (inline, like
251+ ``stream --llm``) and lands as ``{"model", "steps"}`` on the row — the WER score
252+ is untouched. Failed rows have no transcript, so they're skipped.
253+ """
254+ scored = [result for result in results if result .hypothesis is not None ]
255+ with output .status (
256+ f"Running --llm over { len (scored )} transcripts…" , json_mode = json_mode , quiet = quiet
257+ ):
258+ for result in scored :
259+ steps = gateway .run_chain_steps (
260+ api_key ,
261+ llm_opts .prompts ,
262+ transcript_text = result .hypothesis ,
263+ model = llm_opts .model ,
264+ max_tokens = llm_opts .max_tokens ,
265+ )
266+ result .row ["llm" ] = {"model" : llm_opts .model , "steps" : steps }
267+
268+
269+ def _reduce_input (result : _ItemResult ) -> str :
270+ """A row's contribution to the reduce: its last ``--llm`` output, else its transcript."""
271+ llm_data = jsonshape .as_mapping (result .row .get ("llm" ))
272+ if llm_data is not None :
273+ steps = jsonshape .mapping_list (llm_data .get ("steps" ))
274+ if steps :
275+ return str (steps [- 1 ].get ("output" , "" ) or "" )
276+ return result .hypothesis or ""
277+
278+
279+ def _gather_reduce_inputs (results : list [_ItemResult ]) -> str :
280+ """Concatenate every transcribed row's reduce input under an item header."""
281+ blocks : list [str ] = []
282+ for result in results :
283+ if result .hypothesis is None :
284+ continue
285+ text = _reduce_input (result )
286+ if text :
287+ blocks .append (f"### Item: { result .row .get ('item' )} \n { text } " )
288+ return "\n \n " .join (blocks )
289+
290+
291+ def _run_reduce (
292+ api_key : str ,
293+ results : list [_ItemResult ],
294+ llm_opts : _LlmOptions ,
295+ * ,
296+ json_mode : bool ,
297+ quiet : bool ,
298+ ) -> dict [str , object ] | None :
299+ """Run the ``--llm-reduce`` chain once over every row's result; the payload entry.
300+
301+ ``None`` when there's nothing to aggregate (every row failed or transcribed to
302+ empty text) so the caller skips the (billable) gateway call and the payload key.
303+ """
304+ combined = _gather_reduce_inputs (results )
305+ if not combined :
306+ output .emit_warning (
307+ "Nothing to reduce: no transcript text across items." , json_mode = json_mode
308+ )
309+ return None
310+ with output .status ("Running --llm-reduce over all items…" , json_mode = json_mode , quiet = quiet ):
311+ result = gateway .run_chain (
312+ api_key ,
313+ llm_opts .reduce_prompts ,
314+ transcript_text = combined ,
315+ model = llm_opts .model ,
316+ max_tokens = llm_opts .max_tokens ,
317+ )
318+ return {"model" : llm_opts .model , "prompts" : llm_opts .reduce_prompts , "output" : result }
319+
320+
207321def _payload (
208322 label : str , speech_model : EvalSpeechModel | None , results : list [_ItemResult ]
209323) -> dict [str , object ]:
@@ -249,6 +363,36 @@ def _secs_cell(row: dict[str, object], key: str) -> str:
249363 return _secs (row [key ]) if key in row else ""
250364
251365
366+ def _final_llm_output (row : dict [str , object ]) -> str | None :
367+ """A row's last ``--llm`` step output, or ``None`` when no chain ran on it."""
368+ llm_data = jsonshape .as_mapping (row .get ("llm" ))
369+ if llm_data is None :
370+ return None
371+ steps = jsonshape .mapping_list (llm_data .get ("steps" ))
372+ return str (steps [- 1 ].get ("output" , "" ) or "" ) if steps else ""
373+
374+
375+ def _llm_block (payload : dict [str , object ]) -> str | None :
376+ """The per-item ``--llm`` outputs as a heading + one ``item: output`` line each,
377+ or ``None`` when no ``--llm`` chain ran."""
378+ lines : list [str ] = []
379+ for row in jsonshape .mapping_list (payload .get ("rows" )):
380+ final = _final_llm_output (row )
381+ if final is not None :
382+ lines .append (f"{ escape (str (row .get ('item' )))} : { escape (final )} " )
383+ if not lines :
384+ return None
385+ return "\n " .join ([output .heading ("--llm" ), * lines ])
386+
387+
388+ def _reduce_block (payload : dict [str , object ]) -> str | None :
389+ """The ``--llm-reduce`` aggregate as a heading + the output, or ``None`` when unset."""
390+ reduce = jsonshape .as_mapping (payload .get ("reduce" ))
391+ if reduce is None :
392+ return None
393+ return f"{ output .heading ('--llm-reduce' )} \n { escape (str (reduce .get ('output' , '' )))} "
394+
395+
252396def _render (payload : dict [str , object ]) -> RenderableType :
253397 has_wer = "wer" in payload
254398 has_failed = "failed" in payload
@@ -271,7 +415,11 @@ def _render(payload: dict[str, object]) -> RenderableType:
271415 table .add_row (* cells )
272416 model = payload .get ("speech_model" ) or "default model"
273417 return output .stack (
274- output .muted (f"{ payload .get ('dataset' )} · { model } " ), table , _summary (payload )
418+ output .muted (f"{ payload .get ('dataset' )} · { model } " ),
419+ table ,
420+ _summary (payload ),
421+ _llm_block (payload ),
422+ _reduce_block (payload ),
275423 )
276424
277425
@@ -310,7 +458,14 @@ def run_evaluate(opts: EvalOptions, state: AppState, *, json_mode: bool) -> None
310458 strict = True , # pragma: no mutate (defensive invariant; _transcripts returns one outcome per item)
311459 )
312460 ]
461+ llm_opts = opts .llm_options ()
462+ if llm_opts .prompts :
463+ _run_llm_map (api_key , results , llm_opts , json_mode = json_mode , quiet = state .quiet )
313464 payload = _payload (data .label , opts .speech_model , results )
465+ if llm_opts .reduce_prompts :
466+ reduce = _run_reduce (api_key , results , llm_opts , json_mode = json_mode , quiet = state .quiet )
467+ if reduce is not None :
468+ payload ["reduce" ] = reduce
314469 output .emit (payload , _render , json_mode = json_mode )
315470 failed = jsonshape .as_int (payload .get ("failed" ))
316471 if failed :
0 commit comments