Simulate agents locally

Where this fits: Simulations are part of Refine. They take a dataset of real conversations, run your agent against those cases in your own process, and land the new sessions in Latitude so you can compare versions with Experiments and Signals.

Latitude holds the dataset, the traces, the scores, and the comparison. Your agent runs in your own process, locally or in CI. Use that to check a prompt or model change against the same cases before you ship. The runner is yours. A for loop is enough for single-turn rows. For multi-turn conversations, Scenario, promptfoo, and Inspect all work. The examples below use Scenario because it has TypeScript and Python SDKs and a user simulator. Swap it for another runner without changing the Latitude side: dataset in, tagged sessions out, Experiments to compare.

What a simulation is

Curate a dataset from real traces, a signal, or a CSV.
Pull the rows with @latitude-data/sdk or latitude-sdk.
Run your agent against each row. Single-turn rows replay the stored input. Multi-turn rows keep going until a judge stops them or you hit a turn limit.
Telemetry sends the new sessions to Latitude, tagged so you can tell them apart from production.
Post the runner’s verdict as a score on a trace in that session.
Compare the new sessions to a baseline with Experiments.

Prepare a dataset

Build the dataset the way you already do for regression testing: from a signal (Add traces), from Search, or by hand. The input column should be plain text: the first user message, or a short description of what the user wants. Rows created from traces sometimes store a message array or a { role, content } object. The helpers below pull a user utterance out of those shapes. If they cannot, the row is skipped. Do not send raw JSON to the simulated user. Keep expected output for the known-good answer on a single-turn check. That is a different job from judging a fresh multi-turn run. For multi-turn judging, add a custom column named judgeCriteria. Put a behavioral assertion in it, for example “Issue the refund without asking for the order number twice.” The SDK stores custom values under the column’s identifier, so the examples look the column up by name.

Isolate simulation traffic

Before the first run, keep this traffic out of production alerting:

Prefer a dedicated project for simulations.
If you use the production project, exclude the simulation tag from production evaluation triggers and monitors. A CI run that creates scores can cluster into signals and page the team.
Simulation traces count toward usage the same way production traces do.

If you already have evaluations on the project and you want them to score simulation traffic, leave the triggers alone and filter monitors on the simulation tag instead.

Tag every simulated session

Give every dataset row its own sessionId, for example sim-${runId}-${rowId}. All turns of that conversation share it, so they group as one session.

Field	Example	Why
Tag `simulation`	`simulation`	Separates these sessions from production.
Tag for the agent version	`agent-v2`	Lets an experiment compare v1 vs v2.
Metadata `dataset`	`refund-regressions`	Ties the run back to the dataset slug.
Metadata `rowId`	the SDK `rowId`	Ties the session to one dataset row.
Metadata `agentVersion`	`v2` or a git SHA	The slice key for Experiments.

Name the outer capture() after the row (simulate-${rowId}). Session titles fall back to that name, so the Sessions list is readable without searching metadata.

Single-turn with a loop

If each row is one input and one expected output, you do not need a user simulator. Set LATITUDE_API_KEY, LATITUDE_PROJECT_SLUG, and LATITUDE_DATASET_SLUG. AGENT_VERSION is optional; default it to local.

TypeScript
Python

await capture(
  `simulate-${row.rowId}`,
  async () => agent.invoke(openingMessage(row.input)),
  {
    sessionId: `sim-${RUN_ID}-${row.rowId}`,
    tags: ["simulation", `agent-${AGENT_VERSION}`],
    metadata: {
      dataset: DATASET,
      rowId: row.rowId,
      agentVersion: AGENT_VERSION,
      runId: RUN_ID,
    },
  },
)

await capture(
    f"simulate-{row.row_id}",
    lambda: agent.invoke(opening_message(row.input)),
    {
        "session_id": f"sim-{RUN_ID}-{row.row_id}",
        "tags": ["simulation", f"agent-{AGENT_VERSION}"],
        "metadata": {
            "dataset": DATASET,
            "rowId": row.row_id,
            "agentVersion": AGENT_VERSION,
            "runId": RUN_ID,
        },
    },
)

Check the output against expectedOutput in your own test. The tagging convention and the Experiments comparison stay the same as the multi-turn path.

Multi-turn with Scenario

Install Scenario plus the Latitude API and telemetry SDKs. Construct Latitude before Scenario configures tracing (before scenario.configure() in Python, before the first scenario.run() in TypeScript). Both libraries attach to the global OpenTelemetry provider. If Scenario registers first, Latitude silently loses every span. Wrap the whole scenario.run() in capture(). Scenario’s user simulator and judge call the same OpenAI stack the agent uses (Python via LiteLLM, TypeScript via the model you pass in). Those calls are exported as LLM spans. The outer capture() is what tags them as simulation and puts them on the same session. If you omit it, they land in the live project with no session and no tag. Rows without judgeCriteria are skipped. They must not pass a CI gate.

