Skip to content

bogware/scriven

BranchesTags

Repository files navigation

LangSmith Evals WebSocket API

A FastAPI WebSocket service that runs LangSmith evaluations on demand, plus a CLI entrypoint (run_evals.py) for one-shot CI/CD runs (e.g. AWS ECS Fargate tasks triggered after a backend deploy).

Phase 1 ships LLM-as-judge evaluators (Claude with forced tool-use for structured scores, via the Anthropic API by default or Amazon Bedrock) plus a set of heuristic / code evaluators, and supports targets that are either an existing chatbot HTTP API or a fallback Claude LLM call.

Two ways to run it

Mode Use it when Entrypoint
WebSocket server Interactive UI / dashboards / long-lived service python -m app.server
CLI (one-shot) CI/CD gates, ECS RunTask, scheduled cron python run_evals.py

The CLI does not require the WebSocket server — it imports the runner in-process and exits 0/1 based on threshold gates.

What it does

A client opens a WebSocket connection to /ws/evaluate and sends one JSON payload describing the run. The server then:

  1. Validates the payload.
  2. Resolves the requested evaluator names against the registry.
  3. Syncs the LangSmith dataset:
    • if examples are omitted, it just loads the existing dataset;
    • if examples are provided and the dataset doesn't exist, it's created;
    • if the dataset exists, incoming examples are compared via a deterministic fingerprint of (inputs, outputs). If they match the dataset is left alone; otherwise the existing examples are deleted and replaced.
  4. For each example, calls the configured target to produce an output (concurrency capped by max_concurrency).
  5. Runs every evaluator on the example.
  6. Streams a message per example, then a final summary message.

Setup

This project uses uv (Python 3.14, pinned in .python-version):

uv sync                             # creates .venv and installs from uv.lock
cp .env.example .env                # then fill in keys

If you'd rather not install uv, a plain venv works too:

python -m venv .venv
. .venv/Scripts/activate            # Windows: .venv\Scripts\activate
# macOS/Linux: source .venv/bin/activate
pip install -e .                    # installs the project + its dependencies

Required env vars: LANGSMITH_API_KEY, plus ANTHROPIC_API_KEY for the default LLM-as-judge backend.

By default the judge calls Claude directly via the Anthropic API, so ANTHROPIC_API_KEY is all you need to run the examples below. ANTHROPIC_API_KEY is also used when target.type = "llm" (the Claude-as-system-under-test fallback).

To run the judge on Amazon Bedrock instead, set JUDGE_PROVIDER=bedrock (or LLM_PROVIDER=bedrock to switch both judge and target). Bedrock needs AWS credentials with bedrock:InvokeModel in the judge's region (defaults to us-east-1) — supplied however boto3 normally finds them (AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY, an AWS_PROFILE, or an attached IAM role) — plus model access enabled for the cross-region inference profile in JUDGE_MODEL (Bedrock default us.anthropic.claude-sonnet-4-6; Anthropic default claude-sonnet-4-6). See Provider selection for the full resolution order.

For LangSmith Hybrid set LANGSMITH_ENDPOINT to your hybrid URL.

Provider selection

Both the LLM judge and the LLM target (target.type = "llm") can run on either the Anthropic API (anthropic) or Amazon Bedrock (bedrock). Each knob resolves independently:

payload field  ->  specific env  ->  LLM_PROVIDER env  ->  default
Knob Payload field Specific env Default
Judge provider judge.provider JUDGE_PROVIDER anthropic
Judge model judge.model JUDGE_MODEL per-provider
Target provider target.provider TARGET_PROVIDER anthropic
Target model target.model TARGET_MODEL per-provider

LLM_PROVIDER sets the default for both at once. Per-provider model defaults: claude-sonnet-4-6 for anthropic, us.anthropic.claude-sonnet-4-6 for bedrock. The caller must match the model-id format to the provider — there is no auto-translation. The resolved values are echoed back in the started event so a client can audit what the run actually used.

Run

python -m app.server
# or
uvicorn app.server:app --host 0.0.0.0 --port 8000

Payload schema

