/langsmith-online-eval-engineering

LangSmith Online Evaluator API Reference

Working code patterns for creating and managing LangSmith online evaluators. All snippets are derived from the online-evals reference scripts and tested against langsmith >= 0.9.8.

Setup

from langsmith import Client

client = Client()  # uses LANGSMITH_API_KEY from environment

Required environment variables:

LANGSMITH_API_KEY=lsv2_pt_...
LANGSMITH_ENDPOINT=https://api.smith.langchain.com   # optional, this is the default

Inspect traces

Trace inspection uses synchronous SDK methods. Use this to discover field names before building an evaluator.

import json
from langsmith import Client

client = Client()
runs = list(client.list_runs(project_name="my-project", limit=5))

for i, run in enumerate(runs):
    print(f"--- Run {i + 1} (id: {run.id}) ---")
    print(f"Name: {run.name}")
    print(f"Run type: {run.run_type}")
    print(f"Input keys: {list(run.inputs.keys()) if run.inputs else 'None'}")
    print(f"Inputs: {json.dumps(run.inputs, indent=2, default=str)[:2000]}")
    print(f"Output keys: {list(run.outputs.keys()) if run.outputs else 'None'}")
    print(f"Outputs: {json.dumps(run.outputs, indent=2, default=str)[:2000]}")

Each run exposes: .id, .name, .run_type, .inputs (dict), .outputs (dict).

Create an LLM-as-judge evaluator

LLM evaluators use a structured prompt pushed to the LangSmith Prompt Hub. All evaluator operations are async.

1. Define the response schema

from pydantic import BaseModel, Field

class ResponseSchema(BaseModel):
    reasoning: str = Field(
        description="Step-by-step reasoning for the evaluation."
    )
    score: bool = Field(
        description="Whether the response meets the criterion."
    )

Put reasoning first so the LLM explains before scoring.

2. Build and push the prompt

from langchain_core.prompts.structured import StructuredPrompt

PROMPT_MESSAGES = [
    ("system", "Evaluate the following response against the criterion: ..."),
    (
        "human",
        "User question: {input}\n\nAssistant response: {output}",
    ),
]

prompt = StructuredPrompt.from_messages_and_schema(
    PROMPT_MESSAGES,
    schema=ResponseSchema.model_json_schema(),
)

url = client.push_prompt("my-evaluator", object=prompt)

3. Create the evaluator

import asyncio

async def create_llm_evaluator():
    created = await client.evaluators.create(
        name="my-evaluator",
        type="llm",
        llm_evaluator={
            "prompt_repo_handle": "my-evaluator",
            "commit_hash_or_tag": "latest",
            "variable_mapping": {
                "input": "input",
                "output": "output",
            },
        },
    )
    return created.evaluator.id

evaluator_id = asyncio.run(create_llm_evaluator())

Variable mapping

variable_mapping connects prompt template variables to top-level trace fields. The keys are template variable names (appearing as {name} in the prompt), and the values are trace field paths.

Common mappings:

Discover available fields using trace inspection above.

Create a code evaluator

Code evaluators run a Python function directly against each trace. The function must be self-contained (no external imports).

Function signature

def perform_eval(run, example=None):
    """
    Parameters:
        run: dict with keys "inputs" (dict), "outputs" (dict), "attachments"
        example: None for online evaluators (only set for dataset evals)

    Returns:
        dict with keys:
            "key": str       -- evaluator identifier
            "score": value   -- bool, int, or float
            "comment": str   -- optional explanation
    """

Important: Two runtime constraints to be aware of:

1. example must default to None. The online evaluator runtime calls perform_eval(run) with a single argument -- example is only passed for dataset evaluators.
2. run is a plain dict, not an object. Use run.get("inputs") and run.get("outputs") -- not run.inputs or run.outputs. Attribute access will raise AttributeError.

Example: check whether output exists

import textwrap

EVALUATOR_CODE = textwrap.dedent("""\
    def perform_eval(run, example=None):
        \"\"\"Check whether the run produced any output.\"\"\"
        outputs = run.get("outputs")
        has_output = (
            outputs is not None
            and len(outputs) > 0
        )
        return {
            "key": "has_output",
            "score": bool(has_output),
            "comment": "Run produced output" if has_output else "Run had no output",
        }
""")

Create the evaluator

async def create_code_evaluator():
    created = await client.evaluators.create(
        name="has-output",
        type="code",
        code_evaluator={
            "code": EVALUATOR_CODE,
            "language": "python",
        },
    )
    return created.evaluator.id

evaluator_id = asyncio.run(create_code_evaluator())

Attach evaluator to a project (run rules)

Run rules connect evaluators to tracing projects. The SDK does not have a dedicated run-rules wrapper, so use httpx against the REST API.

import os
import httpx
from langsmith import Client

client = Client()
api_key = os.environ["LANGSMITH_API_KEY"]
endpoint = os.environ.get("LANGSMITH_ENDPOINT", "https://api.smith.langchain.com")

# Look up project ID
project = client.read_project(project_name="my-project")

# Create run rule
resp = httpx.post(
    f"{endpoint}/api/v1/runs/rules",
    headers={"x-api-key": api_key},
    json={
        "display_name": f"eval-{project.name}",
        "session_id": str(project.id),
        "sampling_rate": 1.0,        # 1.0 = every trace, 0.1 = 10%
        "evaluator_id": EVALUATOR_ID,
        "is_enabled": True,
    },
    timeout=30,
)
resp.raise_for_status()

rule = resp.json()
print(f"Run rule created -> {rule['id']}")

Payload reference

List, update, and delete evaluators

List all evaluators

async def list_evaluators():
    result = await client.evaluators.list()
    for ev in result.evaluators:
        projects = ""
        if ev.run_rules:
            projects = ", ".join(
                r.session_name or str(r.session_id)
                for r in ev.run_rules
                if r.session_id
            )
        print(f"{ev.name:<30} {str(ev.id):<38} {ev.type:<6} {projects}")

asyncio.run(list_evaluators())

Note: The paginated result object uses .evaluators to access the list of evaluator objects. Some older examples may use .data -- this attribute was renamed in recent SDK versions.

Update an evaluator

async def update_evaluator():
    updated = await client.evaluators.update(
        evaluator_id=EVALUATOR_ID,
        name="renamed-evaluator",
    )
    print("Updated ->", updated.evaluator.id)

asyncio.run(update_evaluator())

Delete an evaluator

async def delete_evaluator():
    await client.evaluators.delete(
        evaluator_id=EVALUATOR_ID,
        delete_run_rules=True,  # also removes attached run rules
    )
    print("Deleted ->", EVALUATOR_ID)

asyncio.run(delete_evaluator())

Async patterns

• Evaluator operations (create, list, update, delete) are async -- use asyncio.run() or await in an async context.
• Trace inspection (client.list_runs) and project lookup (client.read_project) are synchronous.
• Run rules use httpx (synchronous POST).
• client.push_prompt is synchronous.