Escrever avaliadores Python

A pontuação de avaliações é baseada em Python. Cada tarefa define um avaliador Python com um de três contratos:

Contrato	Função	Ideal para
`sample`	`grade(sample, item[, ctx])`	Correspondência exata por amostra, F1, tolerância numérica, regex, verificações de rubrica personalizada.
`batch`	`grade_batch(samples[, ctx])`	Pontuação agregada, métricas macro, juízes no nível da tarefa ou atualizações de amostras após analisar todo o lote.
`model_backed`	`grade` ou `grade_batch` com auxiliares de modelo em `ctx`	LLM como juiz e pontuação de embedding/similaridade por meio do MKA1.

O Python é executado no serviço de sandbox. Não passe chaves de API ou credenciais de provedores para o código do avaliador. Use ctx.responses_create e ctx.embeddings_create para pontuação baseada em modelo.

Declaração do avaliador

Use código-fonte embutido para avaliadores curtos:

{
  "grader": {
    "type": "python",
    "contract": "sample",
    "metric_id": "score",
    "source": "def grade(sample, item):\n    return 1.0 if sample['extracted_output'] == item['target'] else 0.0\n",
    "timeout_seconds": 120
  }
}

Use arquivos enviados para avaliadores reutilizáveis:

{
  "grader": {
    "type": "python",
    "contract": "sample",
    "file_id": "file_grader123",
    "timeout_seconds": 120
  }
}

Campos:

Campo	Descrição
`type`	Deve ser `python`.
`contract`	`sample`, `batch` ou `model_backed`. O padrão é `sample`.
`execution`	Substituição opcional: `per_sample` ou `aggregate`. O padrão é definido por `contract`.
`model_access`	`mka1` habilita `ctx.responses_create` e `ctx.embeddings_create`. O padrão é `mka1` para `model_backed`; caso contrário, `none`.
`metric_id`	Chave de pontuação usada quando o avaliador retorna um único float. O padrão é `score`.
`source`	Código-fonte Python embutido.
`file_id`	ID de arquivo Python enviado com `purpose=evals`.
`timeout_seconds`	Tempo limite de execução no sandbox. O intervalo é de 1 a 600. O padrão é 120.
`max_model_calls`	Máximo de chamadas de ponte a partir de uma execução do avaliador. O intervalo é de 0 a 500. O padrão é 64.

Forneça source ou file_id.

Contrato de amostra

Avaliadores de amostra definem:

def grade(sample, item):
    ...

Eles também podem aceitar ctx:

def grade(sample, item, ctx):
    ...

sample descreve a saída do modelo:

{
  "output_text": "Raw assistant text",
  "extracted_output": "Parsed answer",
  "model": "openai:gpt-4.1-mini",
  "prompt": "Rendered prompt",
  "task_id": "qa_exact_match",
  "run_id": "eval_run_abc123",
  "sample_id": "eval_sample_abc123"
}

item contém a linha do conjunto de dados, além de campos de conveniência:

{
  "question": "2+2?",
  "answer": "4",
  "prompt": "Rendered prompt",
  "target": "4",
  "reference_answer": "4",
  "choices": [],
  "task_id": "qa_exact_match"
}

Os campos exatos da linha dependem do seu conjunto de dados e pré-processador.

Retornar uma única pontuação

Retorne um float finito quando a tarefa tiver uma métrica. O MKA1 o armazena em metric_id.

def grade(sample, item):
    return 1.0 if sample.get("extracted_output") == item.get("target") else 0.0

Se metric_id for omitido, a chave de pontuação será score.

Retornar várias pontuações

Retorne um dict com um objeto scores quando a tarefa tiver várias métricas.

def grade(sample, item):
    output = (sample.get("extracted_output") or "").strip().lower()
    target = (item.get("target") or "").strip().lower()

    exact = 1.0 if output == target else 0.0
    contains = 1.0 if target and target in output else 0.0

    return {
        "scores": {
            "exact_match": exact,
            "contains_target": contains
        },
        "judge": {
            "output": output,
            "target": target
        }
    }

Somente valores numéricos finitos são armazenados em scores. O objeto opcional judge é preservado na amostra para depuração e painéis.

Resultados inválidos

Os seguintes resultados se tornam inválidos e recebem pontuação zero:

Exceções.
Floats não finitos, como NaN ou Infinity.
Retornos booleanos.
Strings ou outros retornos que não sejam dict nem número.
Dicts sem qualquer pontuação numérica finita.

