LangSmith — Testing, CI & Production¶

uv add --dev "langsmith[pytest]"        # rich output and HTTP caching extras

# tests/llm/test_triage.py
import json

import pytest
from langsmith import expect, traceable
from langsmith import testing as t


@traceable
def triage(ticket: str) -> str:
    return json.dumps({"priority": "P1" if "500" in ticket else "P4"})


@pytest.mark.langsmith
def test_priority_p1():
    ticket = "Checkout returns 500 for all EU users"
    t.log_inputs({"ticket": ticket})                 # becomes the dataset example inputs
    t.log_reference_outputs({"priority": "P1"})      # example reference outputs
    out = json.loads(triage(ticket))
    t.log_outputs(out)                               # run outputs
    t.log_feedback(key="json_valid", score=1)        # extra feedback next to the automatic `pass`
    assert out["priority"] == "P1"


@pytest.mark.langsmith(output_keys=["expected"])     # parametrized args: `expected` is a reference output
@pytest.mark.parametrize("ticket,expected", [("Typo on About page", "P4"), ("API 500 on login", "P1")])
def test_priority_param(ticket, expected):
    out = json.loads(triage(ticket))
    t.log_outputs(out)
    expect(out["priority"]).to_equal(expected)       # logs an `expectation` score and asserts

LANGSMITH_TEST_SUITE="triage-suite" uv run pytest tests/llm

LANGSMITH_TEST_CACHE=tests/cassettes uv run pytest tests/llm     # first run records, later runs replay

# scripts/eval_gate.py
import os
import sys
from collections import defaultdict
from statistics import mean

from langsmith import Client, evaluate

THRESHOLDS = {"priority_match": 0.90, "json_valid": 1.0}      # minimum mean score per feedback key


def triage(inputs: dict) -> dict:
    return call_triage_model(inputs["ticket"])               # application entry point


def priority_match(outputs: dict, reference_outputs: dict) -> bool:
    return outputs["priority"] == reference_outputs["priority"]


def json_valid(outputs: dict) -> bool:
    return isinstance(outputs, dict) and {"priority", "team"} <= outputs.keys()


def check(rows: list[dict], thresholds: dict[str, float]) -> list[str]:
    scores: dict[str, list[float]] = defaultdict(list)
    failures = []
    for row in rows:
        if row["run"].error:
            failures.append(f"run failed on {row['example'].inputs}: {row['run'].error[:120]}")
        for res in row["evaluation_results"]["results"]:
            if res.score is not None:
                scores[res.key].append(float(res.score))
    for key, minimum in thresholds.items():
        got = mean(scores[key]) if scores[key] else 0.0
        if got < minimum:
            failures.append(f"{key}: {got:.2f} < {minimum:.2f}")
    return failures


def main() -> int:
    sha = os.getenv("GITHUB_SHA", "local")[:7]
    client = Client()
    results = evaluate(
        triage,
        data=client.list_examples(dataset_name="triage-regression", splits=["regression"]),
        evaluators=[priority_match, json_valid],
        experiment_prefix=f"ci-{sha}",
        metadata={"git_sha": sha, "branch": os.getenv("GITHUB_REF_NAME", "local")},
        max_concurrency=4,
        num_repetitions=2,
    )
    failures = check(list(results), THRESHOLDS)
    for line in failures:
        print(f"GATE FAILED: {line}")
    return 1 if failures else 0


if __name__ == "__main__":
    sys.exit(main())

name: llm-evals
on:
  pull_request:
    paths: ["src/**", "prompts/**", "tests/llm/**", "scripts/eval_gate.py"]
permissions:
  contents: read
jobs:
  unit:                                    # no secrets: fork PRs, no network
    runs-on: ubuntu-latest
    env:
      LANGSMITH_TRACING: "false"
      LANGSMITH_TEST_TRACKING: "false"
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v6
      - run: uv sync
      - run: uv run pytest tests --ignore=tests/llm -q      # LLM tests need keys: run them in `evals`
  evals:
    if: github.event.pull_request.head.repo.full_name == github.repository
    runs-on: ubuntu-latest
    env:
      LANGSMITH_TRACING: "true"
      LANGSMITH_API_KEY: ${{ secrets.LANGSMITH_API_KEY }}      # service key, workspace-scoped
      LANGSMITH_PROJECT: support-bot-ci
      LANGSMITH_TEST_SUITE: support-bot-pytest
      LANGSMITH_EXPERIMENT: ${{ github.sha }}
      LANGSMITH_EXPERIMENT_METADATA: '{"git_sha": "${{ github.sha }}"}'
      OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v6
      - run: uv sync
      - run: uv run pytest tests/llm -q
      - run: uv run python scripts/eval_gate.py

# tests/conftest.py
import pytest
import langsmith as ls


@pytest.fixture
def capture_trace():
    def _run(fn, *args, **kwargs):
        captured = {}
        with ls.tracing_context(enabled="local"):                 # builds the run tree, uploads nothing
            result = fn(*args, langsmith_extra={"on_end": lambda rt: captured.setdefault("root", rt)}, **kwargs)
        return result, captured["root"]
    return _run


def flatten(run):
    yield run
    for child in run.child_runs:
        yield from flatten(child)

def test_rag_retrieves_before_generating(capture_trace):
    answer, root = capture_trace(rag, "How do I reset my password?")
    steps = [(r.name, r.run_type) for r in flatten(root)]
    assert steps == [("rag", "chain"), ("search_kb", "retriever"), ("answer_llm", "llm")]
    assert root.error is None
    llm = next(r for r in flatten(root) if r.run_type == "llm")
    assert llm.metadata["ls_model_name"] == "gpt-4o-mini"

import asyncio
import datetime as dt
import time

from langsmith import Client

client = Client()


async def names_in_trace(project_id: str, trace_id: str) -> set[str]:
    since = dt.datetime.now(dt.timezone.utc) - dt.timedelta(hours=1)    # window defaults to 24 h if omitted
    return {
        run.name
        async for run in client.runs.query(
            project_ids=[project_id], trace_id=trace_id, min_start_time=since, selects=["ID", "NAME", "RUN_TYPE", "STATUS"]
        )
    }


def wait_for_trace(project_id: str, trace_id: str, expected: set[str], timeout: float = 60.0) -> set[str]:
    deadline = time.monotonic() + timeout
    names: set[str] = set()
    while time.monotonic() < deadline:
        names = asyncio.run(names_in_trace(project_id, trace_id))
        if expected <= names:
            return names
        time.sleep(2)
    raise AssertionError(f"trace {trace_id} is missing {expected - names}")