# Wrap API 参考文档

> 自动从 pixie 源代码文档字符串生成。
> 请勿手动编辑——运行 `uv run python scripts/generate_skill_docs.py`。

`pixie.wrap` —— 面向数据的观测 API。

`wrap()` 在处理管道的命名点观测一个数据值或可调用对象。其行为取决于当前模式：

- **无操作模式**（追踪已禁用，无 eval 注册表）：直接返回 `data`，不做任何改变。
- **追踪模式**（在 `pixie trace` 期间）：写入追踪文件并发出一个 OTel 事件（如果 span 处于活动状态则通过 span event，否则通过 OTel logger），然后返回 `data` 不做改变（或者包装一个可调用对象，使得事件在调用时触发）。
- **评估模式**（eval 注册表已激活）：为 `purpose="input"` 注入依赖数据，为 `purpose="output"` / `purpose="state"` 捕获输出/状态。

---

## CLI 命令

| 命令                                                                                     | 描述                                                                                                                                             |
| ---------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ |
| `pixie trace --runnable <filepath:ClassName> --input <kwargs.json> --output <file.jsonl>` | 使用 JSON 文件中的 kwargs 运行一次 Runnable 并写入追踪文件。`--input` 是一个**文件路径**（非内联 JSON）。                                         |
| `pixie format --input <trace.jsonl> --output <dataset_entry.json>`                        | 将追踪文件转换为格式化数据集条目模板。展示 `input_data`、`eval_input` 和 `eval_output`（实际捕获的输出）。                                      |
| `pixie trace filter <file.jsonl> --purpose input`                                         | 仅打印匹配指定 purpose 的 wrap 事件。每个匹配事件输出一行 JSON。                                                                                 |

---

## 类

### `pixie.Runnable`

```python
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: ...
```

供评估框架使用的结构化 Runnable 协议。`T` 是 `pydantic.BaseModel` 的子类，其字段与数据集 JSON 中的 `input_data` 键对应。

生命周期：

1. `create()` —— 类方法，用于构造并返回一个 runnable 实例。
2. `setup()` —— **异步**，在首次 `run()` 调用之前**仅调用一次**。在此初始化共享资源（例如 `TestClient`、数据库连接）。可选——具有默认的空操作实现。
3. `run(args)` —— **异步**，对**每个数据集条目并发调用**（最多并行 4 个条目）。`args` 是根据 `input_data` 构建的经过验证的 Pydantic 模型。调用应用程序的真实入口点。
4. `teardown()` —— **异步**，在最后一次 `run()` 调用之后**仅调用一次**。释放在 `setup()` 中获取的任何资源。可选——具有默认的空操作实现。

`setup()` 和 `teardown()` 具有默认的空操作实现；仅在需要共享资源时才需重写它们。

**并发性**：`run()` 通过 `asyncio.gather` 并发调用。你的实现**必须是并发安全的**。如果它使用了共享的可变状态（例如 SQLite 连接、内存缓存、文件句柄），请使用 `asyncio.Semaphore` 或 `asyncio.Lock` 进行保护：

```python
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)
```

常见的并发陷阱：

- **SQLite**：对并发写入不安全——使用 `Semaphore(1)` 或启用 WAL 模式的 `aiosqlite`。
- **全局可变状态**：在 `run()` 中修改的模块级 dict/list 需要保护。
- **速率受限的 API**：添加信号量以避免 429 错误。

**导入解析**：项目根目录（即调用 `pixie test` / `pixie trace` 的目录）会在加载 runnable 和 evaluator 之前自动添加到 `sys.path` 中。这意味着你的 runnable 可以使用常规的 `import` 语句来引用项目模块（例如 `from app import service`）。

**示例**：

```python
# 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 服务器示例**（使用异步 HTTP 客户端）：

```python
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`

```python
pixie.wrap(data: 'T', *, purpose: "Literal['input', 'output', 'state']", name: 'str', description: 'str | None' = None) -> 'T'
```

在处理管道的某个点观测一个数据值或数据提供者可调用对象。

`data` 可以是一个纯值，也可以是一个产生值的可调用对象。两种情况下返回类型均为 `T` —— 在无操作模式或追踪模式下，调用者取回与传入时完全相同的类型。

在评估模式下且 `purpose="input"` 时，返回值（或可调用对象）会被替换为反序列化后的注册表值。当 `data` 为可调用对象时，返回的包装器会忽略原始函数，每次调用都返回注入的值；在所有其他模式下，返回的可调用对象会包装原始函数并添加追踪或捕获行为。

参数：
data：一个数据值或数据提供者可调用对象。
purpose：数据点的分类：- "input"：来自外部依赖的数据（数据库记录、API 响应）- "output"：发往外部系统或用户的数据 - "state"：用于评估的中间状态（路由决策等）
name：此数据点的唯一标识符。用作 eval 注册表中的键以及追踪日志中的键。
description：可选的人类可读描述，说明此数据是什么。

返回：
原始数据不做改变（追踪/无操作模式），或注册表值（评估模式下且 purpose="input"）。当 `data` 为可调用对象时，返回值也是可调用的。

---

## 错误类型

### `WrapRegistryMissError`

```python
WrapRegistryMissError(name: 'str') -> 'None'
```

当在 eval 注册表中未找到 wrap(purpose="input") 的名称时抛出。

### `WrapTypeMismatchError`

```python
WrapTypeMismatchError(name: 'str', expected_type: 'type', actual_type: 'type') -> 'None'
```

当反序列化后的注册表值与期望类型不匹配时抛出。

---

## 追踪文件工具函数

用于 wrap 日志条目的 Pydantic 模型及 JSONL 加载工具函数。

`WrapLogEntry` 是记录在 JSONL 追踪文件中的单个 `wrap()` 事件的类型化表示。代码库中有多个地方会加载这些对象——`pixie trace filter` CLI、数据集加载器和验证脚本——因此它们共享同一个模型。

### `pixie.WrapLogEntry`

```python
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) -> None
```

记录到 JSONL 追踪文件中的单个 wrap() 事件。

属性：
type：对于 wrap 事件，始终为 `"wrap"`。
name：wrap 点名称（与 `wrap(name=...)` 匹配）。
purpose：`"input"`、`"output"` 或 `"state"` 之一。
data：序列化后的数据（jsonpickle 字符串）。
description：可选的人类可读描述。
trace_id：OTel 追踪 ID（如有）。
span_id：OTel span ID（如有）。

### `pixie.load_wrap_log_entries`

```python
pixie.load_wrap_log_entries(jsonl_path: 'str | Path') -> 'list[WrapLogEntry]'
```

从 JSONL 文件加载所有 wrap 日志条目。

跳过非 wrap 行（例如 `type=llm_span`）及格式错误的行。

参数：
jsonl_path：JSONL 追踪文件的路径。

返回：
:class:`WrapLogEntry` 对象的列表。

### `pixie.filter_by_purpose`

```python
pixie.filter_by_purpose(entries: 'list[WrapLogEntry]', purposes: 'set[str]') -> 'list[WrapLogEntry]'
```

按 purpose 过滤 wrap 日志条目。

参数：
entries：wrap 日志条目列表。
purposes：要包含的 purpose 值集合。

返回：
过滤后的列表。