A carga útil judge da amostra inclui a carga útil inválida bruta e os detalhes do erro.

Contrato de lote

Avaliadores de lote definem:

def grade_batch(samples):
    ...

Eles são executados uma vez por tarefa e modelo durante a finalização. Use-os quando a métrica precisar de todo o conjunto de amostras. Cada amostra de lote contém:

{
  "sample_id": "eval_sample_abc123",
  "task_id": "macro_f1_task",
  "model": "openai:gpt-4.1-mini",
  "prompt": "Rendered prompt",
  "target": "positive",
  "output_text": "The answer is positive.",
  "extracted_output": "positive",
  "dataset_row": {
    "text": "Great support experience.",
    "label": "positive"
  },
  "response_id": "resp_...",
  "scores": {},
  "judge": null
}

Retorne métricas agregadas:

def grade_batch(samples):
    total = len(samples)
    correct = 0

    for sample in samples:
        if sample.get("extracted_output") == sample.get("target"):
            correct += 1

    return {
        "metrics": {
            "accuracy": correct / total if total else 0.0
        }
    }

Retorne atualizações de amostras quando quiser adicionar pontuações por amostra, detalhes do juiz ou saídas extraídas corrigidas:

def grade_batch(samples):
    updates = []
    for sample in samples:
        correct = sample.get("extracted_output") == sample.get("target")
        updates.append({
            "sample_id": sample["sample_id"],
            "scores": {
                "batch_correct": 1.0 if correct else 0.0
            },
            "judge": {
                "checked_in_batch": True
            }
        })

    return {
        "metrics": {
            "batch_accuracy": sum(u["scores"]["batch_correct"] for u in updates) / len(updates)
        },
        "samples": updates
    }

Campos de atualização de amostra:

Campo	Descrição
`sample_id`	Obrigatório. A amostra a atualizar.
`scores`	Chaves de pontuação numéricas opcionais para mesclar à amostra.
`judge`	Objeto opcional para substituir os detalhes do juiz da amostra. Use `null` para limpar.
`extracted_output`	String opcional ou `null` para substituir a saída extraída da amostra.

Se uma tarefa declarar metrics, IDs de métricas de lote inesperados serão removidos dos agregados finais. Isso protege os painéis contra desvios acidentais de métricas.

Avaliadores baseados em modelo

Use contract: "model_backed" quando o Python precisar de chamadas de modelo ou embedding. O código Python solicita uma chamada de ferramenta. O Gateway realiza a chamada com o contexto autenticado da execução e retorna o resultado ao sandbox.

{
  "grader": {
    "type": "python",
    "contract": "model_backed",
    "model_access": "mka1",
    "max_model_calls": 4,
    "file_id": "file_judge_grader123"
  }
}

A execução deve definir judge_model e embedding_model quando o avaliador usar model="auto":

{
  "suite_id": "eval_suite_abc123",
  "models": ["openai:gpt-4.1-mini"],
  "judge_model": "openai:gpt-4.1-mini",
  "embedding_model": "openai:text-embedding-3-small"
}

Se o Python passar um ID de modelo explícito, esse modelo explícito será usado. Se passar model="auto" ou omitir model, o MKA1 usará o judge_model ou embedding_model da execução.

LLM como juiz

import json

def grade(sample, item, ctx):
    response = ctx.responses_create(
        model="meetkai:functionary-pt",
        input=(
            "Você está avaliando uma resposta. Retorne somente JSON.\n"
            f"Pergunta: {sample.get('prompt', '')}\n"
            f"Referência: {item.get('target', '')}\n"
            f"Previsão: {sample.get('output_text', '')}\n"
            'Schema: {"verdict":"correct"|"incorrect","rationale":"..."}'
        ),
        temperature=0,
        text={"format": {"type": "json_object"}},
        metadata={"judge": "reference_binary"}
    )

    parsed = json.loads(response["output_text"])
    score = 1.0 if parsed.get("verdict") == "correct" else 0.0

    return {
        "scores": {
            "judge_score": score
        },
        "judge": {
            "verdict": parsed.get("verdict"),
            "rationale": parsed.get("rationale"),
            "response_id": response.get("id")
        }
    }

ctx.responses_create aceita o mesmo formato de solicitação da API de Responses do MKA1, exceto que o sistema de avaliações força stream=false, store=true e background=false. As respostas do juiz são armazenadas e podem ser auditadas como o tráfego normal de Responses.

