wehub/skillhub-048-eval-driven-dev

Fork 0

Files

T

wehub-skill-sync 4ac4773608 chore: import zh skill eval-driven-dev

2026-07-13 21:35:56 +08:00

7.4 KiB

Raw Permalink Blame History

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 中填充。定义包含应用所需字段的子类：

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 保护访问：

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 接口

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() 中的资源。

最小示例

# 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 类。

7.4 KiB Raw Permalink Blame History Unescape Escape