API Reference - Trajectory

Session Management

`tj.init()`

Configure provider imports and legacy SDK helpers. The generated API client and the trajectory.lib upload workflows do not require tj.init().

tj.init(
    *,
    project_id: str,
    provider: str | None = None,
    api_key: str | None = None,
    workspace_id: str | None = None,
    destination_id: str | None = None,
    transforms: list[BaseTransform] | None = None,
) -> None

Parameter	Type	Default	Description
`project_id`	`str`	required	Your project ID.
`provider`	`str \| None`	`None`	Trace provider (e.g. `"langsmith"`). Omit for telemetry-only / BYO-data usage.
`api_key`	`str \| None`	`None`	Provider API key. Falls back to `LANGSMITH_API_KEY` / `LANGCHAIN_API_KEY` env vars when `provider` is set.
`workspace_id`	`str \| None`	`None`	LangSmith workspace/tenant ID. Required for bulk export.
`destination_id`	`str \| None`	`None`	Bulk export destination ID. Required for live bulk export.
`transforms`	`list[BaseTransform] \| None`	`None`	Transforms applied after building (e.g. PII redaction).

Harness telemetry

During a live rollout, the harness writes events, rewards, and completion through the generated client. These calls require an API key and an in-progress trajectory_id.

from trajectory import Client

client = Client()
client.trajectories.log_event(
    "traj_...",
    event_id="event_eval_1",
    name="evaluation_completed",
    payload={"all_pass": True},
)
client.trajectories.log_reward(
    "traj_...",
    reward_id="reward_primary",
    name="criterion_pass_rate",
    value=1.0,
)
client.trajectories.complete("traj_...", termination_reason="ENV_DONE")

event_id and reward_id dedupe retries. New events or rewards after the trajectory is terminal are rejected. complete() is idempotent for an already-completed trajectory.

Benchmarks

Runtime-backed benchmark authoring

Author a generated BenchmarkSpec, then use trajectory.lib.benchmarks.submit() to upload artifacts and bounded task parts. It returns a BenchmarkOperation after the upload is accepted; registration and optional runtime image builds continue on the server. This workflow requires the asynchronous ingestion backend to be deployed and enabled.

from pathlib import Path

from trajectory import Client
from trajectory.lib import benchmarks
from trajectory.types.benchmarks import BenchmarkSpec, TaskSpec

manifest = BenchmarkSpec(
    name="example",
    tasks=[
        TaskSpec(
            name="task-1",
            runtime=benchmarks.DockerfileBuild("environment/Dockerfile"),
            run_command="python /app/agent.py",
        )
    ],
)
client = Client()
operation = benchmarks.submit(
    client,
    manifest,
    root=Path("."),
    build_images=True,
    idempotency_key="example-submission-1",
)
print(operation.id)  # Save this ID to reconnect later.
print(operation.refresh().status)

result = operation.result(timeout=1800)
print(result.status.bench_id, result.status.ready)
for task in result.tasks():
    print(task["name"], task["task_id"])

# Reconnect after restarting without uploading again.
operation = benchmarks.get_operation(client, operation.id)

ImageRef() accepts an immutable registry/image@sha256:... reference, a bpt_ blueprint ID, or an im- Modal image ID. RuntimeRef() refers to an existing organization runtime. DockerfileBuild() resolves the Dockerfile relative to root; its parent directory is the build context, and root-level .dockerignore rules determine which files are uploaded. Unresolved runtimes, mutable image references, and empty run commands are rejected before session creation. The runtime owns environment interaction and reports rewards through log_reward.

`trajectory.lib.benchmarks.submit()`

