evaluate()

Submit agent output for risk-based evaluation. The Coalex platform computes quality metrics, assigns a risk score, and returns an automated decision: auto-approve or escalate for human review. evaluate() decides whether a human is asked, not whether the output is good; a refusal is a human's resolution, never something this call returns.

---

Signature

PythonTypeScript

def evaluate(
    *,
    request_id: str,
    input: dict,
    output: dict,
    metrics: dict[str, list[str]] | None = None,
    metadata: dict | None = None,
    agent_id: str | None = None,
    resume_hook: ResumeHook | None = None,
    resolver: str | None = None,
    ttl_business_hours: float | None = None,
    tool_name: str | None = None,
    run_id: str | None = None,
    parent_agent_id: str | None = None,
) -> EvaluationDecision

interface EvaluateOptions {
    requestId: string;
    input: Record<string, unknown>;
    output: Record<string, unknown>;
    metrics?: Record<string, string[]>;
    metadata?: Record<string, unknown>;
    agentId?: string;
    resumeHook?: ResumeHook;
    toolName?: string;
    runId?: string;
    parentAgentId?: string;
}

async function evaluate(options: EvaluateOptions): Promise<EvaluationDecision>

---

Parameters

Parameter	Type	Default	Description
request_id	str	required	Links this evaluation to an observed trace. Use the same request ID passed to coalex_context().
input	dict	required	The agent's input data (e.g., user question, context documents).
output	dict	required	The agent's actual output (e.g., generated answer).
metrics	dict[str, list[str]] \| None	None	Optional. Map of output field names to lists of metric function names. Metrics are declared at evaluate-time and computed at resolve-time from human corrections.
metadata	dict \| None	None	Arbitrary metadata to attach to the evaluation (e.g., model name, prompt version, user segment).
agent_id	str \| None	the active context's	The agent that asks; the decision is attributed to it. TypeScript agentId.
resume_hook	ResumeHook \| None	None	How the producer resumes after the escalation is resolved (checkpoint or task hook). Stored on the escalation so the Console can continue the run. Omitted: no hook is sent. TypeScript resumeHook.
resolver	str \| None	None	Who may resolve the escalation: "self" or "role:<name>". Omitted: the platform default applies. A resolver the organisation set for the tool takes precedence. (Python SDK)
ttl_business_hours	float \| None	None	Business hours the approval may wait before it expires and the agent is told nobody decided. Counted only from 08:00 to 18:00 UTC, Monday to Friday, so a request raised on a Friday evening keeps its full allowance into the next working day. Accepted above 0 and up to 1000. Omitted, or outside that range: the approval waits 120 consecutive hours from the moment it is raised, weekends included. (Python SDK)
tool_name	str \| None	None	The name of the tool this call is about, e.g. "wire_funds". Stored on the escalation and in the audit-chain entry, so an auditor reads which action was approved. Only the name: the arguments belong in input / output. Omitted: the platform reads a tool key from output, input or metadata. TypeScript toolName.
run_id	str \| None	the active context's	The run (one execution of the agent system, across pauses and resumes) the call belongs to, stored on the escalation. TypeScript runId.
parent_agent_id	str \| None	the active context's	The parent the asking agent claims, stored on the escalation as a claim; it governs nothing. Taken from the context only when the call is for the context's own agent. TypeScript parentAgentId.

All parameters are keyword-only (enforced by *).

Renamed from trace_id

The trace_id parameter was renamed to request_id in v1.0.0 to better reflect its purpose. The old trace_id parameter is still accepted for backward compatibility but is deprecated.

Requires register() first

evaluate() uses the endpoint and API key configured by register(). Calling it before register() raises RuntimeError.

---

Returns

EvaluationDecision

PythonTypeScript

@dataclass(frozen=True)
class EvaluationDecision:
    status: str  # "auto_approved" | "escalated"
    risk_score: float  # 0.0 - 1.0
    escalation_id: str | None  # present when status == "escalated"

interface EvaluationDecision {
    readonly status: "auto_approved" | "escalated";
    readonly riskScore: number;        // 0.0 - 1.0
    readonly escalationId?: string;    // present when status === "escalated"
}

