wehub-skill-sync
222 lines
9.0 KiB
Markdown
222 lines
9.0 KiB
Markdown
# 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 值集合。
|
||
|
||
返回：
|
||
过滤后的列表。
|