企业官网建设流程全解析

eval-driven-dev 技能指南：用 pixie.Runnable 驱动真实应用完成 LLM 评测（Step 2b 实战解析）

【免费下载链接】awesome-copilotCommunity-contributed instructions, agents, skills, and configurations to help you make the most of GitHub Copilot.项目地址: https://gitcode.com/GitHub_Trending/aw/awesome-copilot

本文基于 awesome-copilot 仓库中 eval-driven-dev 技能的 2b-implement-runnable.md 编写，面向希望在 Python LLM 应用中搭建端到端评测管线的开发者。读完本文，你将掌握：如何编写一个pixie.Runnable类作为评测管线的"程序化用户"、如何用 PydanticBaseModel表达真实用户输入、如何保证并发安全并正确接入pixie test与pixie trace，从而在不改动应用代码的前提下对应用的请求处理、上下文组装、路由与响应格式化做真实评测。

1. Runnable 在整个评测工作流中的位置

eval-driven-dev 技能定义了一条完整的六步评测工作流（见 SKILL.md）：理解应用与定义评测标准（Step 1）→ 插桩、运行应用并捕获参考 trace（Step 2）→ 定义评估器（Step 3）→ 构建数据集（Step 4）→ 运行pixie test（Step 5）→ 分析结果（Step 6）。

其中Step 2b 的产出物pixie_qa/run_app.py是整个管线的心脏：

• Step 2a通过wrap()在数据边界注入受控输入、捕获输出（详见 2a-instrumentation.md）；
• Step 2b编写的 Runnable 负责"按真实用户的方式调用应用"，让pixie trace和pixie test能把每个测试用例真正跑起来；
• Step 2c 通过它捕获参考 trace（pixie_qa/reference-trace.jsonl），Step 4 数据集中的"runnable"字段会直接引用它（详见 testing-api.md 的 Dataset JSON 格式）。

Step 2b 的目标（原文档原文）：编写一个 Runnable 类，让评测 harness 能够完全像真实用户那样调用应用。

2. 核心思想：程序化的"真实用户"替身

Runnable 就是pixie test与pixie trace运行应用的方式。把它想象成真实用户的程序化替身：它启动应用、发送请求、然后让应用自己完成后续一切。评测 harness 会对每个测试用例调用run()，并传入用户的输入参数；应用则通过其真实代码路径处理这些参数——真实的路由、真实的 prompt 组装、真实的 LLM 调用、真实的响应格式化——而 harness 通过 Step 2a 加入的wrap()插桩观察过程中发生的一切。

wrap()在不同模式下行为不同（详见 wrap-api.md）：pixie trace时写入 trace 文件并发出 OTel 事件；pixie test的 eval 模式下，purpose="input"会注入数据集中的依赖数据，purpose="output"/purpose="state"则捕获输出与中间状态。Runnable 无需关心这些细节——它只负责把应用的真实入口点接到 harness 接口上。

因此 Runnable 应当保持简单：它只是把应用的真实入口点接到 harness 接口上。如果 Runnable 变得复杂——开始写自定义逻辑、重新实现应用行为、或替换组件——说明哪里出了问题。

3. 四项要求详解

3.1 运行真实的生产代码

Runnable 必须调用应用的真实入口点——真实用户会触发的那个函数、类或端点。不得重新实现、偷工减料或替换应用的任何部分。

这条要求尤其指向 LLM 调用：应用的 LLM 调用必须走真实代码路径，不要 mock、伪造或替换任何应用组件。评测型测试的全部意义在于 LLM 输出是非确定性的，因此要用评估器（而非断言）来打分；如果你用 fake 替换了任何组件，就消除了真实行为，评测将毫无意义。

原文档进一步规定了处理环境问题的边界：

如果应用因缺少环境变量或配置而无法运行，且你无法解决，应停下来请用户修复环境设置。不要通过 mock 组件来绕过。

技能主文档 SKILL.md 将这一原则扩展为明确的"何时停下求助"清单：应用因缺少环境变量/配置无法运行、核心模块因系统依赖或 Python 版本不兼容而导入失败、入口点存在多重歧义时，都应当向用户求助；而缺失 Python 包、pixie包、端口冲突等则应当自己解决。另外注意：项目自身的tests/、fixtures/、mock server 属于开发基础设施，不是评测数据集的数据来源。

3.2 用 Pydantic BaseModel 表达启动参数

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

from pydantic import BaseModel class AppArgs(BaseModel): user_message: str # 应用入口点还需要什么就继续加字段。 # 这些字段与数据集 input_data 的键一一对应。