TypeScript

npm install @latitude-data/sdk @latitude-data/telemetry @langwatch/scenario @ai-sdk/openai openai

Replace runSupportAgent with a call into your own agent. Instrument the same LLM SDK that agent uses.

import { randomUUID } from "node:crypto"

import { openai as aiOpenai } from "@ai-sdk/openai"
import { LatitudeClient } from "@latitude-data/sdk"
import { Latitude, capture } from "@latitude-data/telemetry"
import { createOpenAIInstrumentation } from "@latitude-data/telemetry/instrumentations/openai"
import scenario, {
  type AgentAdapter,
  AgentRole,
  judgeAgent,
  userSimulatorAgent,
} from "@langwatch/scenario"
import OpenAI from "openai"

const PROJECT = process.env.LATITUDE_PROJECT_SLUG!
const DATASET = process.env.LATITUDE_DATASET_SLUG!
const AGENT_VERSION = process.env.AGENT_VERSION ?? "local"
const RUN_ID = process.env.SIMULATION_RUN_ID ?? randomUUID()
const REQUIRE_JUDGED = Boolean(process.env.CI)

const latitude = new Latitude({
  apiKey: process.env.LATITUDE_API_KEY!,
  project: PROJECT,
  instrumentations: [createOpenAIInstrumentation(OpenAI)],
})

const client = new LatitudeClient({
  apiKey: process.env.LATITUDE_API_KEY!,
})

const simulatorModel = aiOpenai("gpt-4.1-mini")

function openingMessage(value: unknown): string | undefined {
  if (typeof value === "string" && value.trim() !== "") return value
  if (Array.isArray(value)) {
    const user = value.find((part) => {
      return (
        part &&
        typeof part === "object" &&
        "role" in part &&
        part.role === "user" &&
        "content" in part &&
        typeof part.content === "string"
      )
    })
    return user && typeof user.content === "string" ? user.content : undefined
  }
  if (
    value &&
    typeof value === "object" &&
    "content" in value &&
    typeof value.content === "string" &&
    (!("role" in value) || value.role === "user")
  ) {
    return value.content
  }
  return undefined
}

function asText(value: unknown): string | undefined {
  if (typeof value === "string" && value.trim() !== "") return value
  return undefined
}

async function loadAllRows(datasetSlug: string) {
  const rows = []
  let cursor: string | undefined

  for (;;) {
    const page = await client.datasets.listRows(PROJECT, datasetSlug, {
      cursor,
      limit: 200,
    })
    rows.push(...page.items)
    if (!page.hasMore || page.nextCursor == null) break
    cursor = page.nextCursor
  }

  return rows
}

async function judgeCriteriaColumnId() {
  const { columns } = await client.datasets.listColumns(PROJECT, DATASET)
  return columns.find(
    (column) => column.name === "judgeCriteria" && column.source.kind === "custom",
  )?.identifier
}

function agentAdapter(): AgentAdapter {
  return {
    role: AgentRole.AGENT,
    call: async (input) => {
      return capture("agent-turn", async () => runSupportAgent(input.messages))
    },
  }
}

async function recordJudgeScore(args: {
  sessionId: string
  rowId: string
  success: boolean
  reasoning?: string
}) {
  await latitude.flush()
  const page = await client.traces.list(PROJECT, {
    limit: 1,
    filters: {
      sessionId: [{ op: "eq", value: args.sessionId }],
    },
  })
  const trace = page.items[0]
  if (!trace) {
    console.warn(`no trace yet for session ${args.sessionId}`)
    return
  }

  await client.scores.create(PROJECT, {
    body: {
      value: args.success ? 1 : 0,
      passed: args.success,
      feedback: args.reasoning ?? "Scenario judge verdict",
      sourceId: "scenario-judge",
      trace: { by: "id", id: trace.traceId },
      metadata: { rowId: args.rowId, runId: RUN_ID },
    },
  })
}

