LangSmith Evals at Scale: Catch Prompt Regressions for 97% Less

import os
import time
import json
from langsmith import Client as LangSmithClient
from doubleword import Client as DoublewordClient

# 1. Initialize Clients
ls_client = LangSmithClient()  # Uses LANGSMITH_API_KEY
dw_client = DoublewordClient(api_key=os.getenv("DOUBLEWORD_API_KEY"))

DATASET_NAME = "truthful-qa-baseline"
print(f"Fetching dataset '{DATASET_NAME}' from LangSmith...")
examples = list(ls_client.list_examples(dataset_name=DATASET_NAME))

# 2. Prepare the Batch Payload for Doubleword
batch_inputs = []
for ex in examples:
    batch_inputs.append({
        "trace_id": str(ex.id),
        "question": ex.inputs.get("question", ""),
        "app_answer": ex.outputs.get("answer", "") if ex.outputs else "",
        "reference": ex.outputs.get("reference", "") if ex.outputs else "",
    })

print(f"Queueing {len(batch_inputs)} evals to Doubleword's batch API...")

# 3. Dispatch to Doubleword (Saves up to 96% vs Real-Time APIs)
eval_template = """
Evaluate the app's answer against the reference answer.
Question: {question}
App Answer: {app_answer}
Reference: {reference}
Output strictly valid JSON: {"relevance": float(0-1), "truthfulness": float(0-1), "tone": float(0-1)}
"""

batch_job = dw_client.batches.create(
    model="deepseek-ai/DeepSeek-V4-Pro",  # Top-tier judge at batch pricing
    template=eval_template,
    inputs=batch_inputs,
    output_key="eval_scores",
    response_format={"type": "json_object"},
)

# 4. Wait for the Async Processing
print(f"Batch {batch_job.id} running. Doubleword is managing concurrency...")
while True:
    status = dw_client.batches.retrieve(batch_job.id).status
    if status == "completed":
        break
    time.sleep(10)

# 5. Retrieve Results and Log back to LangSmith
print("Batch complete! Pushing evaluation scores to LangSmith...")
results = dw_client.batches.list_results(batch_job.id)

for result in results:
    example_id = result["trace_id"]
    scores = json.loads(result["eval_scores"])
    for key, score in scores.items():
        ls_client.create_feedback(
            run_id=example_id,
            key=key,
            score=score,
        )

print("Success! Check your LangSmith dashboard to view the regression metrics.")