字段必须反映真实用户实际提供的内容。原文档要求阅读pixie_qa/00-project-analysis.md（Step 1a 的产出，见 1-a-project-analysis.md）中的 "Realistic input characteristics" 一节——它描述了真实输入的复杂度、规模与多样性。模型要按这种真实度设计，而不是简化的玩具版本。

用户参数与世界数据的边界

这是最容易踩坑的设计决策。需要分清两类数据：

• 用户提供的参数（BaseModel 上的字段）：真实用户输入或配置的内容——prompt、查询、配置标志、URL、schema 定义；
• 世界数据（由 Step 2a 的wrap(purpose="input")处理）：应用执行期间从外部来源获取的内容——网页、数据库记录、API 响应。这不属于 BaseModel。

原文档给出的典型映射表：

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

3.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 为键）或本质上是无状态的，并发调用天然隔离，无需加锁。

wrap-api.md 补充了常见的并发陷阱清单：

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

3.4 遵守 Runnable 接口

Runnable 是pixie.Runnable协议的具体实现（完整协议定义见 wrap-api.md 的pixie.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()—— 可选的 async 方法；初始化共享资源（HTTP 客户端、数据库连接、服务器）。有默认空实现；
• run(args)—— async 方法；每个数据集条目调用一次。在这里调用应用的真实入口点；
• teardown()—— 可选的 async 方法；清理setup()中获取的资源。有默认空实现。

协议的生命周期顺序是：create()→setup()（一次）→run()（每个条目一次，asyncio.gather并发）→teardown()（一次）。其中run()接收的args是根据input_data构建、经过校验的 Pydantic 模型。

4. 最小示例：三行核心逻辑

原文档给出的最小可运行示例：

# pixie_qa/run_app.py from pydantic import BaseModel import pixie class AppArgs(BaseModel): user_message: str class AppRunnable(pixie.Runnable[AppArgs]): """Drives the application for tracing and evaluation.""" @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 导入应用的真实入口点并调用它。没有自定义逻辑、没有组件替换、没有聪明的变通方案。注意导入放在run()内部（from myapp import handle_request），这可以避免模块加载时的副作用并延迟依赖解析。

5. 按架构类型选择实现方式

原文档要求：根据应用的运行方式，只阅读与你的应用类型匹配的那一个示例文件。仓库中的三个示例文件（均在 runnable-examples 目录下）：

5.1 独立函数（无服务器）

最简单的情况：直接从run()导入并调用函数。如果函数是同步的，用asyncio.to_thread包装：

import asyncio async def run(self, args: AppArgs) -> None: from myapp.agent import answer_question await asyncio.to_thread(answer_question, args.question)

如果函数依赖外部服务（如向量库），Step 2a 加入的wrap(purpose="input")会自动处理——eval 模式下 registry 注入测试数据。大多数独立函数不需要生命周期方法；仅在函数需要共享资源（如预加载的 embedding 模型、数据库连接）时才覆写setup()/teardown()：

class AppRunnable(pixie.Runnable[AppArgs]): _model: SomeModel @classmethod def create(cls) -> "AppRunnable": return cls() async def setup(self) -> None: from myapp.models import load_model self._model = load_model() async def run(self, args: AppArgs) -> None: from myapp.agent import answer_question await answer_question(args.question, model=self._model)

5.2 FastAPI / Web 服务器

推荐方案：用httpx.AsyncClient+ASGITransport在进程内运行 ASGI 应用。最快、最可靠——无需子进程、无需管理端口：

# pixie_qa/run_app.py import httpx from pydantic import BaseModel import pixie class AppArgs(BaseModel): user_message: str class AppRunnable(pixie.Runnable[AppArgs]): """Drives a FastAPI app via in-process ASGI transport.""" _client: httpx.AsyncClient @classmethod def create(cls) -> "AppRunnable": return cls() async def setup(self) -> None: from myapp.main import app # 你的 FastAPI/Starlette 应用实例 transport = httpx.ASGITransport(app=app) self._client = httpx.AsyncClient(transport=transport, base_url="http://test") 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()

关键陷阱：ASGITransport不会触发 ASGI lifespan 事件（startup/shutdown）。如果应用在 lifespan 中初始化资源（数据库连接、缓存、服务客户端），必须在setup()中手动复刻这些初始化：

async def setup(self) -> None: # 手动复刻应用 lifespan 所做的工作 from myapp.db import get_connection, init_db, seed_data import myapp.main as app_module conn = get_connection() init_db(conn) seed_data(conn) app_module.db_conn = conn # 设置应用期望的模块级全局变量 transport = httpx.ASGITransport(app=app_module.app) self._client = httpx.AsyncClient(transport=transport, base_url="http://test") async def teardown(self) -> None: await self._client.aclose() # 清理手动初始化的资源 import myapp.main as app_module if hasattr(app_module, "db_conn") and app_module.db_conn: app_module.db_conn.close()