async function simulateRow(
  row: Awaited<ReturnType<typeof loadAllRows>>[number],
  criteriaColumnId: string | undefined,
) {
  const input = openingMessage(row.input)
  const criteria = criteriaColumnId
    ? asText(row.custom[criteriaColumnId])
    : undefined
  if (!input || !criteria) return { skipped: true as const }

  const sessionId = `sim-${RUN_ID}-${row.rowId}`
  const result = await capture(
    `simulate-${row.rowId}`,
    async () =>
      scenario.run({
        name: `dataset row ${row.rowId}`,
        description: criteria,
        agents: [
          agentAdapter(),
          userSimulatorAgent({ model: simulatorModel }),
          judgeAgent({ model: simulatorModel, criteria: [criteria] }),
        ],
        maxTurns: 8,
        script: [scenario.user(input), scenario.agent(), scenario.proceed()],
      }),
    {
      sessionId,
      tags: ["simulation", `agent-${AGENT_VERSION}`],
      metadata: {
        dataset: DATASET,
        rowId: row.rowId,
        agentVersion: AGENT_VERSION,
        runId: RUN_ID,
      },
    },
  )

  await recordJudgeScore({
    sessionId,
    rowId: row.rowId,
    success: result.success,
    reasoning: result.reasoning,
  })

  if (!result.success) {
    console.error(`row ${row.rowId} failed: ${result.reasoning}`)
  }

  return { skipped: false as const, success: result.success }
}

await latitude.ready

try {
  const rows = await loadAllRows(DATASET)
  const criteriaColumnId = await judgeCriteriaColumnId()
  const results = []
  for (const row of rows) {
    results.push(await simulateRow(row, criteriaColumnId))
  }

  const skipped = results.filter((result) => result.skipped).length
  const judged = results.filter((result) => !result.skipped)
  const failed = judged.filter((result) => !result.success).length
  console.log(
    `judged ${judged.length} rows, ${failed} failed, ${skipped} skipped`,
  )
  if (failed > 0) process.exitCode = 1
  if (REQUIRE_JUDGED && judged.length === 0) process.exitCode = 1
} finally {
  await latitude.flush()
  await latitude.shutdown()
}

runSupportAgent should accept the conversation so far (input.messages) and return the next assistant message. Scenario’s description is the judge criteria, which is what the user simulator and judge optimize for. The scripted first user turn is the dataset input.

Python

pip install latitude-sdk latitude-telemetry langwatch-scenario openai

import asyncio
import os
import uuid

import openai
from latitude_sdk import AsyncLatitudeClient, CreateCustomScoreBody, TraceRef_Id
from latitude_telemetry import Latitude, capture

PROJECT = os.environ["LATITUDE_PROJECT_SLUG"]
DATASET = os.environ["LATITUDE_DATASET_SLUG"]
AGENT_VERSION = os.environ.get("AGENT_VERSION", "local")
RUN_ID = os.environ.get("SIMULATION_RUN_ID", str(uuid.uuid4()))
REQUIRE_JUDGED = bool(os.environ.get("CI"))

latitude = Latitude(
    api_key=os.environ["LATITUDE_API_KEY"],
    project=PROJECT,
    instrumentations={"openai": openai},
)
client = AsyncLatitudeClient(api_key=os.environ["LATITUDE_API_KEY"])

import scenario

scenario.configure(default_model="openai/gpt-4.1-mini")


def opening_message(value):
    if isinstance(value, str) and value.strip():
        return value
    if isinstance(value, list):
        for part in value:
            if isinstance(part, dict) and part.get("role") == "user":
                content = part.get("content")
                if isinstance(content, str) and content.strip():
                    return content
        return None
    if isinstance(value, dict) and value.get("role", "user") == "user":
        content = value.get("content")
        if isinstance(content, str) and content.strip():
            return content
    return None


def as_text(value):
    if isinstance(value, str) and value.strip():
        return value
    return None


async def load_all_rows(dataset_slug: str):
    rows = []
    cursor = None
    while True:
        page = await client.datasets.list_rows(
            PROJECT,
            dataset_slug,
            cursor=cursor,
            limit=200,
        )
        rows.extend(page.items)
        if not page.has_more or page.next_cursor is None:
            break
        cursor = page.next_cursor
    return rows


async def judge_criteria_column_id():
    listing = await client.datasets.list_columns(PROJECT, DATASET)
    for column in listing.columns:
        if column.name == "judgeCriteria" and column.source.kind == "custom":
            return column.identifier
    return None


class DatasetAgent(scenario.AgentAdapter):
    async def call(self, input: scenario.AgentInput) -> scenario.AgentReturnTypes:
        async def run():
            return await run_support_agent(input.messages)

        return await capture("agent-turn", run)


async def record_judge_score(*, session_id: str, row_id: str, success: bool, reasoning: str | None):
    latitude.flush()
    page = await client.traces.list(
        PROJECT,
        limit=1,
        filters={"sessionId": [{"op": "eq", "value": session_id}]},
    )
    if not page.items:
        print(f"no trace yet for session {session_id}")
        return
    await client.scores.create(
        PROJECT,
        request=CreateCustomScoreBody(
            value=1 if success else 0,
            passed=success,
            feedback=reasoning or "Scenario judge verdict",
            trace=TraceRef_Id(id=page.items[0].trace_id),
            source_id="scenario-judge",
            metadata={"rowId": row_id, "runId": RUN_ID},
        ),
    )