Similaridade de embedding

import math

def cosine(a, b):
    dot = sum(x * y for x, y in zip(a, b))
    na = math.sqrt(sum(x * x for x in a))
    nb = math.sqrt(sum(y * y for y in b))
    return 0.0 if na == 0 or nb == 0 else dot / (na * nb)

def grade(sample, item, ctx):
    result = ctx.embeddings_create(
        model="meetkai:functionary-pt",
        input=[
            sample.get("output_text", ""),
            item.get("target", "")
        ]
    )

    output_embedding = result["data"][0]["embedding"]
    target_embedding = result["data"][1]["embedding"]
    score = cosine(output_embedding, target_embedding)

    return {
        "scores": {
            "semantic_similarity": score
        },
        "judge": {
            "embedding_model": result.get("model")
        }
    }

ctx.embeddings_create é roteado pela API de Embeddings do MKA1. O uso é registrado no contexto autenticado da execução.

Pré-processadores Python

Pré-processadores não são avaliadores, mas usam o mesmo modelo de execução em sandbox e regras de carregamento de arquivos. Eles são executados antes da renderização do prompt. Pré-processador de linha:

def transform(row):
    row["question"] = row["question"].strip()
    row["answer"] = str(row["answer"]).strip()
    return row

Pré-processador de lote:

def transform_batch(rows):
    out = []
    for row in rows:
        if row.get("skip"):
            continue
        row["difficulty"] = row.get("difficulty") or "unknown"
        out.append(row)
    return out

Declare-os em uma tarefa:

{
  "preprocess": {
    "type": "python",
    "contract": "batch",
    "file_id": "file_preprocessor123",
    "timeout_seconds": 120
  }
}

Receitas de pontuação comuns

Correspondência exata

def grade(sample, item):
    output = (sample.get("extracted_output") or "").strip()
    target = (item.get("target") or "").strip()
    return {"scores": {"exact_match": 1.0 if output == target else 0.0}}

Correspondência exata sem distinção entre maiúsculas e minúsculas

def grade(sample, item):
    output = (sample.get("extracted_output") or "").strip().lower()
    target = (item.get("target") or "").strip().lower()
    return {"scores": {"exact_match": 1.0 if output == target else 0.0}}

Tolerância numérica

def grade(sample, item):
    try:
        output = float(sample.get("extracted_output"))
        target = float(item.get("target"))
    except (TypeError, ValueError):
        return {"scores": {"numeric_match": 0.0}}

    return {
        "scores": {
            "numeric_match": 1.0 if abs(output - target) <= 0.01 else 0.0,
            "absolute_error": abs(output - target)
        }
    }

F1 de tokens

import re
from collections import Counter

def tokens(text):
    return re.findall(r"\w+", (text or "").lower())

def f1(prediction, target):
    pred = tokens(prediction)
    gold = tokens(target)
    if not pred and not gold:
        return 1.0
    if not pred or not gold:
        return 0.0

    overlap = Counter(pred) & Counter(gold)
    common = sum(overlap.values())
    if common == 0:
        return 0.0

    precision = common / len(pred)
    recall = common / len(gold)
    return 2 * precision * recall / (precision + recall)

def grade(sample, item):
    return {
        "scores": {
            "token_f1": f1(sample.get("output_text", ""), item.get("target", ""))
        }
    }

F1 macro com `grade_batch`

from collections import defaultdict

def grade_batch(samples):
    labels = sorted({
        s.get("target")
        for s in samples
        if s.get("target") is not None
    })

    per_label = []
    for label in labels:
        tp = fp = fn = 0
        for sample in samples:
            pred = sample.get("extracted_output")
            target = sample.get("target")
            if pred == label and target == label:
                tp += 1
            elif pred == label and target != label:
                fp += 1
            elif pred != label and target == label:
                fn += 1

        precision = tp / (tp + fp) if tp + fp else 0.0
        recall = tp / (tp + fn) if tp + fn else 0.0
        score = 2 * precision * recall / (precision + recall) if precision + recall else 0.0
        per_label.append(score)

    return {
        "metrics": {
            "macro_f1": sum(per_label) / len(per_label) if per_label else 0.0
        },
        "samples": [
            {
                "sample_id": s["sample_id"],
                "scores": {
                    "correct": 1.0 if s.get("extracted_output") == s.get("target") else 0.0
                }
            }
            for s in samples
        ]
    }

