skillhub-048-eval-driven-dev/references/wrap-api.md

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

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 进行保护：

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）。

示例：

# 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 客户端）：

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

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

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

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

WrapTypeMismatchError

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

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

---

追踪文件工具函数

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

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

pixie.WrapLogEntry

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

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

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

按 purpose 过滤 wrap 日志条目。

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

返回： 过滤后的列表。