Parameter	Type	Default	Description
`client`	`Client`	required	Generated API client. Reads `TRAJECTORY_API_KEY` and `TRAJECTORY_BASE_URL` by default.
`manifest`	`BenchmarkSpec`	required	Benchmark specification containing a non-empty `tasks` list.
`root`	`Path`	`Path(".")`	Root for local runtime and task files.
`bench_id`	`str \| None`	`None`	Existing benchmark to append to; omit to create one.
`build_images`	`bool`	`False`	Build or resolve runtime images during ingestion.
`idempotency_key`	`str \| None`	`None`	Generated if omitted. Reuse only for identical metadata, task order, and artifact contents.
`progress`	`Callable[[dict[str, int]], None] \| None`	`None`	Upload callback with `uploaded_files`, `uploaded_bytes`, `total_files`, and `total_bytes`.

To add or update tasks, submit another manifest with bench_id set to the existing benchmark ID. Each task name is its stable key. Async append uploads the supplied artifacts and skips unchanged task writes on the server. Each submission has its own operation and idempotency key.

Operation status and results

get_operation(client, operation_id) returns a handle without making a request. operation.refresh() retrieves a BenchmarkIngestionOperationStatus. operation.result(timeout=1800, poll_interval=2) waits for a terminal status and returns a BenchmarkOperationResult. A TimeoutError stops local waiting; server processing continues. Use timeout=None to wait without a deadline. result.status.ready is true only when builds were explicitly requested, the operation succeeded, all submitted tasks registered, and every required runtime is ready. Registration-only success leaves ready false. result.tasks(), runtimes(), failures(), and affected_tasks(runtime_id) iterate paginated results. Failures are typed BenchmarkIngestionFailure objects. Pass cursor=failure.affected_tasks_cursor to affected_tasks() to start at a supplied failure cursor. Partial failure raises BenchmarkPartialFailureError; a failed operation raises BenchmarkOperationError. Both are available from trajectory.lib.benchmark_operations and expose partial_result and failures. Successful registrations and runtimes remain available. See benchmark submission operations for failure handling and limits.

Blocking and prepared-upload helpers

benchmarks.push(client, manifest, root=Path(".")) and benchmarks.append(client, bench_id, manifest, root=Path(".")) submit an operation and wait for registration. They preserve their successful IngestBenchmarkResponse and BenchmarkTaskAppendSummary return types, respectively, and raise on failed or partially failed operations. Use submit() for image builds, custom waiting, and reconnection. trajectory.lib.upload_benchmark() and trajectory.lib.append_benchmark_tasks() accept a prepared BenchmarkSpec or mapping and UploadFile objects. Small requests retain the legacy registration and append flow; large manifests or artifact maps automatically use ingestion operations and wait for completion. The small append path diffs tasks before uploading and commits changed tasks in batches. Its batch limits do not control task parts on the ingestion path.

from trajectory.lib import append_benchmark_tasks

summary = append_benchmark_tasks(client, "bm_...", manifest, artifacts_by_task)
print(summary.added_count, summary.updated_count, summary.skipped_count, summary.task_count)

Parameter	Type	Default	Description
`client`	`Client`	required	Generated API client.
`bench_id`	`str`	required	Existing benchmark to extend.
`manifest`	`BenchmarkSpec \| Mapping[str, Any]`	required	Benchmark specification containing a non-empty `tasks` list.
`artifacts_by_task`	`Mapping[str, Iterable[UploadFile]] \| None`	`None`	Artifact files keyed by task name.
`max_tasks_per_batch`	`int`	`500`	Maximum changed tasks per legacy append batch.
`max_files_per_batch`	`int`	`1024`	Maximum artifact files per legacy append batch.

Successful task writes remain if a later batch or ingestion step fails. Re-running the same input skips unchanged tasks. Async ingestion does not perform the legacy unchanged-artifact upload optimization.

Trace Import

`tj.list_conversations()`

List available conversations from the configured provider.

tj.list_conversations(*, limit: int = 50) -> list[ConversationSummary]

`tj.import_conversations()`

Import conversations and return one Trajectory per conversation.