{
  "dataset_name": "my-dataset",          // required
  "examples": [                           // optional
    {
      "inputs":  { "input": "..." },
      "outputs": { "output": "..." },     // ground truth (optional)
      "metadata": { "topic": "..." }      // per-example metadata (optional)
    }
  ],
  "evaluators": "correctness,semantic_similarity",   // CSV string OR ["..."]
  "metadata": { "client_app": "...", "release": "v1.2.3" },
  "target": {
    "type": "http" | "llm" | "none",
    "url": "https://your-chatbot/api/chat",          // type=http
    "headers": { "Authorization": "Bearer ..." },     // type=http
    "input_key": "input",
    "response_path": "data.message",                  // dot-path into JSON
    "timeout_s": 60.0,
    "model": "claude-sonnet-4-6",                     // type=llm
    "system_prompt": "Answer concisely."              // type=llm
  },
  "experiment_prefix": "my-run",
  "max_concurrency": 4,
  "allow_overwrite": false                            // must be true to replace examples in an existing dataset
}

Rules:

  • dataset_name is always required.
  • If examples is supplied, see the sync rules above.
  • target.type = "none" means each example already includes its outputs; evaluators run against those without hitting any chatbot/LLM.

Available evaluators

LLM-as-judge (Claude via the Anthropic API or Amazon Bedrock; forced submit_evaluation tool call so every response is a typed {score, reasoning} object — no regex JSON extraction): correctness, semantic_similarity, pairwise, faithfulness, answer_relevance, toxicity, conciseness, coherence.

Heuristic / code: exact_match, json_validity, regex_match, key_phrases, trajectory.

GET /evaluators returns the full list.

Per-evaluator data hints

Evaluator Reads from example
correctness outputs (reference) compared to model output
semantic_similarity same
pairwise outputs is treated as the baseline (A); model output is B. Set metadata.pairwise_criterion to override the criterion.
faithfulness inputs.context or inputs.retrieved_context or metadata.context
answer_relevance inputs (the question)
toxicity model output only
conciseness question + model output
coherence model output only
exact_match deep-equality of model output and outputs reference (skips if no reference)
json_validity model output
regex_match metadata.regex against model output
key_phrases metadata.key_phrases (or outputs.key_phrases) inside model output
trajectory outputs.tool_calls (expected) vs model outputs.tool_calls/trajectory

Streamed messages

Every server -> client message is JSON { "type": "...", "data": { ... } }:

type Sent when
started The run was accepted; echoes evaluators + metadata
dataset_synced After create/update/loaded; includes action
run_planned After examples are pulled; gives total + evaluator names
example_completed One example finished. Includes per-evaluator scores and progress
example_failed Target invocation raised; that example is skipped
completed Final summary with per-evaluator mean
error Fatal error — connection then closes

Calling the service from a developer's machine

Two paths, depending on whether you want to drive evals through the long-running WebSocket service or kick off a one-shot ECS task. Both assume the Terraform stack from Deploying to AWS has been applied and your laptop is on the corporate network / VPN (the ALB is internal-facing).

A) WebSocket against the running service

Grab the ALB hostname from Terraform:

ALB=$(terraform -chdir=iac output -raw alb_dns_name)
# WSS_URL is ws://...  by default (HTTP listener),
# or wss://... when acm_certificate_arn is set on the Terraform stack.
WSS_URL="ws://$ALB/ws/evaluate"

Interactive (wscat):

# install once: npm i -g wscat
wscat -c "$WSS_URL"
# now paste a single-line JSON payload at the > prompt, hit Enter,
# and watch the streamed `started` -> `dataset_synced` ->
# `example_completed` -> ... -> `completed` events come back.

A minimal payload (one example, baseline against the LLM target):

{"dataset_name":"demo-qa","examples":[{"inputs":{"input":"What color is the sky on a clear day?"},"outputs":{"output":"Blue."}}],"evaluators":["correctness","answer_relevance"],"target":{"type":"llm"},"allow_overwrite":true}

Programmatic (Python, websockets):

# pip install websockets
import asyncio, json, websockets

PAYLOAD = {
    "dataset_name": "demo-qa",
    "examples": [{"inputs": {"input": "What color is the sky on a clear day?"},
                  "outputs": {"output": "Blue."}}],
    "evaluators": ["correctness", "answer_relevance"],
    "target": {"type": "llm"},
    "allow_overwrite": True,
}

async def run():
    async with websockets.connect("ws://<alb-dns>/ws/evaluate") as ws:
        await ws.send(json.dumps(PAYLOAD))
        async for raw in ws:
            evt = json.loads(raw)
            print(evt["type"], evt["data"])
            if evt["type"] in ("completed", "error"):
                break

asyncio.run(run())

The connection closes after a terminal completed or error event — one payload per connection by design.

B) Ad-hoc ECS one-shot task

