CI/CD Evaluation

from agentx import AgentX, CIGateFailure

client = AgentX.from_env()

def my_agent(query: str) -> str:
    # your agent here
    return call_my_agent(query)

result = client.tracer.run_eval(
    dataset_id="6876ddd222bbb333ccc444ee",
    agent_fn=my_agent,
    agent_name="customer-support-agent",
    # Keys must be camelCase. The API's gitContext schema is strict
    # and silently drops unrecognized (e.g. snake_case) keys.
    git_context={
        "branch": "feat/new-retrieval",
        "commitSha": "a1b2c3d4e5f6",
    },
    fail_on_gate=True,   # raises CIGateFailure if gate is "fail"
)

print(f"Gate: {result.gate}  |  Pass rate: {result.pass_rate:.0%}")

@dataclass
class CIRunResult:
    run_id: str
    gate: Literal["pass", "fail"]
    pass_rate: float                     # e.g. 0.875
    total_questions: int
    passed_questions: int
    scores: list[CIQuestionScore]
    violations: list[ThresholdViolation]
    finalized_at: str | None

from agentx import AgentX
from agentx.tracing import CIRun, CIRunResult

client = AgentX.from_env()
tracer = client.tracer

# 1. Create the run, receive test cases
run: CIRun = tracer.create_ci_run(
    dataset_id="6876ddd222bbb333ccc444ee",
    agent_name="my-agent",
    git_context={"branch": "main", "commitSha": "abc1234"},
)

# 2. Run agent against each test case
for tc in run.test_cases:
    output = my_agent(tc.query or "")       # tc.query is None if exposeTestInputs is off
    score = tracer.submit_result(
        run.run_id,
        tc.index,
        output,
        latency_ms=350,
    )
    if score.gate_fired:
        print("failFast triggered, run already finalized as FAIL")
        break

# 3. Finalize and get the gate decision
result: CIRunResult = tracer.finalize_ci_run(run.run_id)
print(f"Gate: {result.gate}  ({result.passed_questions}/{result.total_questions} passed)")

from agentx import AgentX, CIGateFailure, CINotEnabled, DatasetNotFound

try:
    result = client.tracer.run_eval(
        dataset_id=dataset_id,
        agent_fn=my_agent,
        fail_on_gate=True,
    )
except DatasetNotFound:
    print("Dataset not found, check the dataset_id")
    sys.exit(1)
except CINotEnabled:
    print("CI not enabled on this dataset, enable it in dataset settings")
    sys.exit(1)
except CIGateFailure as e:
    result = e.result
    print(f"Gate FAILED, {result.pass_rate:.0%} passed")
    for v in result.violations:
        print(f"  Q{v.question_index}: {v.metric} was {v.actual:.2f}, threshold {v.threshold:.2f}")
    sys.exit(1)

result = client.tracer.run_eval(
    dataset_id=dataset_id,
    agent_fn=my_agent,
    concurrency=4,       # run 4 questions in parallel
    fail_on_gate=True,
)

for score in result.scores:
    status = "✓" if score.passed else "✗"
    print(f"  [{status}] Q{score.question_index}: {score.rating}/10, {score.justification[:80]}")

import time
from agentx.tracing import CIRunStatus

while True:
    status: CIRunStatus = tracer.get_ci_run(run_id)
    if status.status in ("completed", "failed"):
        break
    time.sleep(5)

print(f"Gate: {status.gate}")