备选方案：外部服务器 + httpx。当应用无法直接导入（启动流程复杂、uvicorn.run()在__main__中）时，先以子进程启动服务器再用 HTTP 访问。启动命令示例：bash resources/run-with-timeout.sh 120 uv run python -m myapp.server，随后sleep 3等待就绪。

5.3 CLI 应用

当应用通过命令行调用（如python -m myapp、argparse/click 编写的 CLI 工具）时，用asyncio.create_subprocess_exec调用 CLI 并捕获输出：

# pixie_qa/run_app.py import asyncio import sys from pydantic import BaseModel import pixie class AppArgs(BaseModel): query: str class AppRunnable(pixie.Runnable[AppArgs]): """Drives a CLI application via subprocess.""" @classmethod def create(cls) -> "AppRunnable": return cls() async def run(self, args: AppArgs) -> None: proc = await asyncio.create_subprocess_exec( sys.executable, "-m", "myapp", "--query", args.query, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=120) if proc.returncode != 0: raise RuntimeError(f"App failed (exit {proc.returncode}): {stderr.decode()}")

CLI 需要补丁依赖时：如果 CLI 读取外部服务，可创建一个先补丁依赖再运行真实 CLI 的包装入口点（如pixie_qa/patched_app.py），再让 Runnable 指向该包装。重要限制：对 CLI 应用，wrap(purpose="input")的注入只在应用与 harness 同进程时有效；若走子进程，可能需要通过环境变量或配置文件传递测试数据。

6. 文件放置与数据集引用

• 文件放在pixie_qa/run_app.py；
• 数据集的"runnable"字段引用格式为："pixie_qa/run_app.py:AppRunnable"；
• 项目根目录会自动加入sys.path，所以可以直接使用普通导入（如from app import service）。

runnable字段在数据集 JSON 中是必填项，格式为filepath:ClassName，指向驱动评测期间应用的 Runnable 子类。entries[].input_data是必填的 kwargs，会以 Pydantic 模型形式传给Runnable.run()，其键必须与run(args: T)中 Pydantic 模型的字段一致（详见 testing-api.md 的 Dataset JSON 格式与 Entry structure）。

7. 技术注意事项

原文档强调了一个具体的技术约束：

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

此外，setup()与teardown()自带默认空实现，只有需要共享资源时才覆写；create()中实例化时若需初始化信号量等属性，应像第 3.3 节那样在create()中赋值后再返回实例。

8. 自检清单：一个合格的 Runnable 长什么样

综合原文档与配套文档，写出 Runnable 后可对照以下清单自检：

1. 运行真实代码：run()是否调用了应用的真实入口点？是否存在 mock、fake 或组件替换？（违者直接判废）
2. 参数模型真实：AppArgs的字段是否与数据集input_data一一对应？字段是否代表"用户真实提供"的参数而非应用自行获取的世界数据？
3. 并发安全：应用有共享可变状态时是否用asyncio.Semaphore保护？无状态/按请求隔离的应用是否避免了不必要的加锁？
4. 接口完整：四个生命周期方法签名是否正确？create()是否使用带引号的返回类型？
5. 文件位置与引用：是否位于pixie_qa/run_app.py？数据集"runnable"字段是否写成"pixie_qa/run_app.py:AppRunnable"？
6. 没有未来的注解：是否避免使用from __future__ import annotations？

9. 与后续步骤的衔接

Runnable 写好后，工作流进入Step 2c：捕获并验证参考 trace（见 2c-capture-and-verify-trace.md），产出的pixie_qa/reference-trace.jsonl用于验证插桩与 Runnable 是否正确工作。随后 Step 3 定义评估器（3-define-evaluators.md）、Step 4 基于 trace 的数据形状构建数据集（其中wrap(purpose="input")的世界数据来自eval_input，input_data则直接喂给 Runnable 的参数模型）、Step 5 运行pixie test得到真实评分。

整个链条中，Runnable 的简洁性是贯穿始终的质量信号：如果它开始变复杂，说明你正在绕过评测的本质——让应用在真实输入下、走真实代码路径、产生可被评估器打分的真实输出。

产出物：pixie_qa/run_app.py—— 一个 Runnable 类。

【免费下载链接】awesome-copilotCommunity-contributed instructions, agents, skills, and configurations to help you make the most of GitHub Copilot.项目地址: https://gitcode.com/GitHub_Trending/aw/awesome-copilot