tj.import_conversations(
    conversations: list[str] | list[ConversationSummary] | None = None,
    *,
    bulk: bool = False,
    runs_query: bool = False,
    source: str | None = None,
    limit: int | None = None,
    since: timedelta | datetime | None = None,
) -> list[Trajectory]

Parameter	Type	Default	Description
`conversations`	`list`	`None`	Conversation IDs or `ConversationSummary` objects.
`bulk`	`bool`	`False`	Use bulk export instead of individual API calls.
`runs_query`	`bool`	`False`	Use the runs-query path (mutually exclusive with `bulk`).
`source`	`str \| None`	`None`	Local parquet file path. When `bulk=True` and `source` is omitted, the SDK triggers a live export.
`limit`	`int \| None`	`None`	Max conversations (bulk mode only).
`since`	`timedelta \| datetime \| None`	`None`	Time window for bulk export.

`tj.transform()`

Transform raw provider data into a Trajectory without fetching. Useful for custom pipelines that already have the source payload.

tj.transform(raw_data: dict, *, provider: str = "langsmith", api_key: str = "", project_id: str = "") -> Trajectory

Building Trajectories

See Bring Your Own Data for the full walkthrough.

`tj.build_trajectory_from_messages()`

Build a Trajectory from a flat list of Message objects. Cumulative steps, tool-call metrics, and telemetry fields (content hash, idempotency key) are computed automatically.

tj.build_trajectory_from_messages(
    messages: list[Message],
    *,
    conversation_id: str,
    data_source: str,
    reward: Reward | None = None,
    task_metadata: dict | None = None,
    error: str | None = None,
    start_time: str | None = None,
    end_time: str | None = None,
    termination_reason: TerminationReason | None = None,
    extra_telemetry: dict | None = None,
    trace_id: str | None = None,
    model_id: str | None = None,
) -> Trajectory

task_metadata recognizes num_turns, total_tokens, total_cost, and completion_tokens; other keys are ignored.

`tj.build_trajectory_from_parsed()`

Build a Trajectory from a ParsedConversation. Provider subclasses delegate to this so the ParsedConversation → Trajectory logic has a single implementation.

tj.build_trajectory_from_parsed(parsed: ParsedConversation) -> Trajectory

`tj.build_reward_from_scalar()`

Build a single-component Reward from a numeric score. scaled_value is normalized to [0, 1] using score_range.

tj.build_reward_from_scalar(
    value: float,
    *,
    name: str = "score",
    score_range: tuple[float, float] = (0.0, 1.0),
    weight: float = 1.0,
) -> Reward

Message Recipes

Adapters from common third-party message formats into Message objects. All recipes fail loud on unexpected input — see Bring Your Own Data for examples.

Function	Source format
`tj.messages_from_openai_chat(raw)`	OpenAI ChatCompletion
`tj.messages_from_anthropic_messages(raw)`	Anthropic Messages API
`tj.messages_from_vercel_ai_sdk(raw)`	Vercel AI SDK (UIMessage / CoreMessage)
`tj.messages_from_prompt_response(prompt, response, *, system=None)`	Flat prompt/response strings
`tj.messages_from_role_content_pairs(pairs)`	`[(role, content), ...]` tuples

Helpers

Function	Purpose
`tj.normalize_role(role_str)`	Canonicalize to `system` / `user` / `assistant` / `tool`. Recognizes aliases (`human`, `ai`, `function`, etc.).
`tj.flatten_text_content(content)`	Collapse multi-modal content (string, list of blocks, or dict) to plain text.
`tj.parse_tool_arguments(raw)`	Normalize tool-call arguments (dict, JSON string, or `None`) into a dict.

Trace Workflow

See Linking Events to Trajectories for the workflow.

`tj.start_trace()`

Create a TraceContext for buffering correlated telemetry and trajectory data.

tj.start_trace(trace_id: str | None = None) -> TraceContext

If trace_id is omitted, the SDK generates a UUID4 hex string.

`TraceContext`