Useful when you want to run the CI evaluator suite right now without waiting for a backend deploy. Same image, same task definition the GitHub Actions workflow uses — just kicked off by hand. Pulls all the IDs from Terraform output so you don't have to look them up:

CLUSTER=$(terraform -chdir=iac output -raw ecs_cluster_name)
TASKDEF=$(terraform -chdir=iac output -raw cli_task_definition_family)
TASK_SG=$(terraform -chdir=iac output -raw task_security_group_id)
SUBNETS=$(terraform -chdir=iac output -json private_subnet_ids | jq -r 'join(",")')

aws ecs run-task \
  --cluster "$CLUSTER" \
  --task-definition "$TASKDEF" \
  --launch-type FARGATE \
  --network-configuration "awsvpcConfiguration={subnets=[$SUBNETS],securityGroups=[$TASK_SG],assignPublicIp=DISABLED}" \
  --overrides '{
    "containerOverrides":[{
      "name":"evals",
      "environment":[
        {"name":"DATASET_NAME","value":"demo-qa"},
        {"name":"EVALUATORS","value":"correctness,answer_relevance"},
        {"name":"TARGET_URL","value":"http://your-chatbot.internal/api/chat"},
        {"name":"THRESHOLDS","value":"correctness>=0.5"},
        {"name":"ALLOW_OVERWRITE","value":"true"}
      ]
    }]
  }'

Wait for it to finish and pull logs:

TASK_ARN=...   # from the run-task output above
aws ecs wait tasks-stopped --cluster "$CLUSTER" --tasks "$TASK_ARN"
aws ecs describe-tasks --cluster "$CLUSTER" --tasks "$TASK_ARN" \
  --query 'tasks[0].containers[0].exitCode' --output text

# Streamed JSON events from run_evals.py end up here:
aws logs tail /ecs/langsmith-evals-dev-cli --follow

Exit code 0 = all gates passed, 1 = threshold breach or example error, 2 = configuration error. Your IAM principal needs ecs:RunTask, ecs:DescribeTasks, logs:GetLogEvents, and iam:PassRole on the task execution + task roles for this to work from your machine.

CI/CD: ECS one-shot task

run_evals.py is built for a "deploy backend → spin up an ECS task to run evals → tear it down" flow. It reads all configuration from environment variables (or matching CLI flags), prints one JSON event per line to stdout, and exits non-zero on threshold breach or example failure.

Required env vars (per-run)

Variable Notes
DATASET_NAME LangSmith dataset name
EVALUATORS CSV: correctness,faithfulness,toxicity
TARGET_URL Chatbot HTTP endpoint (omit to use Claude LLM fallback)
THRESHOLDS CSV gates, e.g. correctness>=0.9,toxicity==1.0
METADATA_JSON JSON dict, e.g. {"git_sha":"abc1234"}

Required secrets (Secrets Manager / SSM, mounted on the task definition)

LANGSMITH_API_KEY, optionally LANGSMITH_ENDPOINT. ANTHROPIC_API_KEY is only required when target.type = "llm". Bedrock access for the judge comes from the task IAM role — give it bedrock:InvokeModel on the model ARN you target (and enable model access in the AWS account); no API key needed.

Optional

TARGET_INPUT_KEY, TARGET_RESPONSE_PATH, TARGET_HEADERS_JSON, TARGET_MODEL, TARGET_SYSTEM_PROMPT, TARGET_TIMEOUT_S, EXAMPLES_FILE (path to JSON array, baked into image or mounted), EXPERIMENT_PREFIX, MAX_CONCURRENCY, JUDGE_MODEL (Bedrock model ID, e.g. us.anthropic.claude-sonnet-4-6), ALLOW_OVERWRITE (set true to replace examples in an existing dataset; defaults to false so a run with the wrong DATASET_NAME fails cleanly instead of wiping the dataset), FAIL_ON_EXAMPLE_ERROR (default 1).

Threshold spec

correctness>=0.9, semantic_similarity>=0.85, toxicity==1.0, faithfulness>=0.8

Operators: >= <= > < == !=. Each gate is checked against the mean score of that evaluator across the run. Missing evaluator (no scores produced) counts as a breach.

Exit codes

Code Meaning
0 All gates passed and no example errors
1 Threshold breach or example error
2 Configuration error (bad payload, missing required field, etc.)

Container image

docker build -t langsmith-evals .
docker run --rm \
  -e LANGSMITH_API_KEY \
  -e AWS_ACCESS_KEY_ID -e AWS_SECRET_ACCESS_KEY -e AWS_REGION=us-east-1 \
  -e DATASET_NAME=chatbot-regression \
  -e EVALUATORS="correctness,faithfulness,toxicity" \
  -e TARGET_URL=https://chatbot.example.com/api/chat \
  -e THRESHOLDS="correctness>=0.9,toxicity==1.0" \
  langsmith-evals