Field	Type	Description
status	str	The evaluation decision: "auto_approved" or "escalated". There is no third value — evaluate() decides whether a human is asked, not whether an output is good. A refusal is a human's resolution, not a decision this call returns.
risk_score	float	Risk score between 0.0 (no risk) and 1.0 (maximum risk).
escalation_id	str \| None	Unique escalation identifier. Present only when status == "escalated". Use this with resolve() to submit human review.

---

Known Metrics

The metrics parameter accepts the following metric function names. Unknown metric names raise ValueError at call time.

Metric	Description
f1	Token-level F1 score. Measures precision and recall of overlapping tokens between output and expected.
word_overlap	Share of the distinct words found in either version that appear in both (Jaccard); word order is ignored, and words added count against it.
bleu	BLEU score. Standard machine translation metric measuring n-gram overlap.
rouge_l	ROUGE-L score. Longest common subsequence-based metric.
exact_match	Binary metric: 1.0 if output exactly matches expected, 0.0 otherwise.
levenshtein	Normalized Levenshtein similarity (1.0 = identical, 0.0 = completely different).
contains	Binary metric: 1.0 if the expected text is contained within the output, 0.0 otherwise.

Choosing metrics

• Use f1 or rouge_l for extractive QA where token overlap is meaningful.
• Use exact_match for structured fields like codes, IDs, or categories.
• Use contains to check if specific phrases or keywords appear in the output.
• Combine multiple metrics per field for more robust evaluation.

---

Examples

Basic evaluation

PythonTypeScript

import coalex

decision = coalex.evaluate(
    request_id="abc-123-def",
    input={"question": "What is the maximum coverage?"},
    output={"answer": "The maximum coverage is $500,000."},
    metrics={"answer": ["f1", "rouge_l"]},
)

print(decision.status)  # "auto_approved"
print(decision.risk_score)  # 0.12

import { evaluate } from "@coalex-ai/sdk";

const decision = await evaluate({
    requestId: "abc-123-def",
    input: { question: "What is the maximum coverage?" },
    output: { answer: "The maximum coverage is $500,000." },
    metrics: { answer: ["f1", "rouge_l"] },
});

console.log(decision.status);    // "auto_approved"
console.log(decision.riskScore); // 0.12

Multi-field evaluation

PythonTypeScript

decision = coalex.evaluate(
    request_id="trace-789",
    input={
        "claim_text": "Patient reports knee pain after fall.",
    },
    output={
        "diagnosis": "Acute knee injury, likely meniscus tear",
        "icd_code": "S83.2",
        "recommendation": "MRI recommended within 48 hours",
    },
    metrics={
        "diagnosis": ["rouge_l", "f1"],
        "icd_code": ["exact_match"],
        "recommendation": ["rouge_l", "contains"],
    },
)

const decision = await evaluate({
    requestId: "trace-789",
    input: {
        claim_text: "Patient reports knee pain after fall.",
    },
    output: {
        diagnosis: "Acute knee injury, likely meniscus tear",
        icd_code: "S83.2",
        recommendation: "MRI recommended within 48 hours",
    },
    metrics: {
        diagnosis: ["rouge_l", "f1"],
        icd_code: ["exact_match"],
        recommendation: ["rouge_l", "contains"],
    },
});

Handling escalations

PythonTypeScript

decision = coalex.evaluate(
    request_id="trace-456",
    input={"question": "Is this drug safe during pregnancy?"},
    output={"answer": "Yes, it is generally safe."},
    metrics={"answer": ["rouge_l", "f1"]},
)

if decision.status == "escalated":
    print(f"Escalated! ID: {decision.escalation_id}")
    print(f"Risk score: {decision.risk_score}")
    # Human review needed -- use coalex.resolve()
else:
    print("Output approved automatically -- no human was asked.")

const decision = await evaluate({
    requestId: "trace-456",
    input: { question: "Is this drug safe during pregnancy?" },
    output: { answer: "Yes, it is generally safe." },
    metrics: { answer: ["rouge_l", "f1"] },
});

if (decision.status === "escalated") {
    console.log(`Escalated! ID: ${decision.escalationId}`);
    console.log(`Risk score: ${decision.riskScore}`);
    // Human review needed -- use resolve()
} else {
    console.log("Output approved automatically -- no human was asked.");
}