Attribute	Type	Description
`trace_id`	`str`	Correlation key shared by all events and trajectories produced through this context.
`events`	`list[TelemetryEvent]`	Buffered events produced via `.event(...)`.

trace.event(event_type, properties=None, *, user_id=None, trajectory_id=None, source="sdk", metadata=None) -> TelemetryEvent Append a TelemetryEvent to trace.events with trace_id and session_id set to trace.trace_id. trace.build_trajectory(messages, *, data_source, conversation_id=None, ...) -> Trajectory Convenience wrapper around build_trajectory_from_messages that injects trace_id and defaults conversation_id to trace_id. Accepts the same kwargs as build_trajectory_from_messages (minus trace_id).

`upload_trace()`

Upload trajectories, stamp matching events with returned trajectory_id values, then ingest the events. Import this workflow from trajectory.lib.

upload_trace(
    client: Client,
    trajectories: Any | list[Any],
    events: list[TelemetryEvent],
    *,
    dataset: str,
) -> TraceUploadResult

Each input trajectory must carry a unique string trace_id. The result contains typed upload and telemetry summaries plus the event mappings as sent. A trajectory failure raises PartialUploadError; a missing correlation ID raises TraceCorrelationError. In either case, no telemetry is ingested. TraceCorrelationError is also raised when the upload response omits an item for any submitted trajectory, even if the call contained no events. Events are serialized before trajectory upload so their trace_id values can be inspected. An unsupported JSON value raises TypeError before any request is made. Events that serialize but fail telemetry field validation are returned in result.telemetry.failures; valid events in the same call are still ingested and the uploaded trajectories remain stored.

Upload & Ingest

`upload_trajectories()`

Filter empty trajectories, batch valid trajectories, and upload through the generated client. The workflow uses up to eight workers, batches at 1,000 items or 128 MiB, and gives each upload request a 120-second timeout.

upload_trajectories(
    client: Client,
    trajectories: Iterable[Any],
    *,
    dataset: str,
) -> TrajectoryUploadSummary

Returns a typed summary:

Key	Type	Description
`uploaded`	`int`	Trajectories accepted by the backend.
`skipped`	`int`	Total skipped (client + backend).
`client_skipped`	`int`	Empty trajectories filtered before upload.
`backend_skipped`	`int`	Trajectories rejected by the backend.
`client_failures`	`list[BatchItemFailure]`	Empty-trajectory failures detected before requests are sent.
`failures`	`list[BatchItemFailure]`	All structured client and backend failures, with original request indices.
`trajectories`	`list[TrajectoryUploadItem]`	Typed per-trajectory results with `request_index`, `trace_id`, `trajectory_id`, `status`, and optional `failure`.
`batches`	`int`	Number of HTTP batches.
`elapsed_s`	`float`	Wall-clock upload time.

`ingest_events()`

Validate telemetry locally, split valid events into bounded chunks, and ingest them through the generated client.

ingest_events(
    client: Client,
    events: list[Any],
    *,
    chunk_size: int = 1000,
) -> TelemetryIngestSummary

The result contains ingested, total skipped, and structured failures. Local validation requires a non-empty event_type, a non-empty session_id, mapping properties, and an ISO 8601 timestamp. Events that are not mappings or contain unsupported JSON values are also returned as structured failures. Failure indices are mapped back to the original input list. chunk_size must be an integer greater than zero; other types raise TypeError, while zero and negative values raise ValueError.

`tj.save()`

Save trajectories to JSON files on disk. Each file is named {output_dir}/{conversation_id}.json.

tj.save(
    trajectories: Trajectory | list[Trajectory],
    output_dir: str,
) -> None

Model ID Helpers

For chaining a Trajectory-served chat completion back to its trajectory id, the backend returns the model id on the response headers.

