# Evaluations source: https://docs.chalk.ai/docs/compute/evaluations ## Evaluate any LLM workload against historical production data and compare outcomes based on criteria you define. ### Overview An evaluation is defined by three things: a completed dataset revision, one deployed task @cc.function version, and one or more deployed scorer @cc.function versions. Creating an evaluation saves that definition in the active enviroment. Running it queues a separate execution that calls the task once per dataset row, passes each output to every scorer, and writes the scores to a result dataset. The definition is immutable, so two runs of the same evaluation stay comparable: the data, the task, and the scorers are fixed at creation time. To evaluate a new version of the task, create a new evaluation. ``` ┌──────────────────┐ define execute │ dataset revision │──┐ ┌──────────────┐ ┌─────────────────┐ │ task version │──┼──────▸│ Evaluation │──────▸│ EvaluationRun │ │ scorer versions │──┘ │ immutable │ │ repeatable │ └──────────────────┘ └──────────────┘ └────────┬────────┘ pinned inputs │ produces ▾ result dataset revision ``` Tasks and scorers are ordinary Chalk Functions, so everything on that page applies to them: resource configuration, secrets, retries, rate limits, and tracing. ### Quick start ``` import chalkcompute dataset = chalkcompute.DatasetClient().upload("support_goldens", "support_goldens.csv") suite = chalkcompute.EvaluationSuite.create("Release") @chalkcompute.function def answer(input: str) -> str: return call_support_model(input) @chalkcompute.function def response_quality(input: str, output: str) -> chalkcompute.EvaluationScorerResult: score, details = score_response(input=input, output=output) return chalkcompute.EvaluationScorerResult(score=score, metadata={"details": details}) evaluation = chalkcompute.Evaluation.create( "Customer Support Chatbot", dataset=dataset, task=answer, scorers=[response_quality], suite_id=suite.id, ) run = evaluation.run(metadata={"git_sha": "abc123"}).wait() print(run.status, run.result_dataset) ``` ### Datasets Upload a dataset with DatasetClient.upload(name, data). It registers a dataset revision and returns a DatasetRevisionRef you pass straight to Evaluation.create. Your data is ordinary tabular data, so this needs neither ChalkPy nor feature definitions. Pass any of the following as data: - a CSV or Parquet path, or a sequence of them, whose Arrow schemas match exactly, - a PyArrow table or record batch, - a mapping of column name to values, or a sequence of row mappings, - any dataframe exposing to_arrow or the dataframe interchange protocol. CSV files and in-memory values are converted to temporary Parquet files locally before upload. Uploading under a name that already exists creates a new revision of that dataset. ``` import chalkcompute client = chalkcompute.DatasetClient() # From files revision = client.upload("support_goldens", ["jan.parquet", "feb.parquet"]) # From rows, with progress revision = client.upload( "support_goldens", [{"input": "where is my order?", "expected": "order status"}], on_progress=lambda done, total: print(f"{done}/{total}"), ) ``` upload also accepts part_size_bytes to set the multipart chunk size and timeout to bound each request, in seconds. ### Referring to a dataset Pass Evaluation.create(dataset=...) the reference upload returned, or build a DatasetRevisionRef for a dataset that is already there: ``` import chalkcompute # What upload returns: the revision it just registered revision = chalkcompute.DatasetClient().upload("support_goldens", "support_goldens.csv") # An exact revision, by id chalkcompute.DatasetRevisionRef(dataset_id="...", revision_id="...") # The latest revision of a dataset, resolved when the evaluation is created chalkcompute.DatasetRevisionRef(dataset_name="support_goldens") ``` A name resolves once, when you create the evaluation, and the resolved revision is then pinned. Later uploads under the same name leave existing evaluations alone. DatasetRevisionRef rejects a mix of the two forms: pass either dataset_name, or dataset_id together with revision_id. create also takes any object exposing dataset_id and revision_id, which pins that revision, or one exposing dataset_name, whose latest revision it resolves. A dataset handle from another Chalk client therefore passes straight through. ### Tasks Write the task as an ordinary @chalkcompute function. The task function parameters bind to your dataset's columns by name, so a task declaring input: str requires the dataset to have an input column. ``` import chalkcompute @chalkcompute.function def answer(input: str) -> str: return call_support_model(input) ``` @chalkcompute.function starts deploying in the background as soon as the module runs, so consecutive definitions build concurrently. Evaluation.create waits on those handles before reading their immutable version IDs, so you do not need to call deploy() yourself for decorated functions. ### Scorers Write a scorer function the same way as a task function. Its parameters bind to dataset columns by name, plus two arguments the runner supplies: output, the task's return value for that row, and trace, covered in Reading task telemetry below. ``` import chalkcompute @chalkcompute.function def exact_match(expected: str, output: str) -> float: return 1.0 if output.strip() == expected.strip() else 0.0 ``` Here expected comes from the dataset and output comes from the task, so this scorer runs against any dataset with an expected column. ### Scorer results A scorer must return a floating point number between 0 and 1 inclusive, one 'EvaluationScorerResult', or a list[EvaluationScorerResult]. A list lets one scorer emit several scores from shared computation, such as a single model call you grade on three axes. An empty list emits no scores for that row. ``` import chalkcompute @chalkcompute.function def rubric(input: str, output: str) -> list[chalkcompute.EvaluationScorerResult]: grades = grade_all_axes(input, output) return [ chalkcompute.EvaluationScorerResult(score=value, metadata={"axis": axis}) for axis, value in grades.items() ] ``` EvaluationScorerResult takes a score and optional metadata: - score must be a finite number between 0 and 1, inclusive. A boolean raises TypeError, and a score outside the range raises ValueError. - metadata must be JSON-serializable, with no NaN or infinity. Chalk stores it per row alongside the score. The return annotation declares the Arrow schema for the function, and EvaluationScorerResult carries its own Arrow contract. Return the dataclass rather than a plain dictionary, which has no such contract and fails to serialize. ### LLM judges chalkcompute.scorers.llm_judge deploys a scorer described by a pydantic model (v1 or v2) instead of a function body. The model's score field becomes the score, and every other field is recorded as row metadata. The judge requests the reply through the OpenAI client using structured outputs and validates it against the model, so a malformed grade fails the row. ``` import chalkcompute from pydantic import BaseModel, Field class TraceQuality(BaseModel): score: float = Field(ge=0, le=1, description="Overall quality.") directness: int = Field(ge=0, le=2, description="Shortest reasonable path to the goal.") task_correctness: int = Field(ge=0, le=2, description="Was the task actually completed?") reason: str trace_quality = chalkcompute.scorers.llm_judge( TraceQuality, model="gpt-5", instructions="You are grading a browser agent's login attempt.", api_key=chalkcompute.Secret.from_chalk_env("OPENAI_API_KEY"), ) ``` The scorer deploys under the model's class name in kebab case, so TraceQuality becomes trace-quality. Pass name= to choose another. The remaining keywords: - inputs names the dataset columns the judge reads, defaulting to ("output",). - prompt_fn replaces the default prompt, and parse_fn replaces structured outputs for an endpoint that does not support them. - completion_kwargs, for example {"temperature": 0, "max_tokens": 400}, are sent with every request. - base_url points at any OpenAI-compatible endpoint. - api_key takes a Secret that the judge reads in the pod. - image supplies a base image, and any other keyword goes to @chalkcompute.function. Every function deployed from a module imports that module, so give sibling functions in the same file an image that installs pydantic as well. A judge is generated at import time, which means it cannot be deployed in strip mode. ### Reading task telemetry Declare a scorer parameter as chalkcompute.Trace and your scorer receives a handle to the telemetry of the task invocation it is scoring, covering every trace and span under that one evaluated call, including nested model calls and each turn of a multi-turn agent. ``` import chalkcompute @chalkcompute.function def total_cost(output: str, trace: chalkcompute.Trace) -> float: return sum(span.attribute_float("llm.cost") or 0.0 for span in trace.get_spans()) ``` Trace holds an agent session id, and every query it makes is session-scoped. A multi-turn agent emits one OpenTelemetry trace per turn, and those traces are siblings of each other, so a session id is what groups them. The id travels as a large_utf8 scalar exactly as a str would, so a scorer can equally declare trace: str and call chalkcompute.Trace.from_id(trace). That same call reads a session from a notebook or from CI when you already have its id. Three reads are available: ``` # Every span in the session. In attribute_filters, a None value matches any # span carrying the key at all. spans = trace.get_spans( operation_name="chat.completion", attribute_filters={"llm.vendor": "openai"}, limit=500, ) # One SessionSummary: trace and span counts, start, end, duration summary = trace.summary() # One TraceSummary per turn turns = trace.traces(limit=20) ``` Span attributes arrive as a map, so span.attributes["llm.cost"] is the string "0.0012" and summing it concatenates text. Use span.attribute_float(key) or span.attribute_int(key) instead, which return None when the attribute is absent or unparseable. ### Waiting for span ingest Spans reach Chalk asynchronously, through a batching span processor, an OTLP collector, and finally ClickHouse. A scorer that runs immediately after the task it scores can therefore observe an incomplete session and compute a plausible but wrong number from it. All three reads poll until the span count they can see stops growing, and all take the same four arguments: | Parameter | Type | Default | Description | | --------------- | ------------- | -------------------- | ------------------------------------------------------------------------------------------------------------------------------------- | | `wait` | `bool` | `True` | Poll until the span count holds steady across two consecutive reads. Pass `False` to read once, for a session that finished long ago. | | `timeout` | `float` | `60.0` | Seconds to wait. Exceeding it raises `TimeoutError` rather than returning an incomplete result. | | `poll_interval` | `float` | `2.0` | Seconds between polls. | | `expect_min` | `int \| None` | `None`, treated as 1 | Minimum span count to wait for. The default treats an empty result as pending ingest. Pass `0` to accept an empty session. | A steady count cannot distinguish a finished session from a quiet one, so a long pause between agent turns can settle the read early. Pass expect_min when you know how many spans to expect. ``` # A judged agent turn always emits at least three spans, so wait for them spans = trace.get_spans(expect_min=3, timeout=120.0) # A session that finished hours ago has nothing left to wait for spans = chalkcompute.Trace.from_id(session_id).get_spans(wait=False) ``` An image built in strip mode does not install chalkcompute, so the handler shim publishes no call context and the call degrades instead of failing. A function deployed that way never joins its caller's session, and get_spans() silently finds nothing. Deploy with a published chalkcompute installed for session-aware scoring. ### Attaching deployed functions by reference Attach functions that already exist rather than redefining them: ``` import chalkcompute evaluation = chalkcompute.Evaluation.create( "Customer Support Chatbot", dataset=dataset, task=chalkcompute.RemoteFunction.from_name("answer"), scorers=[chalkcompute.RemoteFunction.from_version_id("fn_brand_alignment_v2")], ) ``` RemoteFunction.from_name(name) resolves the currently selected version at lookup time, and creating the evaluation then pins that version. A decorated function registers under its Python name, so answer here is the def answer from the quick start. RemoteFunction.from_version_id(id) attaches an exact, immutable version; from_id is a compatibility alias for it. Deploy an imperative RemoteFunction before you use it. Evaluation.create never deploys on your behalf: an undeployed handle raises EvaluationError before any request goes out, and a bare callable raises TypeError. ### Creating an evaluation Pass the dataset, the task, and the scorers to Evaluation.create, along with any metadata you want stored on the definition: ``` import chalkcompute evaluation = chalkcompute.Evaluation.create( "Customer Support Chatbot", dataset=dataset, task=answer, scorers=[response_quality, exact_match], metadata={"owner": "support-eng"}, suite_id=suite.id, ) ``` name is required and must be non-empty, and at least one scorer is required. metadata must be a JSON-serializable mapping with string keys; Chalk stores it on the definition and returns it on every read. suite_id files the evaluation under a suite. The returned Evaluation exposes the pinned identifiers as dataset_id, dataset_revision_id, task_function_version_id, and scorers, a tuple of EvaluationScorer carrying an evaluation-local id and either a function_version_id or a builtin_id. Attach to an existing definition with chalkcompute.Evaluation.from_id("..."), and read the latest server state with evaluation.refresh(). ### Running an evaluation evaluation.run() is an asynchronous process that queues a run and returns a handle immediately: ``` run = evaluation.run(metadata={"git_sha": "abc123"}) print(run.status) # queued, not yet finished final = run.wait() print(final.status, final.result_dataset) ``` wait() polls until the run reaches a terminal state and returns the final handle: | Parameter | Type | Default | Description | | ------------------ | --------------- | ------- | -------------------------------------------------------------------------------------------------------------------- | | `timeout` | `float \| None` | `600.0` | Seconds to wait; `None` waits indefinitely. Exceeding it raises `TimeoutError`. | | `poll_interval` | `float` | `2.0` | Seconds between status polls. | | `raise_on_failure` | `bool` | `True` | Raise `EvaluationRunFailedError` for a failed or canceled run. Pass `False` to inspect the terminal handle yourself. | EvaluationRunStatus covers PENDING, RUNNING, FINALIZING, SUCCEEDED, FAILED, CANCELED, and UNSPECIFIED. run.is_terminal is true for the three that end a run: SUCCEEDED, FAILED, and CANCELED. A successful run exposes run.result_dataset, a DatasetRevisionRef for the dataset revision holding the task output and each scorer's score and metadata per row. It stays None until the run produces one. run.error_message carries the failure detail, and a raised EvaluationRunFailedError exposes the terminal handle as error.run. Read a definition's history with evaluation.runs(limit=...), newest first, or attach to one run with chalkcompute.EvaluationRun.from_id("...") and run.refresh(). ### Suites A suite is a named collection of evaluations and their runs, such as a release gate, a regression set, or one team's work. ``` import chalkcompute suite = chalkcompute.EvaluationSuite.create("Release") for existing in chalkcompute.EvaluationSuite.list_all(): print(existing.id, existing.name) for evaluation in suite.evaluations(limit=20): print(evaluation.name) for run in suite.runs(limit=20): print(run.id, run.status) ``` suite.evaluations() and suite.runs() both yield newest first and page transparently. ### Using EvaluationClient directly The class methods above each construct a client for a single call. Hold an EvaluationClient when you make several calls, or when you want it to borrow credentials from a ConnectClient or a ChalkPy client you already have: ``` import chalkcompute client = chalkcompute.EvaluationClient(chalk_client=my_chalk_client) suite = client.create_suite("Release") evaluation = client.create("Nightly", dataset=dataset, task=answer, scorers=[exact_match]) run = client.run(evaluation.id, metadata={"git_sha": "abc123"}) for candidate in client.list(suite_id=suite.id, limit=50): print(candidate.name) for previous in client.list_runs(evaluation_id=evaluation.id, limit=10): print(previous.id, previous.status) ``` The client carries create_suite and list_suites for suites, create, get, and list for definitions, and run, get_run, and list_runs for runs. list and list_runs are generators that page in batches of up to 100 and stop at limit. Pass only one of chalk_client or connect_client to the constructor. Every object a client returns keeps a reference to that client, which is what makes evaluation.run(), run.refresh(), and suite.evaluations() work. A handle you construct yourself has no client and raises EvaluationError on those methods, so fetch it through the client instead. ### Errors A failed request raises EvaluationError, which carries status_code, the HTTP equivalent of the underlying code. A missing evaluation or run raises EvaluationNotFoundError, a subclass of it: ``` import chalkcompute try: evaluation = chalkcompute.Evaluation.from_id("does-not-exist") except chalkcompute.EvaluationNotFoundError as e: print(e.status_code) # 404 ``` A run that fails or is canceled raises EvaluationRunFailedError out of wait(), with the terminal handle attached so you can read its status and error message: ``` try: run = evaluation.run().wait() except chalkcompute.EvaluationRunFailedError as e: print(e.run.status, e.run.error_message) ``` wait() raises TimeoutError when it runs out of time, as does a Trace read that never settles. Local validation raises ValueError or TypeError for an empty name, an empty scorer list, metadata that is not JSON-serializable, a score outside [0, 1], or a bare callable passed as a task or scorer. That validation runs before any request goes out, so a malformed call fails locally.