repository·main·Indexed 21 days ago
https://github.com/judgmentlabs/judgevalAn open-source Python SDK for Agent Behavior Monitoring and the continuous improvement of LLM-powered agents. It provides OpenTelemetry-based tracing, prompt-based agent judges for evaluation, and online monitoring to detect and triage production failures. The SDK includes a CLI, an MCP server, and the Judgeval Query Language (JQL) for querying trace history. It supports integrations with LLM providers like OpenAI, Anthropic, Google GenAI, and Together AI, as well as frameworks such as LangGraph, OpenLit, and Claude Agent SDK.
The Judgeval CLI allows you to manage agents, traces, judges, behaviors, and evaluations directly from your terminal. You can query trace history, deploy judges, and run evaluations against production data.
Judgeval includes an MCP (Model Context Protocol) server that allows you to connect Judgment to any MCP-compatible AI tool. This enables you to query traces, invoke judges, and browse detected behaviors directly within an AI assistant or IDE.
Judgeval provides auto-instrumentation and framework support for several major LLM providers and agent frameworks:
The jql_contract directory contains public-safe snapshots of the canonical JQL (Judgment Query Language) contracts used by the system. These snapshots ensure that the Python SDK remains compatible with the upstream services.
There are two primary files:
jql-ir.openapi.json: Contains the public JQL Intermediate Representation (IR) schema-reference closure.public-openapi.json: Defines the judgeval-server public JQL transport contract.Note that the Python package builds automatically regenerate the modules located under src/judgeval/jql from these JSON snapshots.
Install the Judgeval SDK using pip. To authenticate, you must set the JUDGMENT_API_KEY and JUDGMENT_ORG_ID environment variables.
pip install judgeval
export JUDGMENT_API_KEY=...
export JUDGMENT_ORG_ID=...Judgeval uses OpenTelemetry-based tracing to capture inputs, outputs, and LLM token usage. You can instrument your application using Tracer.init(), wrap() for client auto-instrumentation, and the @Tracer.observe() decorator for specific functions.
Tracer.init(project_name="..."): Initializes the tracer for a specific project.wrap(client): Wraps an existing LLM client (e.g., OpenAI()) to enable auto-instrumentation.@Tracer.observe(span_type="..."): Decorates functions to create spans. Common span_type values include "tool" and "agent".from judgeval import Tracer, wrap
from openai import OpenAI
Tracer.init(project_name="my-project")
client = wrap(OpenAI())
@Tracer.observe(span_type="tool")
def search(query: str) -> str:
results = vector_db.search(query)
return results
@Tracer.observe(span_type="agent")
def run_agent(question: str) -> str:
context = search(question)
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": f"{context}\n\n{question}"}],
)
return response.choices[0].message.content
run_agent("What is the capital of the United States?")If there is an intentional upstream contract change in the judgment-mono repository, you must refresh the local snapshots to keep the Python SDK in sync. Use the scripts/generate_jql.py script with the --sync flag, providing the paths to the updated OpenAPI definitions from the upstream services.
python scripts/generate_jql.py --sync \
../judgment-mono/services/data-access-service/openapi.json \
../judgment-mono/services/judgeval-server/openapi.public-jql.jsonThe EvaluatorRunner is an abstract base class used to execute evaluations. To use it, you must create a concrete implementation that defines how to build the evaluation payload and how to submit the evaluation (either to a hosted server or via a local in-process execution).
There are two primary modes of operation determined by the generic type S:
S = str when scorers are identified by strings (typically for server-side execution).S = Judge when using local Judge instances (typically for in-process execution).To implement a runner, you must override:
_build_payload: Constructs the ExampleEvaluationRun object._submit: Handles the actual submission of the evaluation and returns the number of unique examples expected.The base class provides the run method, which orchestrates the lifecycle: building the payload, submitting, polling for results, and displaying them.
from typing import List, TypeVar
from judgeval.evaluation.evaluation_base import EvaluatorRunner
from judgeval.judges import Judge
from judgeval.data.example import Example
from judgeval.internal.api import JudgmentSyncClient
# For local Judge execution
S = Judge
class MyLocalRunner(EvaluatorRunner[S]):
def _build_payload(self, eval_id, project_id, eval_run_name, created_at, examples, scorers):
# Implementation for building payload
pass
def _submit(self, console, project_id, eval_id, examples, scorers, payload, progress):
# Implementation for local execution
return len(examples)
# Usage
runner = MyLocalRunner(client=client, project_id="proj_123", project_name="My Project")
results = runner.run(
examples=my_examples,
scorers=[my_judge_instance],
eval_run_name="test-run",
timeout_seconds=300
)When processing ScoringResult objects, the value field within ScorerData is determined by the score_type defined in the evaluation. The runner handles three main types:
bool_value. The output is converted to a string: "Yes" if True, and "No" if False.str_value. The output is the raw string value.num_value. The output is cast to a float.If a value is missing or of an unexpected type for its score_type, it will return None (displayed as N/A in the console).
When creating a judge with a CategoricalResponse, you must define a subclass that includes a categories class variable. This variable must be a list of Category models. This metadata is required so the system knows the valid possible values for your scorer.
Example pattern for a categorical judge:
from judgeval.hosted.responses import Category, CategoricalResponse
from judgeval import Judge
class MyResponse(CategoricalResponse):
categories = [
Category(value='Passed', description='The agent passed the test'),
Category(value='Not Passed', description='The agent failed the test'),
]
class CategoricalScorer(Judge[MyResponse]):
async def score(self, data: Example) -> MyResponse:
return MyResponse(value='Passed', reason='The agent passed the test')The Dataset class represents a schema-enforced collection of Example objects on the Judgment platform. You can create, retrieve, and manipulate datasets via the client.datasets interface. Datasets are validated server-side against a provided JSON Schema.
Key capabilities include:
Example objects via add_examples(), add_from_json(), or add_from_yaml().# Create a dataset with a schema
dataset = client.datasets.create(
name="golden-set",
schema={
"type": "object",
"properties": {
"input": {"type": "string"},
"expected_output": {"type": "string"},
},
},
)
# Add examples
dataset.add_examples([
Example.create(input="What is AI?", expected_output="Artificial Intelligence"),
])
# Retrieve and iterate
dataset = client.datasets.get(name="golden-set")
for example in dataset:
print(example.properties["input"])The JudgmentSpanProcessor is an extension of OpenTelemetry's BatchSpanProcessor designed specifically for judgment traces. It provides three key capabilities:
SpanContext.emit_partial(). This allows for streaming updates of an in-progress span's state before the span is officially ended.JudgmentBaggageProcessor to handle context propagation automatically.Note: This processor is typically created automatically by Tracer.init(). You should only instantiate it directly if you are building a custom tracing pipeline.
Judgeval provides internal wrappers to patch the Claude Agent SDK, enabling automatic OpenTelemetry tracing for agentic workflows. This includes tracing the high-level agent lifecycle, individual LLM calls, and tool executions (tool use and tool results).
To enable tracing, you can wrap the ClaudeSDKClient class or use the standalone query() function wrapper. The tracing captures:
ToolUseBlock with ToolResultBlock to capture inputs and outputs.