(On ECS Fargate the AWS creds come from the task role; locally you supply them via env vars, ~/.aws/credentials, or AWS_PROFILE. Add -e ANTHROPIC_API_KEY only if you set TARGET_URL to empty / use the LLM target.)

ECS RunTask shape

Pre-create a task definition that references your image in ECR and pulls secrets from Secrets Manager. Per run, override only the per-run env vars:

aws ecs run-task \
  --cluster evals-cluster \
  --task-definition langsmith-evals \
  --launch-type FARGATE \
  --network-configuration 'awsvpcConfiguration={subnets=[...],securityGroups=[...],assignPublicIp=ENABLED}' \
  --overrides '{
    "containerOverrides":[{
      "name":"evals",
      "environment":[
        {"name":"DATASET_NAME","value":"chatbot-regression"},
        {"name":"EVALUATORS","value":"correctness,faithfulness,toxicity"},
        {"name":"TARGET_URL","value":"https://chatbot.example.com/api/chat"},
        {"name":"THRESHOLDS","value":"correctness>=0.9,toxicity==1.0"},
        {"name":"METADATA_JSON","value":"{\"git_sha\":\"abc1234\"}"}
      ]
    }]
  }'

Then aws ecs wait tasks-stopped and read tasks[0].containers[0].exitCode — that is the gate result.

GitHub Actions

A working example is in .github/workflows/evals.yml. It triggers on a successful run of your deploy-backend workflow, builds and pushes the image to ECR, registers a fresh task-definition revision pointing at the new image, runs the ECS task with overrides, waits for it to stop, and propagates the container exit code. Configure these once:

  • Repository secret: AWS_OIDC_ROLE_ARN (an IAM role trusted by GitHub OIDC)
  • Repository variable: CHATBOT_URL (default backend URL to evaluate)
  • Update the workflow's env: block (ECS_CLUSTER, ECS_TASK_DEFINITION, ECS_SUBNETS, ECS_SECURITY_GROUPS) with the values produced by the Terraform stack — see Deploying to AWS.

The OIDC role's permission policy needs (at minimum):

{
  "Version": "2012-10-17",
  "Statement": [
    {
      // Push the freshly-built image
      "Effect": "Allow",
      "Action": [
        "ecr:GetAuthorizationToken",
        "ecr:BatchCheckLayerAvailability",
        "ecr:InitiateLayerUpload",
        "ecr:UploadLayerPart",
        "ecr:CompleteLayerUpload",
        "ecr:PutImage"
      ],
      "Resource": "*"
    },
    {
      // Register a new task-def revision pointing at the new image, then run it
      "Effect": "Allow",
      "Action": [
        "ecs:DescribeTaskDefinition",
        "ecs:RegisterTaskDefinition",
        "ecs:RunTask",
        "ecs:DescribeTasks"
      ],
      "Resource": "*"
    },
    {
      // PassRole on the task's execution + task roles so RunTask can assume them
      "Effect": "Allow",
      "Action": "iam:PassRole",
      "Resource": [
        "arn:aws:iam::<account>:role/<ecsTaskExecutionRole>",
        "arn:aws:iam::<account>:role/<ecsTaskRole>"
      ]
    }
  ]
}

ecs:RegisterTaskDefinition is required because ECS does not support overriding containerDefinitions[].image via run-task containerOverrides — the workflow registers a new revision per run so the freshly-built image is actually executed. Without this permission the "Register new task definition revision" step fails with AccessDenied.

Deploying to AWS (Terraform)