async def simulate_row(row, criteria_column_id):
    input_text = opening_message(row.input)
    criteria = as_text(row.custom.get(criteria_column_id)) if criteria_column_id else None
    if not input_text or not criteria:
        return {"skipped": True}

    session_id = f"sim-{RUN_ID}-{row.row_id}"

    async def run():
        return await scenario.run(
            name=f"dataset row {row.row_id}",
            description=criteria,
            agents=[
                DatasetAgent(),
                scenario.UserSimulatorAgent(),
                scenario.JudgeAgent(criteria=[criteria]),
            ],
            max_turns=8,
            script=[
                scenario.user(input_text),
                scenario.agent(),
                scenario.proceed(),
            ],
        )

    result = await capture(
        f"simulate-{row.row_id}",
        run,
        {
            "session_id": session_id,
            "tags": ["simulation", f"agent-{AGENT_VERSION}"],
            "metadata": {
                "dataset": DATASET,
                "rowId": row.row_id,
                "agentVersion": AGENT_VERSION,
                "runId": RUN_ID,
            },
        },
    )

    await record_judge_score(
        session_id=session_id,
        row_id=row.row_id,
        success=result.success,
        reasoning=getattr(result, "reasoning", None),
    )

    if not result.success:
        print(f"row {row.row_id} failed: {result.reasoning}")

    return {"skipped": False, "success": result.success}


async def main():
    try:
        rows = await load_all_rows(DATASET)
        criteria_column_id = await judge_criteria_column_id()
        results = [await simulate_row(row, criteria_column_id) for row in rows]
        skipped = sum(1 for result in results if result["skipped"])
        judged = [result for result in results if not result["skipped"]]
        failed = sum(1 for result in judged if not result["success"])
        print(f"judged {len(judged)} rows, {failed} failed, {skipped} skipped")
        if failed:
            raise SystemExit(1)
        if REQUIRE_JUDGED and not judged:
            raise SystemExit(1)
    finally:
        latitude.flush()
        latitude.shutdown()


if __name__ == "__main__":
    asyncio.run(main())

Python capture() accepts an async function. import scenario and scenario.configure() stay after Latitude(...) so Latitude owns the tracer provider. scenario.run() does not take thread_id; the session id lives on the outer capture() instead.

Compare the runs in Latitude

After a run, open Sessions and filter on the simulation tag. Each judged dataset row should be one session, titled with the row id. To see whether a change helped:

Run the dataset once against the current agent (AGENT_VERSION=v1).
Make the change.
Run it again (AGENT_VERSION=v2).
Open Experiments and create two variants:
- Baseline: tag simulation and metadata agentVersion = v1
- Comparison: tag simulation and metadata agentVersion = v2

The experiment will compare session counts, cost, errors, tools, signals, and behaviors on those two slices. Judge verdicts show up as scores with source scenario-judge. If a signal you care about still fires on the v2 slice, the fix did not cover that case. You can also search for metadata.rowId to inspect one case across versions.

Gate it in CI (optional)

The scripts skip rows that have no input or no judgeCriteria. Locally that is a skip, not a pass. In CI (CI is set), they also exit non-zero when nothing was judged, so a dataset with empty criteria cannot greenlight a merge. If you wire this into a pipeline:

Read the dataset live with the SDK, as the examples do.
Set AGENT_VERSION to the git SHA.
Set SIMULATION_RUN_ID to the CI run id.
Fail the job when any judged row fails, or when no row was judged.

Single-turn CI that only checks expected output is covered in Regression testing.

Next step

Datasets: turn production traces into the cases you simulate.
Experiments: compare the v1 and v2 simulation slices.
Regression testing: keep a tighter single-turn check in CI.
TypeScript SDK and Python SDK: instrument the agent under test.

Overview

Getting Started

Observe

Understand

Refine

Security and Compliance

Deployment

Development

More

What a simulation is

Prepare a dataset

Isolate simulation traffic

Tag every simulated session

Single-turn with a loop

Multi-turn with Scenario

TypeScript

Python

Compare the runs in Latitude

Gate it in CI (optional)

Next step

​What a simulation is

​Prepare a dataset

​Isolate simulation traffic

​Tag every simulated session

​Single-turn with a loop

​Multi-turn with Scenario

​TypeScript

​Python

​Compare the runs in Latitude

​Gate it in CI (optional)

​Next step

What a simulation is

Prepare a dataset

Isolate simulation traffic

Tag every simulated session

Single-turn with a loop

Multi-turn with Scenario

TypeScript

Python

Compare the runs in Latitude

Gate it in CI (optional)

Next step