# Core Concepts
Source: https://docs.trajectory.ai/concepts
Trajectories, Steps, Turns, Messages, and how they fit together.
## The Hierarchy
A single agent conversation maps to this four-level hierarchy:
```mermaid theme={null}
graph TD
T["Trajectory
One per conversation"]
T --> S1["Step 1 (Turn 1)"]
T --> S2["Step 2 (Turn 2)"]
T --> S3["Step 3 (Turn 3)"]
S1 --> M1["system message"]
S1 --> M2["user message"]
S1 --> M3["assistant message"]
S2 --> M4["system message"]
S2 --> M5["user message"]
S2 --> M6["assistant message"]
S2 --> M7["assistant message (tool call)"]
S2 --> M8["tool message (result)"]
S2 --> M9["assistant message (final)"]
S3 --> M10["system message"]
S3 --> M11["user message 1"]
S3 --> M12["assistant message 1"]
S3 --> M13["user message 2"]
S3 --> M14["assistant message 2"]
S3 --> M15["user message 3"]
S3 --> M16["assistant message (tool call)"]
S3 --> M17["tool message (result)"]
S3 --> M18["assistant message (final)"]
```
| Level | What it is | Count |
| -------------- | ----------------------------------------------------------------------------------------------------------- | -------------------------- |
| **Trajectory** | The entire conversation, start to finish | One per conversation |
| **Step** | A snapshot of the conversation at one decision point, containing all messages up to that point (cumulative) | One per turn |
| **Turn** | One user → agent exchange (user speaks, agent thinks/acts/responds) | One or more per trajectory |
| **Message** | A single message: user input, assistant output, tool call, or tool result | One or more per turn |
## Trajectory
A **Trajectory** is the SDK's top-level output — one per conversation. It wraps the full multi-turn agent session into a single, structured object.
| Field | Type | Description |
| ------------------- | ------------------- | ------------------------------------------------------------ |
| `task` | `Task` | Conversation metadata (ID, source, turn count, tokens, cost) |
| `steps` | `list[Step]` | Ordered decision-point snapshots (one per turn) |
| `reward` | `Reward` | Optional aggregate reward signal for the whole conversation |
| `metrics` | `TrajectoryMetrics` | Token counts, tool call stats, error rates |
| `execution_metrics` | `ExecutionMetrics` | Timing data (LLM time, total time) |
| `error` | `str` | Error message if the conversation failed |
## Step & Turn
A **Turn** is one user-to-agent exchange: the user sends a message, the agent may call tools, and the agent responds. Turns are the conceptual building block of a conversation.
A **Step** is a Turn's representation inside a Trajectory. Each Step contains the **cumulative** message history up to that point — not just the messages from that turn, but every message from the start of the conversation. This cumulative design means each Step is a self-contained training example: given this full context, here is what the agent did next.
```mermaid theme={null}
sequenceDiagram
participant User
participant Agent
participant Tool
rect rgb(40, 40, 50)
note right of User: Turn 1 → Step 1
User->>Agent: "What's the weather in SF?"
Agent->>Tool: get_weather("SF")
Tool-->>Agent: 62°F, foggy
Agent-->>User: "It's 62°F and foggy in SF."
end
rect rgb(40, 40, 50)
note right of User: Turn 2 → Step 2
User->>Agent: "Compare that to NYC."
Agent->>Tool: get_weather("NYC")
Tool-->>Agent: 45°F, clear
Agent-->>User: "NYC is 45°F and clear — 17° cooler than SF."
end
```
**Step 1** would contain 4 messages (system, user, tool, assistant). **Step 2** would contain all 4 messages from Step 1 *plus* the 4 new messages from Turn 2 — 8 messages total.
## Message
A **Message** is the atomic unit — a single piece of content in the conversation. Each Message has a `role` that identifies who produced it:
| Role | What it represents | Example |
| ----------- | --------------------------------- | ------------------------- |
| `system` | System prompt | Instructions to the agent |
| `user` | Human input | "What's the weather?" |
| `assistant` | Model output (text or tool calls) | "It's 62°F and foggy." |
| `tool` | Tool execution result | `{"temp": 62}` |
When the assistant invokes a tool, its message has a `tool_calls` list instead of (or in addition to) text content. The matching `tool` message carries a `tool_response` with the result or error.
## How the SDK Builds This
```mermaid theme={null}
graph LR
A[Provider API] -->|fetch| B[Raw Traces]
B -->|parse| C[Turns]
C -->|accumulate messages| D[Steps]
D -->|wrap| E[Trajectory]
E -->|save| F[JSON file]
```
The SDK fetches raw traces from your observability provider, groups them into turns, builds cumulative steps, wraps everything in a Trajectory, and saves it to disk.
See the [API Reference](/sdk/api-reference) for full field definitions on all data models.
# Introduction
Source: https://docs.trajectory.ai/introduction
Turn your traces and telemetry into Trajectories - the core primitive for Continual Learning.
# Trajectory SDK
Trajectory SDK turns raw agent traces and product telemetry into our standardized **Trajectory**
format — the primitive for training, evaluation, and continual learning. Ingest from
observability platforms like [LangSmith](https://smith.langchain.com/), build trajectories from
your own data, or push live telemetry events from your app.
## Install
```bash theme={null}
pip install trajectory-sdk
```
## Key Features
* **Ingest traces from anywhere** — LangSmith today, extensible to any trace source
* **Bring your own data** — build trajectories from CSV/JSONL/OpenAI/Anthropic/Vercel messages with shipped recipes
* **Push product telemetry events** — first-class `TelemetryEvent` primitive with idempotent push
* **Correlate trajectories and events** — caller-owned `trace_id` links uploaded trajectories to the telemetry produced in the same trace
* **Standardized Trajectory schema** — typed dataclasses for messages, tool calls, rewards, and metrics
* **E2E bulk export** — discover all conversations, trigger a LangSmith bulk export, download, parse, and upload in three lines
* **PII redaction** — pluggable transform-based redaction before data leaves your environment
* **Three-line setup** — go from raw traces (or your own data) to the Trajectory platform in minutes
All processing runs locally. Your API key and data never leave your environment.
Export your first trajectories in under 5 minutes.
Build trajectories from CSV / JSONL / OpenAI / Anthropic.
Push product telemetry from your app into Trajectory.
Correlate uploaded trajectories with telemetry via `trace_id`.
Standard LangSmith API import walkthrough.
Full function signatures and parameters.
# Platform
Source: https://docs.trajectory.ai/platform
A web UI for browsing and managing imported traces.
# Trajectory Platform
The Trajectory Platform provides the interface for continual learning, via trajectories imported via the SDK.
The Platform is currently in development. Check back soon for updates.
# Quickstart
Source: https://docs.trajectory.ai/quickstart
Install the SDK and export your first trajectories in under 5 minutes.
## Prerequisites
* Python 3.11 or higher
* A [LangSmith API key](https://smith.langchain.com/) (starts with `lsv2_pt_...`)
* A LangSmith project with existing traces
## Installation
```bash pip theme={null}
pip install trajectory-sdk
```
```bash uv theme={null}
uv add trajectory-sdk
```
## Configure the SDK
Initialize the SDK with your LangSmith credentials and (optionally) a Trajectory API key for uploading:
```python theme={null}
import trajectory_sdk as tj
tj.init(
provider="langsmith",
project_id="your-project-id",
workspace_id="your-workspace-id",
api_key="lsv2_pt_...", # or set LANGSMITH_API_KEY env var
trajectory_api_key="your-trajectory-api-key", # or set TRAJECTORY_API_KEY env var
)
```
| Parameter | Required | Description |
| -------------------- | -------- | --------------------------------------------------------------------------------------------------------------------------- |
| `project_id` | Yes | Your project ID |
| `provider` | No | Trace provider (e.g. `"langsmith"`). Omit for telemetry-only or [bring-your-own-data](/sdk/bring-your-own-data) usage |
| `workspace_id` | No | LangSmith workspace UUID (required for bulk export) |
| `destination_id` | No | Bulk export destination UUID (required for live bulk export) |
| `api_key` | No | Provider API key. Reads from `LANGSMITH_API_KEY` / `LANGCHAIN_API_KEY` env var if omitted |
| `trajectory_api_key` | No | Required for `tj.upload()`, `tj.upload_trace()`, and `tj.push_events()`. Reads from `TRAJECTORY_API_KEY` env var if omitted |
| `transforms` | No | List of transforms applied after building (e.g. [PII redaction](/sdk/pii-redaction)) |
| `debug` | No | Enable debug logging |
## Individual Import
List conversations, pick the ones you want, and import them:
```python theme={null}
conversations = tj.list_conversations(limit=5)
ids = [c.conversation_id for c in conversations]
trajectories = tj.import_conversations(ids)
```
This fetches the full run tree for each conversation, extracts messages, and builds Trajectory objects.
You can also pass IDs directly if you already know which conversations you want:
```python theme={null}
trajectories = tj.import_conversations(["cc_abc123", "cc_def456"])
```
## Bulk Import
Import all conversations from a project at once:
```python theme={null}
trajectories = tj.import_conversations(bulk=True)
```
You can also scope the export to a time window:
```python theme={null}
from datetime import timedelta
trajectories = tj.import_conversations(bulk=True, since=timedelta(hours=1))
```
Bulk import discovers conversations in your project, triggers a LangSmith export, and parses the result. See [Bulk Export](/sdk/bulk-import) for details and how to find your workspace ID.
## Upload
After importing (either way), upload trajectories to the Trajectory platform:
```python theme={null}
tj.upload(trajectories, dataset="my_dataset")
```
`trajectory_api_key` must be set in `tj.init()` (or via the `TRAJECTORY_API_KEY` env var) for upload to work.
## Save to Disk
If you prefer to save locally instead of uploading:
```python theme={null}
tj.save(trajectories, "./exports")
```
Each Trajectory is written as a JSON file named by conversation ID.
## Next Steps
Learn what Trajectories, Steps, and Messages represent.
Export all conversations from a project at once.
Build trajectories without a provider — from CSV, JSONL, OpenAI, Anthropic, etc.
Push product telemetry from your app into Trajectory.
Correlate uploaded trajectories with telemetry via `trace_id`.
Full function signatures and parameters.
# API Reference
Source: https://docs.trajectory.ai/sdk/api-reference
Complete reference for all Trajectory SDK functions, primitives, and helpers.
## Session Management
### `tj.init()`
Configure the SDK. Call once before other functions. For telemetry-only usage, only `project_id`
and `trajectory_api_key` are required; for trace import, also pass `provider` and `api_key`.
```python theme={null}
tj.init(
*,
project_id: str,
provider: str | None = None,
api_key: str | None = None,
workspace_id: str | None = None,
destination_id: str | None = None,
trajectory_api_key: str | None = None,
transforms: list[BaseTransform] | None = None,
debug: bool = False,
) -> 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](/sdk/bulk-import). |
| `destination_id` | `str \| None` | `None` | Bulk export destination ID. Required for live bulk export. |
| `trajectory_api_key` | `str \| None` | `None` | API key for `upload`, `upload_trace`, `push_events`. Falls back to `TRAJECTORY_API_KEY` env var. |
| `transforms` | `list[BaseTransform] \| None` | `None` | Transforms applied after building (e.g. [PII redaction](/sdk/pii-redaction)). |
| `debug` | `bool` | `False` | Enable debug logging. |
***
## Benchmarks
### `tj.append_benchmark_tasks()`
Append tasks from a Harbor directory, Trajectory manifest directory, or `BenchmarkSpec` to an
existing benchmark. This is one logical SDK operation: it handles hashing, artifact uploads,
bounded concurrent requests, retries, and publication internally. Call it repeatedly to grow a
benchmark continuously.
```python theme={null}
result = tj.append_benchmark_tasks(
"bm_...",
"./harbor-tasks",
build_images=True,
)
print(result.added_count, result.updated_count, result.skipped_count, result.task_count)
```
Each task name is its stable key. The SDK asks the API which tasks are new, unchanged, or updated
before uploading anything, so re-running over a mostly-unchanged source uploads only new and
changed tasks. Reusing a task name with different content replaces the live task at a new revision.
There is no import session, operation id, or finalize step to manage. The API key falls back to
`TRAJECTORY_API_KEY`.
| Parameter | Type | Default | Description |
| ----------------- | ------------------------------ | -------- | ------------------------------------------------------------------------------- |
| `bench_id` | `str` | required | Existing benchmark to extend. |
| `source` | `str \| Path \| BenchmarkSpec` | required | Tasks and their artifact references. |
| `root` | `str \| Path \| None` | `None` | Artifact root; required when a `BenchmarkSpec` references local artifact paths. |
| `build_images` | `bool` | `True` | Start builds after all requested batches are appended. |
| `wait_for_images` | `bool` | `False` | Wait for image builds to finish. |
Each batch is published in a single transaction, so a batch is never half-visible. If one batch of
a multi-batch append fails, the batches that already committed stay appended and the call raises;
re-running it re-checks what is present and resumes with only the remainder.
See [`examples/append_benchmark_tasks.py`](https://github.com/Trajectorylabs/trajectory-platform/blob/main/examples/append_benchmark_tasks.py)
for a complete command-line example.
***
## Trace Import
### `tj.list_conversations()`
List available conversations from the configured provider.
```python theme={null}
tj.list_conversations(*, limit: int = 50) -> list[ConversationSummary]
```
### `tj.import_conversations()`
Import conversations and return one `Trajectory` per conversation.
```python theme={null}
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.
```python theme={null}
tj.transform(raw_data: dict, *, provider: str = "langsmith", api_key: str = "", project_id: str = "") -> Trajectory
```
***
## Building Trajectories
See [Bring Your Own Data](/sdk/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.
```python theme={null}
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.
```python theme={null}
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`.
```python theme={null}
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](/sdk/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](/sdk/linking-events-to-trajectories) for the workflow.
### `tj.start_trace()`
Create a `TraceContext` for buffering correlated telemetry and trajectory data.
```python theme={null}
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`).
### `tj.upload_trace()`
Upload trajectories, stamp matching events with their returned `trajectory_id`, then push
events — atomically.
```python theme={null}
tj.upload_trace(
trajectories: Trajectory | list[Trajectory],
events: list[TelemetryEvent],
dataset: str,
organization_id: str | None = None,
chunk_size: int = 1000,
max_retries: int = 3,
) -> dict[str, Any]
```
Each input trajectory must carry a `trace_id`. Returns a dict with `upload`, `push`, `events`
(as actually sent), and `unstamped_events` (events pushed without a `trajectory_id` because no
match was found).
Raises `RuntimeError` if any trajectory's upload failed and there are unstamped events tied to
its `trace_id` — no events are pushed in that case.
***
## Upload & Push
### `tj.upload()`
Upload trajectories via the Trajectory API.
```python theme={null}
tj.upload(
trajectories: Trajectory | list[Trajectory],
dataset: str,
) -> dict[str, Any]
```
Returns a dict:
| 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. |
| `errors` | `list[str]` | Backend error messages. |
| `trajectories` | `list[dict]` | Per-trajectory items with `request_index`, `trace_id`, `trajectory_id`, `status`, etc. — used to map `trace_id → trajectory_id`. |
| `batches` | `int` | Number of HTTP batches. |
| `elapsed_s` | `float` | Wall-clock upload time. |
### `tj.push_events()`
Push telemetry events to the Trajectory API. Invalid events are dropped with logged warnings.
```python theme={null}
tj.push_events(
events: list[TelemetryEvent],
*,
organization_id: str | None = None,
chunk_size: int = 1000,
max_retries: int = 3,
) -> PushResult
```
### `tj.save()`
Save trajectories to JSON files on disk. Each file is named `{output_dir}/{conversation_id}.json`.
```python theme={null}
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](/sdk/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. |
### PushResult
| Field | Type | Description |
| --------- | ----------- | -------------------------------------- |
| `pushed` | `int` | Events accepted by the backend. |
| `skipped` | `int` | Events dropped client-side as invalid. |
| `errors` | `list[str]` | One message per invalid event. |
### 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
```python theme={null}
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"]
```
# Bring Your Own Data
Source: https://docs.trajectory.ai/sdk/bring-your-own-data
Build trajectories from CSV, JSONL, OpenAI, Anthropic, or any custom format — no provider required.
When your traces don't live in LangSmith, build trajectories directly from your own data.
`tj.build_trajectory_from_messages()` takes a flat list of `Message` objects and produces a
fully-formed `Trajectory` ready to upload — same shape as the provider-ingested ones.
The SDK ships recipes (`messages_from_*`) for the most common formats. For anything else,
construct `Message` objects yourself using the same primitives.
## Quickstart
```python theme={null}
import trajectory_sdk as tj
tj.init(project_id="acme", trajectory_api_key="tj_key_...")
trajectories = []
for row in rows:
messages = tj.messages_from_openai_chat(row["messages"])
trajectory = tj.build_trajectory_from_messages(
messages=messages,
conversation_id=row["id"],
data_source="my_export",
)
trajectories.append(trajectory)
tj.upload(trajectories, dataset="my_dataset")
```
No provider credentials are needed — only `trajectory_api_key`.
## Message Recipes
The SDK ships adapters for these source formats:
| Recipe | Source format |
| ----------------------------------------------------------------------- | --------------------------------------------------------------------- |
| [`messages_from_openai_chat`](#messages-from-openai-chat) | OpenAI ChatCompletion (modern `tool_calls` or legacy `function_call`) |
| [`messages_from_anthropic_messages`](#messages-from-anthropic-messages) | Anthropic Messages API content blocks |
| [`messages_from_vercel_ai_sdk`](#messages-from-vercel-ai-sdk) | Vercel AI SDK `UIMessage` / `CoreMessage` |
| [`messages_from_prompt_response`](#messages-from-prompt-response) | Flat prompt/response strings |
| [`messages_from_role_content_pairs`](#messages-from-role-content-pairs) | `[(role, content), ...]` tuples |
**All recipes fail loud on unexpected input.** Silent fallbacks turn parse bugs into silent
data-quality bugs, so unrecognized shapes raise `ValueError` / `TypeError` with descriptive
messages. Fix the input, or write a custom recipe (see [Custom Formats](#custom-formats)).
### messages\_from\_openai\_chat
```python theme={null}
raw = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "What's the weather in SF?"},
{
"role": "assistant",
"content": None,
"tool_calls": [{
"id": "call_1",
"function": {"name": "get_weather", "arguments": '{"city": "SF"}'},
}],
},
{"role": "tool", "tool_call_id": "call_1", "content": '{"temp": 62}'},
{"role": "assistant", "content": "It's 62°F in SF."},
]
messages = tj.messages_from_openai_chat(raw)
```
Handles role normalization (`human → user`, `ai → assistant`), multi-modal content arrays,
modern `tool_calls`, legacy `function_call`, `usage`, `finish_reason`, and `reasoning`
pass-through.
### messages\_from\_anthropic\_messages
```python theme={null}
raw = [
{"role": "user", "content": "What's the weather in SF?"},
{
"role": "assistant",
"content": [
{"type": "text", "text": "Let me check."},
{"type": "tool_use", "id": "tu_1", "name": "get_weather", "input": {"city": "SF"}},
],
},
{
"role": "user",
"content": [
{"type": "tool_result", "tool_use_id": "tu_1", "content": [{"type": "text", "text": "62°F"}]},
],
},
]
messages = tj.messages_from_anthropic_messages(raw)
```
Parallel `tool_result` blocks are fanned out into one `role="tool"` `Message` each so step
builders and tool-failure metrics see every result individually.
### messages\_from\_vercel\_ai\_sdk
```python theme={null}
raw = [
{"role": "user", "content": "What's the weather in SF?"},
{
"role": "assistant",
"parts": [
{"type": "text", "text": "Looking it up."},
{
"type": "tool-invocation",
"toolInvocation": {"toolCallId": "tc_1", "toolName": "get_weather", "args": {"city": "SF"}},
},
],
},
{
"role": "tool",
"parts": [
{"type": "tool-result", "toolCallId": "tc_1", "result": {"temp": 62}},
],
},
]
messages = tj.messages_from_vercel_ai_sdk(raw)
```
Supports both `UIMessage` (`parts`) and `CoreMessage` (`content`) shapes, `text` /
`tool-invocation` / `tool-call` / `tool-result` / `reasoning` parts.
### messages\_from\_prompt\_response
For the simplest case — a single prompt and response, optionally with a system message:
```python theme={null}
messages = tj.messages_from_prompt_response(
prompt="What's 2+2?",
response="4",
system="You are a calculator.",
)
```
### messages\_from\_role\_content\_pairs
For ad-hoc construction from tuples:
```python theme={null}
messages = tj.messages_from_role_content_pairs([
("system", "You are a helpful assistant."),
("user", "Hi"),
("assistant", "Hello!"),
])
```
## Building the Trajectory
```python theme={null}
trajectory = tj.build_trajectory_from_messages(
messages=messages,
conversation_id="conv-1",
data_source="my_export",
reward=tj.build_reward_from_scalar(0.85),
task_metadata={"num_turns": 3, "total_tokens": 1200, "total_cost": 0.012},
start_time="2025-05-01T10:00:00Z",
end_time="2025-05-01T10:00:05Z",
termination_reason="ENV_DONE",
trace_id="req_01HXYZ",
model_id="model_internal_123",
)
```
| Parameter | Type | Required | Description |
| ------------------------- | --------------------------- | -------- | ------------------------------------------------------------------------------------------------------ |
| `messages` | `list[Message]` | Yes | Flat list of messages across all turns. The SDK builds cumulative steps. |
| `conversation_id` | `str` | Yes | Unique conversation identifier. |
| `data_source` | `str` | Yes | Where the trajectory came from (e.g. `"my_export"`, `"prod_agent"`). |
| `reward` | `Reward \| None` | No | Aggregate reward — use `tj.build_reward_from_scalar(value)` for a single number. |
| `task_metadata` | `dict \| None` | No | Recognized keys: `num_turns`, `total_tokens`, `total_cost`, `completion_tokens`. Extras are ignored. |
| `error` | `str \| None` | No | Error message if the conversation failed. Auto-sets `termination_reason` to `"ERROR"` if not provided. |
| `start_time` / `end_time` | `str \| None` | No | ISO timestamps; used to compute `execution_metrics.total_time`. |
| `termination_reason` | `TerminationReason \| None` | No | Defaults to `"ENV_DONE"` or `"ERROR"` based on `error`. |
| `extra_telemetry` | `dict \| None` | No | Extra fields merged into `Trajectory.telemetry.data`. |
| `trace_id` | `str \| None` | No | Caller-owned correlation key for linking to telemetry events. |
| `model_id` | `str \| None` | No | Trajectory model identifier (see [model id helpers](/sdk/api-reference#model-id-helpers)). |
The builder automatically computes cumulative steps, tool-call metrics
(`num_tool_calls` / `num_tool_failures` / `tool_error_rate`), and content-hash / idempotency-key
telemetry fields.
## Custom Formats
If none of the shipped recipes fit your data, write your own. The public helpers cover the
fiddly parts:
| Helper | Purpose |
| ---------------------------------- | ----------------------------------------------------------------------------------------------------------------- |
| `tj.normalize_role(role_str)` | Canonicalize role to `system` / `user` / `assistant` / `tool`. Recognizes aliases like `human`, `ai`, `function`. |
| `tj.flatten_text_content(content)` | Collapse multi-modal `content` (string, list of blocks, dict) into plain text. |
| `tj.parse_tool_arguments(raw)` | Normalize tool-call arguments (dict, JSON string, or `None`) into a dict. |
```python theme={null}
from trajectory_sdk import Message, ToolCall, ToolResponse
def messages_from_my_format(raw):
out = []
for entry in raw:
role = tj.normalize_role(entry["who"])
out.append(Message(
role=role,
content=tj.flatten_text_content(entry.get("body")),
))
return out
```
## Pair With Telemetry
`build_trajectory_from_messages` accepts a `trace_id` so trajectories built this way participate
in the same correlation story as provider-imported ones. Pair it with `tj.start_trace()` and
`tj.upload_trace()` for the cleanest workflow:
```python theme={null}
trace = tj.start_trace()
trajectory = trace.build_trajectory(
messages=tj.messages_from_openai_chat(row["messages"]),
conversation_id=row["id"],
data_source="my_export",
)
trace.event("dataset.row_processed", {"row_id": row["id"]})
tj.upload_trace(trajectory, trace.events, dataset="my_dataset")
```
See [Linking Events to Trajectories](/sdk/linking-events-to-trajectories) for the full
correlation walkthrough.
## Related
Trajectories, Steps, Turns, Messages.
Correlate uploaded trajectories with telemetry via `trace_id`.
Full signatures for builders, recipes, and helpers.
Strip sensitive data before trajectories leave your environment.
# Bulk Export
Source: https://docs.trajectory.ai/sdk/bulk-import
Export all conversations from a LangSmith project, convert to Trajectories, and upload to the Trajectory platform.
Bulk export handles the full pipeline: discover conversations in your LangSmith project, export the complete run trees (including child LLM and tool runs), parse into Trajectories, and upload to the Trajectory platform.
## Prerequisites
Before running a bulk export, you need three pieces of information from LangSmith:
1. **API key** — your LangSmith API key (starts with `lsv2_sk_...` or `lsv2_pt_...`)
2. **Workspace ID** — the tenant/workspace UUID
3. **Destination ID** — a pre-configured bulk export destination UUID
### Finding your Workspace ID
1. Go to [LangSmith](https://smith.langchain.com/) and open **Settings** (gear icon)
2. Under **Workspaces**, select your workspace
3. The workspace ID is in the URL: `https://smith.langchain.com/o/.../workspaces/{workspace_id}`
Alternatively, you can find it via the API:
```python theme={null}
from langsmith import Client
client = Client(api_key="lsv2_sk_...")
import requests
resp = requests.get(
"https://api.smith.langchain.com/workspaces",
headers={"X-API-Key": "lsv2_sk_..."},
)
for ws in resp.json():
print(f"{ws['display_name']}: {ws['id']}")
```
### Finding your Destination ID
A bulk export destination tells LangSmith where to write the exported data (e.g. a GCS bucket). Destinations are configured in the LangSmith UI:
1. Go to **Settings** → **Bulk Exports** in LangSmith
2. Create or select an export destination (e.g. a GCS bucket)
3. Copy the destination ID
You can also list existing destinations via the API:
```python theme={null}
import requests
resp = requests.get(
"https://api.smith.langchain.com/api/v1/bulk-export-destinations",
headers={
"X-API-Key": "lsv2_sk_...",
"X-Tenant-Id": "your-workspace-id",
},
)
for dest in resp.json():
print(f"{dest['display_name']}: {dest['id']}")
```
### Finding your Project ID
1. Go to your project in [LangSmith](https://smith.langchain.com/)
2. The project ID is in the URL: `https://smith.langchain.com/o/.../projects/p/{project_id}`
## E2E Bulk Export
With all three IDs, the full pipeline is three lines:
```python theme={null}
import trajectory_sdk as tj
tj.init(
provider="langsmith",
api_key="lsv2_sk_...", # or set LANGSMITH_API_KEY env var
project_id="your-project-id",
workspace_id="your-workspace-id",
destination_id="your-destination-id",
trajectory_api_key="your-trajectory-api-key", # or set TRAJECTORY_API_KEY env var
)
# Export everything, parse, and return Trajectories
trajectories = tj.import_conversations(bulk=True)
# Upload to the Trajectory platform
tj.upload(trajectories, dataset="my_dataset")
```
### What happens under the hood
1. **Discover trace IDs** — lists all root runs in the project and collects their `trace_id` values. This is necessary because child runs (LLM calls, tool calls) don't carry `thread_id` metadata, so filtering by thread ID alone would miss them.
2. **Trigger bulk export** — sends a `POST` to the LangSmith bulk exports API with an `in(trace_id, [...])` filter, which captures the full run tree for each conversation (parents and all children).
3. **Poll for completion** — the export job runs asynchronously. The SDK polls every 5 seconds until it completes (typically 1-2 minutes).
4. **Download parquet** — fetches the exported parquet file from the configured GCS destination bucket.
5. **Parse into Trajectories** — groups runs by `conversation_id` (from metadata) or `trace_id`, builds run trees, extracts messages, and constructs Trajectory objects using multiprocessing for speed.
## Time-Scoped Export
Export only recent conversations by passing the `since` parameter:
```python theme={null}
from datetime import timedelta
# Export conversations from the last hour
trajectories = tj.import_conversations(bulk=True, since=timedelta(hours=1))
```
You can also pass an absolute `datetime`:
```python theme={null}
from datetime import datetime
trajectories = tj.import_conversations(
bulk=True,
since=datetime(2025, 3, 1),
)
```
When `since` is omitted, the SDK exports all conversations in the project.
## From a Local Parquet File
If you already have a parquet file (e.g. from a previous export or manual download), you can skip the export steps:
```python theme={null}
import trajectory_sdk as tj
tj.init(
provider="langsmith",
project_id="your-project-id",
)
trajectories = tj.import_conversations(
bulk=True,
source="./langsmith_export.parquet",
)
tj.save(trajectories, "./exports")
```
No `workspace_id` or `destination_id` is needed when providing a local file.
## Upload
After importing, upload trajectories to the Trajectory platform:
```python theme={null}
tj.upload(trajectories, dataset="my_dataset")
```
`trajectory_api_key` must be set in `tj.init()` (or via the `TRAJECTORY_API_KEY` env var) for upload to work.
## Limits and Caveats
* The LangSmith bulk export API has a maximum of 100 runs per page when listing root runs. The SDK auto-paginates through all pages.
* Export jobs are asynchronous and typically take 1-2 minutes to complete.
* The `destination_id` must point to a GCS bucket that your service account can read from.
* Empty conversations (root runs with no child LLM/tool runs) produce Trajectories with 0 steps. The upload step automatically skips these.
# Import Traces
Source: https://docs.trajectory.ai/sdk/import-traces
Standard workflow: connect to a provider, list conversations, and import them as Trajectories.
## Prerequisites
* `pip install trajectory-sdk`
* A [LangSmith API key](https://smith.langchain.com/) (starts with `lsv2_pt_...`)
* A LangSmith project ID with existing traces
## Walkthrough
```python theme={null}
import trajectory_sdk as tj
# 1. Initialize
tj.init(
provider="langsmith",
api_key="lsv2_pt_...",
project_id="your-project-id",
trajectory_api_key="your-trajectory-api-key", # required for upload
)
# 2. List conversations
conversations = tj.list_conversations(limit=5)
# 3. Extract IDs and import as Trajectories
ids = [c.conversation_id for c in conversations]
trajectories = tj.import_conversations(ids)
# 4. Upload to the Trajectory platform
tj.upload(trajectories, dataset="my-dataset-v1")
```
You can also save to disk instead of (or in addition to) uploading:
```python theme={null}
tj.save(trajectories, "./exports")
```
Each conversation becomes one JSON file in `./exports/`, named by conversation ID.
## Import by ID
If you already know which conversations you want:
```python theme={null}
trajectories = tj.import_conversations([
"cc_abc123",
"cc_def456",
])
```
## LangSmith Notes
* Conversations are identified by `conversation_id` or `session_id` in run metadata. Runs without either are treated as standalone conversations.
* Roles are normalized: `human` → `user`, `ai` → `assistant`.
* API responses are cached in-memory for 5 minutes to reduce load on repeated imports.
# Linking Events to Trajectories
Source: https://docs.trajectory.ai/sdk/linking-events-to-trajectories
How TraceContext and trace_id correlate telemetry events with uploaded trajectories.
Telemetry events and trajectories are linked by a caller-owned `trace_id`. The SDK creates it
before upload, stamps it on every `TelemetryEvent` you produce in the same trace context, and
stores it on the `Trajectory`. After upload, the backend returns its own `trajectory_id`, which
the SDK can stamp onto the buffered events for you.
## The Mental Model
| ID | Owner | When known | Purpose |
| ----------------- | ---------------------- | ------------- | ---------------------------------------------------------------------------- |
| `trace_id` | SDK / caller | Before upload | Correlates telemetry events and trajectories for the same application trace. |
| `trajectory_id` | Backend | After upload | Stable backend identifier for the persisted trajectory row. |
| `session_id` | SDK / caller | Before upload | Broader grouping across traces; defaults to `trace_id` in `TraceContext`. |
| `conversation_id` | Caller / source system | Before upload | Source-system conversation identifier — not a correlation key. |
Use one `trace_id` for the things you want correlated together. Usually that means one
`trace_id` per trajectory. If a single product trace intentionally produces multiple trajectory
records, reusing the same `trace_id` links them and their telemetry into one application trace.
You should rarely need to think about `trajectory_id` directly. The SDK fetches it from the
upload response and stamps it onto matching events for you — see [Upload Both Together](#upload-both-together).
## Recommended Workflow
Start with `tj.start_trace()` when instrumenting a product or agent run. The returned
`TraceContext` owns the `trace_id` and a buffer of `TelemetryEvent` objects.
```python theme={null}
import trajectory_sdk as tj
tj.init(project_id="acme", trajectory_api_key="tj_key_...")
trace = tj.start_trace()
trace.event("tool_call", {"tool": "search"})
trajectory = trace.build_trajectory(
messages=messages,
conversation_id="conv-1",
data_source="prod_agent",
)
```
At this point `trajectory.trace_id` and every event in `trace.events` share the same `trace_id`.
## Upload Both Together
`tj.upload_trace()` is the simplest path when you have a trajectory and its buffered telemetry
events at the same time. It uploads the trajectory first, reads the returned `trajectory_id`,
stamps matching events that share the same `trace_id`, then pushes telemetry — atomically.
```python theme={null}
import trajectory_sdk as tj
tj.init(project_id="acme", trajectory_api_key="tj_key_...")
trace = tj.start_trace()
trace.event("tool_call", {"tool": "search"})
trajectory = trace.build_trajectory(
messages=messages,
conversation_id="conv-1",
data_source="prod_agent",
)
result = tj.upload_trace(trajectory, trace.events, "prod")
upload_result = result["upload"]
push_result = result["push"]
```
The return value is a dict with four keys:
| Key | Type | Description |
| ------------------ | ---------------------- | ----------------------------------------------------------------------------- |
| `upload` | `dict` | Raw response from `tj.upload()` (counts, errors, per-trajectory items). |
| `push` | `PushResult` | Counts of pushed / skipped events plus any error messages. |
| `events` | `list[TelemetryEvent]` | The events as actually sent, after `trajectory_id` stamping. |
| `unstamped_events` | `list[TelemetryEvent]` | Events that were pushed without a `trajectory_id` because no match was found. |
### Atomic guarantee
If any input trajectory's upload comes back with `status="error"` (or `status="skipped"` with a
non-null `error` — the hash-conflict case), and there are unstamped events tied to that
`trace_id`, `upload_trace` raises `RuntimeError` and **pushes no events**. Partners who want
best-effort pushes should call `upload` and `push_events` separately.
## Upload Events Before Trajectory
Events can be buffered locally before the trajectory exists. Once `upload` returns the backend
`trajectory_id`, stamp it onto the buffered events before pushing.
```python theme={null}
import dataclasses
import trajectory_sdk as tj
tj.init(project_id="acme", trajectory_api_key="tj_key_...")
trace = tj.start_trace()
trace.event("tool_call", {"tool": "search"})
trace.event("tool_result", {"ok": True})
trajectory = trace.build_trajectory(
messages=messages,
conversation_id="conv-1",
data_source="prod_agent",
)
result = tj.upload(trajectory, dataset="prod")
trajectory_id = result["trajectories"][0]["trajectory_id"]
events = [
dataclasses.replace(event, trajectory_id=trajectory_id)
for event in trace.events
]
tj.push_events(events)
```
## Upload Trajectory Before Events
If the trajectory is uploaded first, read the returned `trajectory_id` and pass it when
creating later events.
```python theme={null}
import trajectory_sdk as tj
tj.init(project_id="acme", trajectory_api_key="tj_key_...")
trace = tj.start_trace()
trajectory = trace.build_trajectory(
messages=messages,
conversation_id="conv-1",
data_source="prod_agent",
)
result = tj.upload(trajectory, dataset="prod")
trajectory_id = result["trajectories"][0]["trajectory_id"]
trace.event("tool_call", {"tool": "search"}, trajectory_id=trajectory_id)
trace.event("tool_result", {"ok": True}, trajectory_id=trajectory_id)
tj.push_events(trace.events)
```
## Bring Your Own Trace ID
`start_trace()` accepts an explicit `trace_id` when you want to use an ID that already exists
in your system (a request ID, an OpenTelemetry trace ID, your own UUID, etc.):
```python theme={null}
trace = tj.start_trace("req_01HXYZ123")
```
If you omit it, the SDK generates a UUID4 hex string.
## Related
What `TelemetryEvent` is, how to construct one, and how `push_events` works.
Build trajectories from CSV / JSONL / OpenAI / Anthropic / Vercel formats.
Full signatures for `start_trace`, `upload_trace`, `push_events`, and friends.
Trajectories, Steps, Turns, and Messages.
# PII Redaction
Source: https://docs.trajectory.ai/sdk/pii-redaction
Strip sensitive data from trajectories before they leave your environment.
Trajectory SDK supports pluggable PII redaction via transforms. Transforms run after trajectory building — your raw data never touches disk unredacted.
## Using a Transform
Pass transforms to `tj.init()`:
```python theme={null}
from trajectory_sdk.transforms.pii_transform import RegexPiiTransform
from trajectory_sdk.primitives.primitives import PiiPolicy
redactor = RegexPiiTransform(PiiPolicy(name="default", rules=["EMAIL", "PHONE"]))
tj.init(
provider="langsmith",
project_id="your-project-id",
transforms=[redactor],
)
# All imports will have PII redacted automatically
trajectories = tj.import_conversations(bulk=True)
```
This works with both standard and bulk imports.
## Writing a Custom Transform
Custom PII transforms extend `BasePiiTransform` and implement two methods:
* `transform(trajectory)` — return a new trajectory with PII removed
* `preview(trajectories)` — dry-run that reports what would be redacted
```python theme={null}
import re
from dataclasses import replace
from trajectory_sdk.transforms.pii_transform import BasePiiTransform
class EmailTransform(BasePiiTransform):
pattern = re.compile(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b')
def transform(self, trajectory):
new_steps = []
for step in trajectory.steps:
new_msgs = [
replace(msg, content=self.pattern.sub("[REDACTED]", msg.content or ""))
for msg in step.messages
]
new_steps.append(replace(step, messages=new_msgs))
return replace(trajectory, steps=new_steps)
def preview(self, trajectories):
count = sum(
len(self.pattern.findall(msg.content or ""))
for t in trajectories for s in t.steps for msg in s.messages
)
return RedactionPreview(
total_rule_counts={"EMAIL": count},
samples=[],
)
```
## Preview Before Redacting
Use `preview()` to check what a transform would catch without modifying any data:
```python theme={null}
redactor = EmailTransform()
report = redactor.preview(trajectories)
print(report.total_rule_counts) # {"EMAIL": 42}
```
## Built-in Rules
`RegexPiiTransform` supports the following `PiiRule` values:
| Rule | What it matches |
| --------------- | ----------------------- |
| `"EMAIL"` | Email addresses |
| `"PHONE"` | Phone numbers |
| `"SSN"` | Social Security Numbers |
| `"CREDIT_CARD"` | Credit card numbers |
# Streaming Traces
Source: https://docs.trajectory.ai/sdk/streaming
Capture traces from live agent runs in real time.
Coming soon.
# Telemetry Events
Source: https://docs.trajectory.ai/sdk/telemetry-events
Push product telemetry events from your app into Trajectory.
`TelemetryEvent` is the SDK's primitive for product telemetry — user actions, tool calls, agent
decisions, anything you'd send to a product analytics tool. Events are pushed independently of
trajectories, and can be correlated to them via `trace_id` (see
[Linking Events to Trajectories](/sdk/linking-events-to-trajectories)).
## Quickstart
For telemetry-only usage, only `project_id` and `trajectory_api_key` are required in `init()` —
no provider credentials needed.
```python theme={null}
import trajectory_sdk as tj
tj.init(project_id="acme", trajectory_api_key="tj_key_...")
result = tj.push_events([
tj.TelemetryEvent(
event_type="user.accept",
session_id="session-123",
properties={"surface": "chat", "model": "gpt-4o"},
),
])
print(result.pushed, result.skipped, result.errors)
```
## Constructing Events
`TelemetryEvent` is a frozen dataclass. Only `event_type` and `session_id` are required; every
other field has a sensible default.
```python theme={null}
event = tj.TelemetryEvent(
event_type="tool.call",
session_id="session-123",
properties={"tool": "search", "query": "refunds policy"},
user_id="user-abc",
trace_id="req_01HXYZ",
trajectory_id="traj_xyz",
source="my_app",
metadata={"app_version": "2.4.0"},
)
```
| Field | Type | Default | Description |
| --------------- | -------------- | ------------- | ------------------------------------------------------------------------------------- |
| `event_type` | `str` | required | Dotted event name (e.g. `user.accept`, `tool.call`). |
| `session_id` | `str` | required | Session grouping. Defaults to `trace_id` when produced via `TraceContext.event(...)`. |
| `properties` | `dict` | `{}` | Free-form event payload. |
| `event_id` | `str` | new UUID4 hex | **Idempotency key** (see below). |
| `timestamp` | `str` | now (UTC ISO) | Event time. Pass an explicit value to backdate. |
| `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. Always set when producing events via `TraceContext`. |
| `source` | `str` | `"sdk"` | Where the event originated. |
| `metadata` | `dict \| None` | `None` | Additional context not part of `properties`. |
## Idempotency
`event_id` is the idempotency key. The default UUID4 means re-runs of the same code produce
different IDs (and the backend will accept both as separate events).
For at-least-once delivery pipelines — webhooks, retries, replay from a queue — set `event_id`
**deterministically** from your own primary key so duplicate pushes are coalesced server-side:
```python theme={null}
import hashlib
def event_id_for(record_id: str) -> str:
return hashlib.sha256(record_id.encode()).hexdigest()
event = tj.TelemetryEvent(
event_type="user.accept",
session_id="session-123",
event_id=event_id_for(record["id"]),
)
```
Re-pushing the same `event_id` is a no-op on the backend.
## Pushing Events
`tj.push_events(events)` validates, chunks, retries, and pushes a batch of events.
```python theme={null}
result = tj.push_events(
events,
organization_id="org_42",
chunk_size=1000,
max_retries=3,
)
```
| Parameter | Type | Default | Description |
| ----------------- | ---------------------- | -------- | ----------------------------------------------------------------------------------------------- |
| `events` | `list[TelemetryEvent]` | required | Events to push. |
| `organization_id` | `str \| None` | `None` | Logged client-side for attribution. The backend derives the authoritative org from the API key. |
| `chunk_size` | `int` | `1000` | Max events per HTTP request. The SDK splits longer lists automatically. |
| `max_retries` | `int` | `3` | Per-chunk retry budget on transient failures. |
The call returns a `PushResult`:
| Field | Type | Description |
| --------- | ----------- | -------------------------------------- |
| `pushed` | `int` | Events accepted by the backend. |
| `skipped` | `int` | Events dropped client-side as invalid. |
| `errors` | `list[str]` | One message per invalid event. |
Invalid events are dropped with a logged warning rather than raising. This matches the
pattern used by `upload()` for empty trajectories: bad inputs don't poison a whole batch.
Inspect `result.errors` to see what was rejected.
## Producing Events via TraceContext
When you're already inside a `TraceContext` (the recommended path for SDK-instrumented agent
runs), use `trace.event(...)` so events automatically inherit the `trace_id`:
```python theme={null}
trace = tj.start_trace()
trace.event("tool.call", {"tool": "search"})
trace.event("tool.result", {"ok": True}, user_id="user-abc")
# trace.events is the accumulated buffer
tj.push_events(trace.events)
```
`TraceContext.event(...)` sets `session_id = trace_id` automatically. Override either via the
keyword arguments if you need a different grouping.
See [Linking Events to Trajectories](/sdk/linking-events-to-trajectories) for the full
workflow, including `tj.upload_trace()` which uploads a trajectory and pushes correlated events
in one atomic call.
## Related
The `trace_id` correlation story end-to-end.
Full signatures for `push_events`, `TelemetryEvent`, `PushResult`.