---

Validation

• Unknown metrics: If any metric name in metrics is not in the known metrics list, evaluate() raises ValueError immediately without making a network call.
• Missing register(): Raises RuntimeError if register() has not been called.
• Network errors: HTTP errors from the Coalex API raise httpx.HTTPStatusError.

---

Notes

• Every call sends a fresh random Idempotency-Key, in both SDKs. A network-level resend of the same request carries the same key and the platform replays the first decision instead of creating a second escalation; two deliberate calls get two keys and two decisions. The replay answers the first decision whether the escalation is still pending or a person has already approved, rejected or corrected it, and the same key with a different request is refused with 422.
• An escalation nobody decides expires, the agent is told nobody decided, and a later resolve() on it is refused with 409. By default that happens 120 consecutive hours after it is raised, weekends included; ttl_business_hours sets a deadline counted in business hours instead.
• When Coalex cannot be reached, evaluate() raises. If your agent proceeds anyway, record_ungoverned() puts the action in the compliance report.
• The evaluation is performed server-side by the Coalex platform. The SDK sends the input, output, and metrics to the API and receives the decision.
• Metrics are declared at evaluate-time but computed at resolve-time from human corrections. When a human reviewer provides corrections via resolve(), metrics compare the original output against the corrections.
• All evaluations (including auto_approved) are stored in the escalations table for full audit trail.
• The risk score is based on the agent's health score, not on metric comparisons at evaluate-time.

---

API Reference

coalex.evaluate

SDK evaluate() -- submit agent output for risk assessment.

Classes

EvaluationDecision dataclass

Result of an evaluate() call.

Source code in coalex/evaluate.py

@dataclasses.dataclass(frozen=True)
class EvaluationDecision:
    """Result of an evaluate() call."""

    # "auto_approved" | "escalated". `rejected` was listed here and the server has never
    # returned it: /evaluate chooses between those two and nothing else. A human
    # rejecting an escalation is a different event, recorded on the escalation.
    status: str
    risk_score: float  # 0.0-1.0
    escalation_id: str | None  # present when status == "escalated"
    # The same outcome in MCP vocabulary: "escalated" is "input_required". None against an
    # older platform that does not send it yet.
    task_state: str | None = None

Functions:

evaluate

evaluate(
    *,
    request_id: str | None = None,
    trace_id: str | None = None,
    input: dict[str, Any],
    output: dict[str, Any],
    metrics: dict[str, list[str]] | None = None,
    metadata: dict[str, Any] | None = None,
    agent_id: str | None = None,
    resume_hook: ResumeHook | None = None,
    resolver: str | None = None,
    ttl_business_hours: float | None = None,
    tool_name: str | None = None,
    run_id: str | None = None,
    parent_agent_id: str | None = None,
    idempotency_key: str | None = None,
    _config: CoalexConfig | None = None,
    _api_key: str | None = None,
) -> EvaluationDecision

Submit agent output for risk-based evaluation.

Parameters:

Name	Type	Description	Default
request_id	str | None	Links to an observed request.	None
trace_id	str | None	Deprecated — use request_id instead.	None
input	dict[str, Any]	Agent input data.	required
output	dict[str, Any]	Agent output data.	required
metrics	dict[str, list[str]] | None	Map of output field names to metric function names (optional).	None
metadata	dict[str, Any] | None	Metadata for field-specific quality focus (optional).	None
agent_id	str | None	Agent identifier. Defaults to the active coalex_context()'s agent_id. Propagated so mid-run escalations get the right agent even before the trace ingests.	None
ttl_business_hours	float | None	Business hours this approval may wait before it expires and the agent is told nobody decided. Counted only from 08:00 to 18:00 UTC, Monday to Friday. None → the approval waits 120 consecutive hours from the moment it is raised, weekends included.	None
resume_hook	ResumeHook | None	How the producer resumes after resolution (Decision Request §4.4). When given, the agent's checkpoint/task hook is persisted on the resulting escalation so the Console (and the resume-hook trigger on resolve) can continue the run. Omitted → no hook sent (fully backward compatible).	None
tool_name	str | None	The name of the tool this call is about, e.g. "wire_funds". Stored on the escalation and in the decision's audit-chain entry, so an auditor reads which action was approved. Only the name: the tool's arguments belong in input/output. Omitted → the platform reads a tool key from output, input or metadata.	None
run_id	str | None	The run (one execution of the agent system, across pauses and resumes) this call belongs to, stored on the escalation. Defaults to the active coalex_context()'s run_id.	None
parent_agent_id	str | None	The agent the asking agent claims as its parent, stored on the escalation as a claim; it governs nothing. Defaults to the active coalex_context()'s parent, and only when the call is for that context's own agent: an explicit agent_id naming another agent takes no parent from the context.	None
idempotency_key	str | None	The key that names this call. A call repeated with the same key and the same body gets the first answer back (same verdict, same escalation) instead of a second decision; the same key with a different body is refused. Pass one when the caller repeats the call on purpose, as a LangGraph node does when it re-runs on resume. Omitted → a fresh key for every call, so two calls are two actions.	None
_config	CoalexConfig | None	Override config (testing only).	None
_api_key	str | None	Override API key (testing only).	None