Constant / Function	Description
`tj.TRAJECTORY_MODEL_ID_HEADER`	Header name: `"x-trajectory-model-id"`.
`tj.model_id_from_chat_completion_headers(headers)`	Extract the model id from a headers mapping. Returns `None` if absent.
`tj.model_id_from_chat_completion_response(response)`	Extract from a response object with a `.headers` attribute.

Primitives

All primitives are frozen dataclasses unless noted.

Trajectory

Field	Type	Description
`task`	`Task`	Conversation metadata.
`steps`	`list[Step]`	Cumulative decision-point snapshots.
`reward`	`Reward \| None`	Aggregate reward signal.
`metrics`	`TrajectoryMetrics \| None`	Content metrics.
`execution_metrics`	`ExecutionMetrics \| None`	Timing data.
`reference_trajectory`	`dict \| None`	Optional reference for comparison.
`telemetry`	`Telemetry \| None`	Source telemetry metadata.
`idx`	`int \| None`	Index within a batch.
`error`	`str \| None`	Error if the conversation failed.
`trace_id`	`str \| None`	Caller-owned correlation key.
`model_id`	`str \| None`	Trajectory model identifier.

Task

Field	Type	Description
`id`	`str \| None`	Task identifier (`{data_source}:{conversation_id}` when built locally).
`data_source`	`str \| None`	Provider name (e.g. `"langsmith"`) or your own source label.
`conversation_id`	`str \| None`	Unique conversation identifier.
`num_turns`	`int \| None`	Number of user-agent turns.
`num_steps`	`int \| None`	Number of steps.
`total_tokens`	`int \| None`	Total tokens consumed.
`total_cost`	`float \| None`	Estimated cost in USD.

Step

Field	Type	Description
`messages`	`list[Message]`	Cumulative messages up to this point.
`reward`	`Reward \| None`	Per-step reward.
`info`	`dict \| None`	Provider-specific metadata.
`trainable_status`	`TrainableStatus`	Training-suitability flag.

Message

Field	Type	Description
`role`	`Role`	`"system"`, `"user"`, `"assistant"`, or `"tool"`.
`content`	`str \| None`	Text content.
`tool_calls`	`list[ToolCall] \| None`	Tool invocations by the assistant.
`tool_response`	`ToolResponse \| None`	Tool execution result.
`tool_definitions`	`list[ToolDefinition] \| None`	Tools available to the model at this point.
`usage`	`dict \| None`	Token usage stats.
`finish_reason`	`str \| None`	`"stop"`, `"tool_calls"`, etc.
`metadata`	`dict \| None`	Provider-specific message metadata.
`reasoning`	`str \| None`	Chain-of-thought content.
`trainable_status`	`TrainableStatus`	Training-suitability flag.

ToolCall

Field	Type	Description
`name`	`str`	Tool name.
`arguments`	`dict`	Arguments passed to the tool.
`id`	`str \| None`	Identifier matching the `ToolResponse`.

ToolResponse

Field	Type	Description
`id`	`str`	Matches the `ToolCall` id.
`name`	`str`	Tool name.
`arguments`	`dict`	Arguments passed.
`response`	`Any \| None`	Return value.
`error`	`str \| None`	Error if the call failed.
`metadata`	`dict \| None`	Provider-specific metadata.

ToolDefinition

Field	Type	Description
`name`	`str`	Tool name.
`description`	`str`	What the tool does.
`parameters`	`dict`	JSON Schema for tool arguments.

Reward & RewardComponent

Field (Reward)	Type	Description
`aggregated_value`	`float \| None`	Top-level reward score.
`aggregation_method`	`str \| None`	How components were combined.
`components`	`list[RewardComponent] \| None`	Per-rubric / per-judge components.

Field (RewardComponent)	Type	Description
`name`	`str`	Component label.
`value`	`float`	Raw value.
`scaled_value`	`float \| None`	Normalized to `[0, 1]`.
`explanation`	`str \| None`	Free-form notes (e.g. judge rationale).
`weight`	`float`	Weight in the aggregate.
`range`	`tuple[float, float] \| None`	Original score range.
`metadata`	`dict \| None`	Provider-specific context.

