Langsmith API

from langsmith import Client

client = Client()  # uses LANGSMITH_API_KEY from environment

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

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]}")

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."
    )

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)

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())

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
    """

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",
        }
""")

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())

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']}")

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())

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())

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())