# Step 2b：实现 Runnable

> 关于完整的 `Runnable` 协议和 `wrap()` API，请参阅 `wrap-api.md`。

**目标**：编写一个 Runnable 类，让评估框架能够像真实用户一样调用应用程序。

---

## 核心思路

Runnable 是 `pixie test` 和 `pixie trace` 运行应用程序的方式。可以把它想象成真实用户的程序化替身：它启动应用、发送请求，然后让应用自行处理。评估框架会为每个测试用例调用 `run()` 方法，传入用户的输入参数。应用程序通过其真实代码——真实的路由、真实的提示词组装、真实的 LLM 调用、真实的响应格式化——来处理这些参数，而评估框架则通过第 2a 步中的 `wrap()` 插桩来观察发生的一切。

**这意味着 Runnable 应该很简单。** 它只是将应用程序的真实入口点连接到评估框架的接口。如果你的 Runnable 变得越来越复杂——比如你在构建自定义逻辑、重新实现应用行为、或替换组件——那一定是有问题的。

## 四项要求

### 1. 运行真实的生产代码

Runnable 调用应用程序的实际入口点——也就是真实用户会触发的同一个函数、类或端点。它不会重新实现、偷工减料或替代应用程序的任何部分。

这包括 LLM。应用程序的 LLM 调用必须经过真实的代码路径——不要模拟、伪造或替换应用组件。基于评估的测试的全部意义就在于 LLM 输出是非确定性的，因此你要使用评估器（而不是断言）来打分。如果你用伪造品替换了任何组件，你就消除了真实行为，评估也就毫无意义。

**如果应用因无法解决的环境变量或配置缺失而无法运行，请停止并向用户反馈，让其修复环境配置。** 不要通过模拟组件来绕过问题。

### 2. 用 Pydantic BaseModel 表示启动参数

`run()` 方法接收一个 Pydantic `BaseModel`，其字段从数据集的 `input_data` 中填充。定义包含应用所需字段的子类：

```python
from pydantic import BaseModel

class AppArgs(BaseModel):
    user_message: str
    # 根据应用入口点的需要添加更多字段。
    # 这些字段与数据集的 input_data 键一一对应。
```

**字段必须反映真实用户实际提供的内容。** 请阅读 `pixie_qa/00-project-analysis.md`——其中的「真实输入特征」部分描述了真实输入的复杂度、规模与多样性。设计模型时应使其能够接受该级别的输入，而非简化的玩具版本。

理解用户提供的参数与世界数据之间的边界：

- **用户提供的参数**（BaseModel 上的字段）：真实用户输入或配置的内容——提示词、查询、配置标志、URL、模式定义。
- **世界数据**（由第 2a 步中的 `wrap(purpose="input")` 处理）：应用在执行过程中从外部来源获取的内容——网页、数据库记录、API 响应。这不属于 BaseModel。

| 应用类型           | BaseModel 字段（用户提供）         | 世界数据（wrap 提供）                     |
| ------------------ | ---------------------------------- | ----------------------------------------- |
| 网页爬虫           | URL + 提示词 + 模式定义            | HTML 页面内容                             |
| 研究代理           | 研究问题 + 范围约束                | 源文档、搜索结果                          |
| 客户支持机器人     | 客户的口头消息                     | CRM 中的客户资料、会话存储中的对话历史    |
| 代码审查工具       | PR URL + 审查标准                  | 实际的 diff、文件内容、CI 结果            |

如果某个字段最终保存的是应用本来会自行获取的数据，那它可能应该放在 `wrap(purpose="input")` 调用中，而不是 BaseModel 上。

### 3. 线程安全

`run()` 会被并发调用以处理多条数据集条目（最多并行 4 条）。如果应用使用了共享可变状态——SQLite、基于文件的数据库、全局缓存——请使用 `asyncio.Semaphore` 保护访问：

```python
import asyncio

class AppRunnable(pixie.Runnable[AppArgs]):
    _sem: asyncio.Semaphore

    @classmethod
    def create(cls) -> "AppRunnable":
        inst = cls()
        inst._sem = asyncio.Semaphore(1)
        return inst

    async def run(self, args: AppArgs) -> None:
        async with self._sem:
            await call_app(args.message)
```

仅在应用确实存在共享可变状态时才添加信号量。如果应用使用基于请求的状态（以唯一 ID 为键）或本质上是无状态的，那么并发调用自然就是隔离的。

### 4. 遵循 Runnable 接口

```python
class AppRunnable(pixie.Runnable[AppArgs]):
    @classmethod
    def create(cls) -> "AppRunnable": ...     # 构造实例
    async def setup(self) -> None: ...        # 一次，在第一次 run() 之前
    async def run(self, args: AppArgs) -> None: ...  # 每条数据集条目，并发执行
    async def teardown(self) -> None: ...     # 一次，在最后一次 run() 之后
```

- `create()` —— 类方法，返回一个新实例。使用带引号的返回类型（`-> "AppRunnable"`）以避免前向引用错误。
- `setup()` —— 可选的异步方法；初始化共享资源（HTTP 客户端、数据库连接、服务器）。
- `run(args)` —— 异步方法；每条数据集条目调用一次。在此处调用应用的真正入口点。
- `teardown()` —— 可选的异步方法；清理 `setup()` 中的资源。

## 最小示例

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

仅此而已。Runnable 导入应用的真正入口点并调用它。没有自定义逻辑、没有组件替换、没有巧妙变通。

## 架构特定示例

根据应用程序的运行方式，阅读对应的示例文件：

| 应用类型                            | 入口点                 | 示例文件                                                     |
| ----------------------------------- | ---------------------- | ------------------------------------------------------------ |
| **独立函数**（无服务器）            | Python 函数            | 阅读 `references/runnable-examples/standalone-function.md`   |
| **Web 服务器**（FastAPI、Flask）    | HTTP/WebSocket 端点    | 阅读 `references/runnable-examples/fastapi-web-server.md`    |
| **CLI 应用程序**                    | 命令行调用             | 阅读 `references/runnable-examples/cli-app.md`               |

只阅读与你应用类型匹配的那个示例文件。

## 文件放置

- 将文件放在 `pixie_qa/run_app.py`。
- 数据集中的 `"runnable"` 字段引用：`"pixie_qa/run_app.py:AppRunnable"`。
- 项目根目录会自动加入 `sys.path`，因此可以使用正常的导入方式（`from app import service`）。

## 技术说明

不要在 runnable 文件中使用 `from __future__ import annotations`——它会破坏 Pydantic 对嵌套模型的模型解析。在需要的地方改用带引号的返回类型。

---

## 输出

`pixie_qa/run_app.py` —— Runnable 类。