TelemetryEvent

See Telemetry Events for the full guide.

Field	Type	Default	Description
`event_type`	`str`	required	Dotted event name.
`session_id`	`str`	required	Session grouping.
`properties`	`dict`	`{}`	Free-form payload.
`event_id`	`str`	new UUID4 hex	Idempotency key.
`timestamp`	`str`	now (UTC ISO)	Event time.
`user_id`	`str \| None`	`None`	End-user identifier.
`trajectory_id`	`str \| None`	`None`	Backend trajectory ID once known.
`trace_id`	`str \| None`	`None`	Caller-owned correlation key.
`source`	`str`	`"sdk"`	Where the event originated.
`metadata`	`dict \| None`	`None`	Additional context.

TelemetryIngestSummary

Field	Type	Description
`ingested`	`int`	Events accepted by the backend.
`skipped`	`int`	Events rejected locally or by the backend.
`failures`	`list[BatchItemFailure]`	Structured validation and backend failures.

Telemetry

Field	Type	Description
`source`	`str`	Telemetry source identifier.
`data`	`dict`	Raw telemetry data (includes `trace_id`, `content_hash`, `idempotency_key` when set).

TrajectoryMetrics & ExecutionMetrics

Field	Type	Description
`steps`	`int \| None`	Total steps.
`tokens_generated`	`int \| None`	Total completion tokens.
`aggregated_reward`	`float \| None`	Aggregate reward value.
`num_tool_calls`	`int \| None`	Total tool invocations.
`num_tool_failures`	`int \| None`	Failed tool calls.
`num_tool_response_none`	`int \| None`	Tool calls with no response.
`tool_error_rate`	`float \| None`	Failure ratio.
`env_time`	`float \| None`	Time in tool execution (seconds).
`llm_time`	`float \| None`	Time waiting for LLM (seconds).
`total_time`	`float \| None`	Wall-clock time (seconds).
`termination_reason`	`TerminationReason \| None`	Why the conversation ended.

ConversationSummary

Field	Type	Description
`conversation_id`	`str`	Unique conversation identifier.
`num_turns`	`int`	Number of turns.
`first_seen`	`str \| None`	ISO timestamp of earliest trace.
`last_seen`	`str \| None`	ISO timestamp of most recent trace.
`root_run_names`	`list[str]`	Names of top-level runs.

Type Aliases

Role = Literal["system", "user", "assistant", "tool"]

TrainableStatus = Literal[
    "trainable", "not_trainable", "superseded", "summarization_boundary"
]

TerminationReason = Literal[
    "TIMEOUT", "ENV_DONE", "MAX_STEPS", "TRUNCATION", "STALE", "ERROR", "NONE"
]

PiiRule = Literal["EMAIL", "PHONE", "CREDIT_CARD", "SSN"]

​Session Management

​tj.init()

​Harness telemetry

​Benchmarks

​Runtime-backed benchmark authoring

​trajectory.lib.benchmarks.submit()

​Operation status and results

​Blocking and prepared-upload helpers

​Trace Import

​tj.list_conversations()

​tj.import_conversations()

​tj.transform()

​Building Trajectories

​tj.build_trajectory_from_messages()

​tj.build_trajectory_from_parsed()

​tj.build_reward_from_scalar()

​Message Recipes

​Helpers

​Trace Workflow

​tj.start_trace()

​TraceContext

​upload_trace()

​Upload & Ingest

​upload_trajectories()

​ingest_events()

​tj.save()

​Model ID Helpers

​Primitives

​Trajectory

​Task

​Step

​Message

​ToolCall

​ToolResponse

​ToolDefinition

​Reward & RewardComponent

​TelemetryEvent

​TelemetryIngestSummary

​Telemetry

​TrajectoryMetrics & ExecutionMetrics

​ConversationSummary

​Type Aliases