The iac/ directory contains a Terraform stack that provisions everything the CI workflow and the WebSocket service need to run on AWS Fargate:

  • ECS cluster (Fargate + Fargate Spot capacity providers, Container Insights on)
  • Two task-definition families: -service for the long-running WebSocket server, -cli for one-shot CI runs
  • ECS service wired to an internal ALB with a /health target-group check, 3600s idle timeout, and 1800s deregistration delay (so long-running WebSocket evals don't get killed during deploys)
  • Task execution + task IAM roles (Bedrock InvokeModel, Secrets Manager read for LANGSMITH_API_KEY / optional ANTHROPIC_API_KEY)
  • CloudWatch log groups, ALB + ECS security groups, HTTP listener (HTTPS auto-switched on when acm_certificate_arn is set)

Required inputs

The Terraform stack does not create networking primitives — bring your own VPC and subnets:

Variable Required Notes
vpc_id yes VPC for the ALB and tasks
alb_subnet_ids yes ≥2 subnets across distinct AZs for the internal ALB
private_subnet_ids yes Subnets for ECS tasks; need egress to Bedrock / LangSmith / ECR (NAT or VPC endpoints)
alb_allowed_cidrs yes CIDR blocks permitted to reach the ALB (your VPC range, VPN range, etc.)
acm_certificate_arn no Provide to switch the ALB listener from HTTP to HTTPS
judge_model no Defaults to us.anthropic.claude-sonnet-4-6

The API-key secret (LangSmith + Anthropic, as a JSON object) is created by the Terraform stack — no *_secret_arn inputs to pass. After terraform apply, populate the real values once:

aws secretsmanager put-secret-value \
  --secret-id "$(terraform output -raw api_keys_secret_name)" \
  --secret-string '{"LANGSMITH_API_KEY":"ls_...","ANTHROPIC_API_KEY":"sk-ant-..."}'

Terraform's lifecycle.ignore_changes on the initial version keeps it from reverting the value on the next apply.

Apply

cd iac
terraform init
terraform plan \
  -var vpc_id=vpc-xxxxxxxx \
  -var 'alb_subnet_ids=["subnet-aaaa","subnet-bbbb"]' \
  -var 'private_subnet_ids=["subnet-cccc","subnet-dddd"]' \
  -var 'alb_allowed_cidrs=["10.0.0.0/8"]'
terraform apply

(Consider configuring the S3 + DynamoDB backend stub in providers.tf before applying in a shared environment.)

Plugging the outputs into the GitHub Actions workflow

After apply, terraform output will give you the names you need:

terraform output ecs_cluster_name              # -> langsmith-evals-dev-cluster
terraform output cli_task_definition_family    # -> langsmith-evals-dev-cli
terraform output task_security_group_id        # -> sg-xxxxxxxx
terraform output private_subnet_ids            # -> ["subnet-cccc","subnet-dddd"]

Update .github/workflows/evals.yml's env: block accordingly:

ECS_CLUSTER:          langsmith-evals-dev-cluster
ECS_TASK_DEFINITION:  langsmith-evals-dev-cli      # the -cli family, NOT -service
ECS_SUBNETS:          subnet-cccc,subnet-dddd
ECS_SECURITY_GROUPS:  sg-xxxxxxxx
CONTAINER_NAME:       evals

Quick test

# CLI mode (no server needed):
DATASET_NAME=demo-qa \
EVALUATORS=correctness,answer_relevance \
THRESHOLDS=correctness>=0.5 \
EXAMPLES_FILE=examples/demo_qa.json \
python run_evals.py

Project layout

app/
  server.py          FastAPI + WebSocket endpoint
  schemas.py         Pydantic models
  dataset_sync.py    Create / update / compare LangSmith datasets
  target.py          Chatbot HTTP target + Claude fallback
  runner.py          Orchestrates the run; emits streamed events
  thresholds.py      CI gating spec parser + checker
  evaluators/
    registry.py      name -> async callable
    llm_judge.py     Claude-on-Bedrock evaluators (forced tool-use)
    heuristic.py     Code / exact-match / regex / JSON / trajectory evaluators
run_evals.py         CLI entrypoint (CI/CD; one-shot ECS task)
Dockerfile           Slim image; default CMD = uvicorn (CLI command override in -cli task def)
iac/                 Terraform: ECS cluster, service, task defs, ALB, IAM, SGs, logs
  variables.tf       Required + optional inputs
  outputs.tf         ALB DNS, cluster name, task-def families, SG / subnet IDs
  ...
.github/workflows/
  evals.yml          GitHub Actions: build image, register task-def revision, run ECS task

Future phases

  • Pluggable judge backend (Bedrock is the current default; OpenAI / direct Anthropic API as alternates).
  • Persisting evaluator scores back onto the LangSmith experiment.
  • Auth (API key / JWT) on the WebSocket handshake.
  • Multi-run batching over a single connection.

About

A FastAPI WebSocket service and CI-friendly CLI for running LangSmith evaluations on demand — bundles LLM-as-judge (Claude) and heuristic evaluators, syncs datasets idempotently, and gates deploys via threshold checks.

Resources

Readme

License

MIT license
Activity
Custom properties

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases

No releases published

Packages

 
 
 

Contributors

Languages