Something went wrong. Try again.
End-to-end Tinker model training and serving with Eve.
Something went wrong. Try again.
3.9 kB · 140 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141from __future__ import annotations
import asyncioimport jsonfrom collections.abc import Iterablefrom dataclasses import asdict, dataclassfrom typing import Any
from training.recipes.base import Case, Recipe
@dataclass(frozen=True)class EvaluationRow: id: str kind: str raw: str parsed: dict[str, Any] | None valid_schema: bool exact: bool prompt_tokens: int sampled_tokens: int
def canonical_target(case: Case) -> str: return json.dumps(case.target, separators=(",", ":"))
def evaluate_text( recipe: Recipe, case: Case, raw: str, prompt_tokens: int = 0, sampled_tokens: int = 0,) -> EvaluationRow: stripped = raw.strip() parsed: dict[str, Any] | None = None try: candidate = json.loads(stripped) if isinstance(candidate, dict): parsed = candidate except json.JSONDecodeError: pass
valid_schema = parsed is not None and recipe.validate_output(parsed) return EvaluationRow( id=case.id, kind=case.kind, raw=stripped, parsed=parsed, valid_schema=valid_schema, exact=stripped == canonical_target(case), prompt_tokens=prompt_tokens, sampled_tokens=sampled_tokens, )
def score_rows(rows: Iterable[EvaluationRow]) -> dict[str, Any]: materialized = list(rows) total = len(materialized) if total == 0: raise ValueError("cannot score an empty evaluation") exact = sum(row.exact for row in materialized) valid = sum(row.valid_schema for row in materialized) kind_scores: dict[str, dict[str, int | float]] = {} for kind in sorted({row.kind for row in materialized}): subset = [row for row in materialized if row.kind == kind] kind_exact = sum(row.exact for row in subset) kind_scores[kind] = { "exact": kind_exact, "total": len(subset), "accuracy": kind_exact / len(subset), } return { "exact": exact, "valid_schema": valid, "total": total, "exact_accuracy": exact / total, "schema_accuracy": valid / total, "prompt_tokens": sum(row.prompt_tokens for row in materialized), "sampled_tokens": sum(row.sampled_tokens for row in materialized), "by_kind": kind_scores, }
async def evaluate_native( recipe: Recipe, sampling_client, renderer, cases: tuple[Case, ...], instructions: str, *, max_tokens: int,) -> tuple[list[EvaluationRow], dict[str, Any]]: import tinker from tinker_cookbook.renderers import get_text_content
params = tinker.SamplingParams( max_tokens=max_tokens, temperature=0.0, stop=renderer.get_stop_sequences(), ) prompts = [ renderer.build_generation_prompt( [ {"role": "system", "content": instructions}, {"role": "user", "content": case.prompt}, ] ) for case in cases ] results = await asyncio.gather( *[ sampling_client.sample_async( prompt=prompt, sampling_params=params, num_samples=1, ) for prompt in prompts ] )
rows: list[EvaluationRow] = [] for case, prompt, result in zip(cases, prompts, results, strict=True): tokens = result.sequences[0].tokens parsed_message, _termination = renderer.parse_response(tokens) raw = get_text_content(parsed_message) rows.append( evaluate_text( recipe, case, raw, prompt_tokens=prompt.length, sampled_tokens=len(tokens), ) ) return rows, score_rows(rows)
def serializable_rows(rows: Iterable[EvaluationRow]) -> list[dict[str, Any]]: return [asdict(row) for row in rows]