Depuração de avaliadores

Use primeiro os detalhes da amostra:

curl

curl 'https://apigw.mka1.com/api/v1/llm/evals/runs/eval_run_abc123/samples?limit=10&status=failed' \
  --header 'Authorization: Bearer <mka1-api-key>'

Verifique:

Campo	O que verificar
`prompt`	O prompt renderizado após o pré-processamento e os exemplos few-shot.
`target`	A string de destino renderizada.
`output_text`	Saída bruta do modelo após o tratamento de sequências de parada.
`extracted_output`	Saída após a extração.
`scores`	Valores numéricos que afetaram as métricas agregadas.
`judge`	Detalhes retornados pelo avaliador, além da carga útil bruta de execução quando disponível.
`error`	Detalhes de falhas na geração de candidatos, extração, pré-processamento ou sandbox.

Correções comuns:

Sintoma	Correção
A pontuação é sempre `0`	Verifique se o avaliador retorna um float ou um dict com `scores` numéricos finitos.
`ctx.responses_create` falha	Defina `contract: "model_backed"` e `model_access: "mka1"`, depois forneça um `judge_model` na execução ou um modelo explícito no Python.
`ctx.embeddings_create` falha	Defina um `embedding_model` na execução ou passe um modelo de embedding explícito no Python.
Os campos do prompt estão em branco	Verifique os nomes dos campos da linha do conjunto de dados e a saída do pré-processador.
As métricas de lote estão ausentes	Declare os IDs de métrica em `metrics` ou omita `metrics` para tarefas agregadas que devem aceitar todas as métricas retornadas.
O avaliador excede o tempo limite	Aumente `timeout_seconds`, reduza as chamadas de ponte de modelo ou mova o trabalho agregado caro para tarefas menores.

Modelo de segurança

Avaliadores e pré-processadores Python são executados no serviço de sandbox. Eles são destinados à lógica de avaliação, não a fluxos de trabalho arbitrários de aplicações. Tenha estas regras em mente:

Não coloque segredos no código-fonte do avaliador, nas linhas do conjunto de dados ou nos metadados.
Não espere credenciais de rede brutas dentro do Python.
Use ctx.responses_create e ctx.embeddings_create para chamadas de modelo.
Mantenha os arquivos Python enviados restritos à equipe proprietária da suíte.
Prefira arquivos de avaliador enviados para lógica reutilizável, para que as versões da suíte acompanhem claramente suas dependências.

A API de avaliações preserva os detalhes das chamadas de modelo que consegue observar, incluindo response_ids candidatos e cargas úteis judge do avaliador, para que você possa auditar como as pontuações foram produzidas.

Primeiros passos

Respostas

Recursos

Computação

CLI

Receitas

Benchmarks

Infraestrutura

Escrever avaliadores Python

Declaração do avaliador

Contrato de amostra

Retornar uma única pontuação

Retornar várias pontuações

Resultados inválidos

Contrato de lote

Avaliadores baseados em modelo

LLM como juiz

Similaridade de embedding

Pré-processadores Python

Receitas de pontuação comuns

Correspondência exata

Correspondência exata sem distinção entre maiúsculas e minúsculas

Tolerância numérica

F1 de tokens

F1 macro com `grade_batch`

Depuração de avaliadores

Modelo de segurança

​Declaração do avaliador

​Contrato de amostra

​Retornar uma única pontuação

​Retornar várias pontuações

​Resultados inválidos

​Contrato de lote

​Avaliadores baseados em modelo

​LLM como juiz

​Similaridade de embedding

​Pré-processadores Python

​Receitas de pontuação comuns

​Correspondência exata

​Correspondência exata sem distinção entre maiúsculas e minúsculas

​Tolerância numérica

​F1 de tokens

​F1 macro com grade_batch

​Depuração de avaliadores

​Modelo de segurança

Declaração do avaliador

Contrato de amostra

Retornar uma única pontuação

Retornar várias pontuações

Resultados inválidos

Contrato de lote

Avaliadores baseados em modelo

LLM como juiz

Similaridade de embedding

Pré-processadores Python

Receitas de pontuação comuns

Correspondência exata

Correspondência exata sem distinção entre maiúsculas e minúsculas

Tolerância numérica

F1 de tokens

F1 macro com `grade_batch`

Depuração de avaliadores

Modelo de segurança