Setting the file. One moment.
Subchapter 155.16
references/wrap-api.mdMarkdown9 KBView on GitHub
Auto-generated from pixie source code docstrings. Do not edit by hand — run
uv run python scripts/generate_skill_docs.py.
pixie.wrap — data-oriented observation API.
wrap() observes a data value or callable at a named point in the
processing pipeline. Its behavior depends on the active mode:
data unchanged.pixie trace): writes to the trace file and emits an
OTel event (via span event if a span is active, or via OTel logger
otherwise) and returns data unchanged (or wraps a callable so the
event fires on call).purpose="input", captures output/state for purpose="output"/
purpose="state".| Command | Description |
|---|---|
pixie trace --runnable <filepath:ClassName> --input <kwargs.json> --output <file.jsonl> | Run the Runnable once with kwargs from the JSON file and write a trace file. --input is a file path (not inline JSON). |
pixie format --input <trace.jsonl> --output <dataset_entry.json> | Convert a trace file to a formatted dataset entry template. Shows input_data, eval_input, and eval_output (the real captured output). |
pixie trace filter <file.jsonl> --purpose input | Print only wrap events matching the given purposes. Outputs one JSON line per matching event. |
class pixie.Runnable(Protocol[T]):
@classmethod
def create(cls) -> Runnable[Any]: ...
async def setup(self) -> None: ...
async def run(self, args: T) -> None: ...
async def teardown(self) -> None: ...Protocol for structured runnables used by the evaluation harness. T is a
pydantic.BaseModel subclass whose fields match the input_data keys
in the dataset JSON.
Lifecycle:
create() — class method to construct and return a runnable instance.setup() — async, called once before the first run() call.
Initialize shared resources here (e.g., TestClient, database connections).
Optional — has a default no-op implementation.run(args) — async, called concurrently for each dataset entry
(up to 4 entries in parallel). args is a validated Pydantic model
built from input_data. Invoke the application’s real entry point.teardown() — async, called once after the last run() call.
Release any resources acquired in setup().
Optional — has a default no-op implementation.setup() and teardown() have default no-op implementations;
you only need to override them when shared resources are required.
Concurrency: run() is called concurrently via asyncio.gather. Your
implementation must be concurrency-safe. If it uses shared mutable state
(e.g., a SQLite connection, an in-memory cache, a file handle), protect it
with asyncio.Semaphore or asyncio.Lock:
class AppRunnable(pixie.Runnable[AppArgs]):
_sem: asyncio.Semaphore
@classmethod
def create(cls) -> "AppRunnable":
inst = cls()
inst._sem = asyncio.Semaphore(1) # serialise DB access
return inst
async def run(self, args: AppArgs) -> None:
async with self._sem:
await call_app(args.message)Common concurrency pitfalls:
Semaphore(1) or aiosqlite with WAL mode.run() need protection.Import resolution: The project root directory (where pixie test / pixie trace
is invoked) is automatically added to sys.path before loading runnables and
evaluators. This means your runnable can use normal import statements to
reference project modules (e.g., from app import service).
Example:
# pixie_qa/run_app.py
from pydantic import BaseModel
import pixie
class AppArgs(BaseModel):
user_message: str
class AppRunnable(pixie.Runnable[AppArgs]):
@classmethod
def create(cls) -> "AppRunnable":
return cls()
async def run(self, args: AppArgs) -> None:
from myapp import handle_request
await handle_request(args.user_message)Web server example (using an async HTTP client):
import httpx
from pydantic import BaseModel
import pixie
class AppArgs(BaseModel):
user_message: str
class AppRunnable(pixie.Runnable[AppArgs]):
_client: httpx.AsyncClient
@classmethod
def create(cls) -> "AppRunnable":
return cls()
async def setup(self) -> None:
self._client = httpx.AsyncClient(base_url="http://localhost:8000")
async def run(self, args: AppArgs) -> None:
await self._client.post("/chat", json={"message": args.user_message})
async def teardown(self) -> None:
await self._client.aclose()pixie.wrap(data: 'T', *, purpose: "Literal['input', 'output', 'state']", name: 'str', description: 'str | None' = None) -> 'T'Observe a data value or data-provider callable at a point in the processing pipeline.
data can be either a plain value or a callable that produces a value.
In both cases the return type is T — the caller gets back exactly the
same type it passed in when in no-op or tracing modes.
In eval mode with purpose="input", the returned value (or callable) is
replaced with the deserialized registry value. When data is callable
the returned wrapper ignores the original function and returns the injected
value on every call; in all other modes the returned callable wraps the
original and adds tracing or capture behaviour.
Args: data: A data value or a data-provider callable. purpose: Classification of the data point: - “input”: data from external dependencies (DB records, API responses) - “output”: data going out to external systems or users - “state”: intermediate state for evaluation (routing decisions, etc.) name: Unique identifier for this data point. Used as the key in the eval registry and in trace logs. description: Optional human-readable description of what this data is.
Returns:
The original data unchanged (tracing / no-op modes), or the
registry value (eval mode with purpose=”input”). When data
is callable the return value is also callable.
WrapRegistryMissError(name: 'str') -> 'None'Raised when a wrap(purpose=”input”) name is not found in the eval registry.
WrapTypeMismatchError(name: 'str', expected_type: 'type', actual_type: 'type') -> 'None'Raised when deserialized registry value doesn’t match expected type.
Pydantic model for wrap log entries and JSONL loading utilities.
WrapLogEntry is the typed representation of a single wrap() event
as recorded in a JSONL trace file. Multiple places in the codebase load
these objects — the pixie trace filter CLI, the dataset loader, and
the verification scripts — so they share this single model.
pixie.WrapLogEntry(*, type: str = 'wrap', name: str, purpose: str, data: Any, description: str | None = None, trace_id: str | None = None, span_id: str | None = None) -> NoneA single wrap() event as logged to a JSONL trace file.
Attributes:
type: Always "wrap" for wrap events.
name: The wrap point name (matches wrap(name=...)).
purpose: One of "input", "output", "state".
data: The serialized data (jsonpickle string).
description: Optional human-readable description.
trace_id: OTel trace ID (if available).
span_id: OTel span ID (if available).
pixie.load_wrap_log_entries(jsonl_path: 'str | Path') -> 'list[WrapLogEntry]'Load all wrap log entries from a JSONL file.
Skips non-wrap lines (e.g. type=llm_span) and malformed lines.
Args: jsonl_path: Path to a JSONL trace file.
Returns:
List of :class:WrapLogEntry objects.
pixie.filter_by_purpose(entries: 'list[WrapLogEntry]', purposes: 'set[str]') -> 'list[WrapLogEntry]'Filter wrap log entries by purpose.
Args: entries: List of wrap log entries. purposes: Set of purpose values to include.
Returns: Filtered list.