Source code in coalex/evaluate.py

def evaluate(
    *,
    request_id: str | None = None,
    trace_id: str | None = None,
    input: dict[str, Any],
    output: dict[str, Any],
    metrics: dict[str, list[str]] | None = None,
    metadata: dict[str, Any] | None = None,
    agent_id: str | None = None,
    resume_hook: ResumeHook | None = None,
    resolver: str | None = None,
    ttl_business_hours: float | None = None,
    tool_name: str | None = None,
    run_id: str | None = None,
    parent_agent_id: str | None = None,
    idempotency_key: str | None = None,
    _config: CoalexConfig | None = None,
    _api_key: str | None = None,
) -> EvaluationDecision:
    """Submit agent output for risk-based evaluation.

    Args:
        request_id: Links to an observed request.
        trace_id: Deprecated — use request_id instead.
        input: Agent input data.
        output: Agent output data.
        metrics: Map of output field names to metric function names (optional).
        metadata: Metadata for field-specific quality focus (optional).
        agent_id: Agent identifier. Defaults to the active coalex_context()'s
            agent_id. Propagated so mid-run escalations get the right agent even
            before the trace ingests.
        ttl_business_hours: Business hours this approval may wait before it expires and the agent is
            told nobody decided. Counted only from 08:00 to 18:00 UTC, Monday to Friday. None → the
            approval waits 120 consecutive hours from the moment it is raised, weekends included.
        resume_hook: How the producer resumes after resolution (Decision Request §4.4).
            When given, the agent's checkpoint/task hook is persisted on the
            resulting escalation so the Console (and the resume-hook trigger on resolve) can
            continue the run. Omitted → no hook sent (fully backward compatible).
        tool_name: The name of the tool this call is about, e.g. ``"wire_funds"``. Stored on the
            escalation and in the decision's audit-chain entry, so an auditor reads which action was
            approved. Only the name: the tool's arguments belong in ``input``/``output``. Omitted →
            the platform reads a ``tool`` key from ``output``, ``input`` or ``metadata``.
        run_id: The run (one execution of the agent system, across pauses and resumes) this call
            belongs to, stored on the escalation. Defaults to the active coalex_context()'s run_id.
        parent_agent_id: The agent the asking agent claims as its parent, stored on the escalation
            as a claim; it governs nothing. Defaults to the active coalex_context()'s parent, and
            only when the call is for that context's own agent: an explicit ``agent_id`` naming
            another agent takes no parent from the context.
        idempotency_key: The key that names this call. A call repeated with the same key and the same
            body gets the first answer back (same verdict, same escalation) instead of a second
            decision; the same key with a different body is refused. Pass one when the caller repeats
            the call on purpose, as a LangGraph node does when it re-runs on resume. Omitted → a fresh
            key for every call, so two calls are two actions.
        _config: Override config (testing only).
        _api_key: Override API key (testing only).
    """
    if trace_id is not None and request_id is None:
        import warnings

        warnings.warn("trace_id is deprecated, use request_id", DeprecationWarning, stacklevel=2)
        request_id = trace_id
    if request_id is None:
        raise TypeError("evaluate() requires 'request_id' (or deprecated 'trace_id')")

    import coalex as _sdk

    cfg = _config or _sdk._config
    api_key = _api_key or _sdk._api_key
    if cfg is None:
        raise RuntimeError("coalex.register() must be called before evaluate()")

    # Validate metrics against known catalog (only when explicitly provided)
    if metrics:
        for field, metric_list in metrics.items():
            unknown = set(metric_list) - KNOWN_METRICS
            if unknown:
                raise ValueError(f"Unknown metrics for field '{field}': {unknown}")

    from coalex.context import current_agent_id, current_parent_agent_id, current_run_id

    context_agent_id = current_agent_id()
    resolved_agent_id = agent_id or context_agent_id
    resolved_run_id = run_id or current_run_id()
    # The context's parent is the parent of the context's agent, so it is claimed for that agent only.
    if parent_agent_id is None and resolved_agent_id is not None and resolved_agent_id == context_agent_id:
        parent_agent_id = current_parent_agent_id()

    payload: dict[str, Any] = {
        "request_id": request_id,
        "input": input,
        "output": output,
    }
    if resolved_agent_id:
        payload["agent_id"] = resolved_agent_id
    if parent_agent_id:
        payload["parent_agent_id"] = parent_agent_id
    if resolved_run_id:
        payload["run_id"] = resolved_run_id
    if metrics:
        payload["metrics"] = metrics
    if metadata:
        payload["metadata"] = metadata
    if resume_hook is not None:
        payload["resume_hook"] = resume_hook.to_dict()
    # Who may resolve this escalation ('self' | 'role:<name>'). Persisted on the escalation
    # so role-based resolve authorization can check the reviewer's approval role. Omitted → 'self'.
    if resolver is not None:
        payload["resolver"] = resolver
    if tool_name is not None:
        payload["tool_name"] = tool_name
    if ttl_business_hours is not None:
        # How long this approval is worth waiting for, counted in BUSINESS hours (08:00 to 18:00 UTC,
        # Monday to Friday). Omit and the approval waits 120 consecutive hours, weekends included.
        # Per-call because the right deadline belongs to the call: a wire that must clear today and
        # a report that can wait until Monday should not share one.
        payload["ttl_business_hours"] = ttl_business_hours

    # Without a key from the caller, one key per invocation, minted here rather than taken from the payload.
    #
    # `request_id` looks like the natural key and is the wrong one: for the Console's agents it IS
    # the chat session id, so every governed call in one conversation shares it. Deduplicating on it
    # would collapse a transfer and a later email into one, and the second would vanish in silence.
    #
    # A fresh key per call is what makes this mean the right thing: a transport replay — a proxy, a
    # queue, a socket retried under the hood — resends these exact bytes and carries this exact key,
    # so the platform replays the first decision. An agent deliberately calling evaluate() twice
    # mints two keys and gets two decisions, because those are two actions.
    if idempotency_key is None:
        idempotency_key = uuid.uuid4().hex

    with httpx.Client(timeout=30.0) as client:
        resp = client.post(
            f"{cfg.endpoint}/api/v1/evaluate",
            json=payload,
            headers=sdk_headers(api_key, **{"Idempotency-Key": idempotency_key}),
        )
        raise_for_status(resp, "evaluate()")

    # The platform answered, so it is up — take any backlog of ungoverned records with us.
    # Success is the only honest signal that an outage has ended; a spool that drained
    # only on the next failure would hold the backlog for as long as the outage and then some.
    # Best-effort by construction, and off this path: the drain runs on its own thread, one at a
    # time, so a backlog of up to fifty records never delays the governed call that found it.
    from coalex.ungoverned import flush_in_background

    flush_in_background(cfg, api_key)

    # Surface server-side deprecation notices (X-Deprecation) to the caller.
    deprecation = resp.headers.get("X-Deprecation")
    if isinstance(deprecation, str) and deprecation:
        import warnings

        warnings.warn(f"Coalex API: {deprecation}", DeprecationWarning, stacklevel=2)

    data = resp.json()
    return EvaluationDecision(
        status=data["status"],
        risk_score=data["risk_score"],
        escalation_id=data.get("escalation_id"),
        task_state=data.get("task_state"),
    )