eval-driven-dev 实战指南:为无服务器 Python 函数构建 pixie 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
导读
本文聚焦 awesome-copilot 仓库中 eval-driven-dev 技能(基于 pixie-qa 的评估驱动开发工作流)的一个核心环节:当被测应用是一个纯 Python 函数或模块——没有 Web 框架、没有服务器、没有任何基础设施时,如何编写一个最简的Runnable类,让评测工具链以"真实用户"的方式驱动应用并采集评估数据。读完本文,你将掌握无服务器场景下 Runnable 的四种写法(直接调用、同步函数包装、外部服务注入、共享资源生命周期管理)、setup()/teardown()的适用边界,以及如何将 Runnable 接入pixie trace/pixie test完成可验证的端到端评测。
背景:Runnable 在 eval-driven-dev 工作流中的位置
Runnable是 pixie-qa 评测框架中连接"被测应用"与"评测工具链"的桥梁。在 SKILL.md 描述的六步工作流中,Step 2b 负责实现 Runnable(产出pixie_qa/run_app.py),Step 2c 用它采集参考 trace,Step 5 的pixie test则对每条数据集条目并发调用run()。
Runnable 的定位在 2b-implement-runnable.md 中表述得非常清晰:它只是把应用的真实入口接到评测工具链的接口上。评测工具链通过run()为每个测试用例传入用户输入参数,应用用自己的真实代码处理这些参数——真实路由、真实 prompt 组装、真实 LLM 调用、真实响应格式化——评测工具链则通过 Step 2a 加入的wrap()插桩观察整个过程。
按照应用形态,Step 2b 把 Runnable 划分为三种架构模板,并分别提供了可运行示例:
| 应用形态 | 入口 | 示例文件 |
|---|---|---|
| 独立函数(无服务器) | Python 函数 | standalone-function.md |
| Web 服务器(FastAPI、Flask) | HTTP/WebSocket 端点 | fastapi-web-server.md |
| CLI 应用 | 命令行调用 | cli-app.md |
本文讨论的是第一种形态,也是文档明确标注的"最简单的情况"。
Runnable 接口协议与生命周期
在深入示例前,先建立 Runnable 的接口心智模型。wrap-api.md 给出了完整的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: ...其中泛型T必须是pydantic.BaseModel的子类,其字段与数据集 JSON 中input_data的键一一对应。生命周期由评测工具链驱动:
create()— 类方法,构造并返回一个 Runnable 实例;setup()— 异步,在第一次run()之前仅调用一次,用于初始化共享资源(HTTP 客户端、数据库连接等),默认是 no-op,可选;run(args)— 异步,对每条数据集条目并发调用(最多 4 条并行),args是由input_data校验构建出的 Pydantic 模型,在这里调用应用的真实入口;teardown()— 异步,在最后一次run()之后仅调用一次,释放setup()中获取的资源,默认是 no-op,可选。
两个实现层面的硬性要求:项目根目录(执行pixie test/pixie trace的目录)会自动加入sys.path,因此 Runnable 里可以直接用from myapp import service这类常规导入;同时不要在 Runnable 文件中使用from __future__ import annotations,它会破坏 Pydantic 对嵌套模型的解析,需要时改用带引号的返回类型注解(如-> "AppRunnable")。
场景一:直接调用异步函数(最简形态)
当被测应用就是一个普通函数时,Runnable 的run()直接导入并调用即可,这构成了 standalone-function.md 的核心示例:
# pixie_qa/run_app.py from pydantic import BaseModel import pixie class AppArgs(BaseModel): question: str class AppRunnable(pixie.Runnable[AppArgs]): """Drives a standalone function for tracing and evaluation.""" @classmethod def create(cls) -> "AppRunnable": return cls() async def run(self, args: AppArgs) -> None: from myapp.agent import answer_question await answer_question(args.question)这段代码完整展示了 Runnable 的全部组成要素:
AppArgs:继承自pydantic.BaseModel,字段question代表真实用户提供的输入。run()收到的args就是由数据集条目中input_data的键值校验构建出的实例,因此字段必须与input_data的键一一对应。create():类方法,返回cls()实例,用带引号的返回类型避免前向引用错误。run():内部延迟导入(from myapp.agent import answer_question)应用真实入口并调用。延迟导入是刻意为之——避免模块加载期的循环依赖或副作用,让 Runnable 保持纯净。
按照 2b-implement-runnable.md 的约束,这里必须调用应用的真实生产代码,包括真实的 LLM 调用:评测的出发点就是 LLM 输出具有非确定性,所以用评估器(而非assertEqual断言)来打分;一旦用 mock/fake 替换任何组件,评测就变成了同义反复——你同时控制了输入和输出,分数失去意义。
场景二:同步函数如何接入(asyncio.to_thread)
多数业务函数是同步的,而run()是async。文档给出了明确解法——用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)asyncio.to_thread会在线程池中运行answer_question(args.question),await后run()继续执行。这一写法与场景一在结构上完全等价,只是多了一层线程桥接。
场景三:外部服务依赖交给 wrap 自动注入
无服务器函数最常见的复杂点是依赖外部服务——例如一个回答问题的 agent 需要从向量数据库中检索上下文。此时不需要在 Runnable 里做任何特殊处理,文档明确指出:Step 2a 中添加的wrap(purpose="input")调用会自动处理——在 eval 模式下,注册表会把测试数据注入进去。
理解这句话需要回到wrap()的行为模型。wrap-api.md 定义了wrap()的三种模式:
- No-op 模式(未启用 tracing、无 eval 注册表):原样返回
data; - Tracing 模式(
pixie trace期间):写入 trace 文件并发出 OTel 事件,原样返回data(若data为 callable 则包装之,使其在被调用时触发事件); - Eval 模式(eval 注册表激活):对
purpose="input"注入依赖数据,对purpose="output"/purpose="state"捕获输出与状态。
而 2a-instrumentation.md 补充了关键的操作规则:对purpose="input"的外部调用必须使用函数形式——pixie.wrap(db.get_profile, purpose="input", name="customer_profile")(user_id)。函数形式在 eval 模式下不会真正执行外部调用,而是直接返回注册表中的测试值;值形式(pixie.wrap(db.get_profile(user_id), ...))仍会先执行真实调用再替换结果,浪费时间、制造不稳定测试,并使评测依赖外部服务可用性。
因此场景三中向量检索的接入方式应写成:
# 应用代码中(Step 2a 插桩) retrieved = pixie.wrap(vector_store.search, purpose="input", name="retrieved_context", description="Vector store retrieval results")(query)eval 模式下vector_store.search不会真的被调用,注册表把数据集条目eval_input里{"name": "retrieved_context", "value": ...}的值反序列化后直接返回。Runnable 侧无需感知这一切——它只管调用应用入口,注入由插桩层透明完成。
场景四:共享资源与 setup()/teardown() 的正确用法
文档强调:大多数独立函数不需要生命周期方法。只有当函数需要共享资源(例如预加载的 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)这里_model在setup()中一次性加载并保存在实例属性上,run()每次调用直接复用。注意setup()的定位——它是共享资源的一次性初始化,而不是每个用例的准备工作;teardown()则在所有run()结束后统一释放(例如await self._client.aclose())。
与并发约束的联动
run()最多 4 条并发执行(由asyncio.gather驱动),wrap-api.md 特别列出了常见并发陷阱:
- SQLite:并发写不安全——用
asyncio.Semaphore(1)串行化,或改用 WAL 模式的aiosqlite; - 全局可变状态:
run()中修改模块级 dict/list 需要保护; - 限流 API:加信号量避免 429 错误。
如果应用按唯一 ID(如call_sid、session_id)隔离每请求状态,或本质无状态,则并发调用天然隔离,无需加锁。信号量只在真正存在共享可变状态时才需要。
接入验证:pixie trace 与 pixie test
Runnable 写好后,如何证明它能正确驱动应用?Step 2c 与 Step 5 提供了完整的验证闭环。
采集参考 trace
首先确认应用可导入(python -c "from <module> import <class>"),然后把输入写成 JSON 文件(--input接收的是文件路径,不是内联 JSON,键会成为 Pydantic 模型的 kwargs):
echo '{"question": "a realistic sample input"}' > pixie_qa/sample-input.json uv run pixie trace --runnable pixie_qa/run_app.py:AppRunnable \ --input pixie_qa/sample-input.json \ --output pixie_qa/reference-trace.jsonl生成的 JSONL trace 每行是一条wrap()事件或一个 LLM span:
{"type": "kwargs", "value": {"question": "What are your hours?"}} {"type": "wrap", "name": "customer_profile", "purpose": "input", "data": {...}, ...} {"type": "llm_span", "request_model": "gpt-4o", "input_messages": [...], ...} {"type": "wrap", "name": "response", "purpose": "output", "data": "Our hours are...", ...}验证要点:预期的wrap条目全部出现(代码中每个wrap()调用对应一条)、至少出现一条llm_span(证明真实 LLM 调用发生了)。若 LLM span 缺失,说明 Runnable 配置有误或 LLM 被 mock 掉了——必须先修复再继续。对于无服务器函数,务必采集至少两条输入特征不同的 trace(简单 vs 复杂、不同能力、不同边界条件),防止数据集同质化。
运行评测
uv run pixie test -v # -v 输出每条用例的分数与评估推理pixie test的执行流程(见 5-run-tests.md)与 Runnable 生命周期严格对应:解析数据集runnable字段 →create()构造实例、setup()调用一次 → 条目并发运行(最多 4 条)→teardown()调用一次。运行期间,应用内的wrap(purpose="input")返回注册表值(不再调用外部服务),wrap(purpose="output"/"state")捕获数据供评估器打分。
无服务器场景下最常见的机械性报错及修复(与 Runnable 直接相关):
| 错误 | 原因 | 修复 |
|---|---|---|
WrapRegistryMissError: name='<key>' | 数据集条目缺少应用wrap(purpose="input", name="<key>")期望的eval_input项 | 在每个受影响条目补上{"name": "<key>", "value": ...} |
WrapTypeMismatchError | 反序列化后的注册表值与应用期望类型不符 | 修正数据集中的 value |
| Runnable 解析失败 | runnable路径或类名错误,或类未实现 Runnable 协议 | 修正数据集中的filepath:ClassName,确保类具备create()与run() |
ModuleNotFoundError: pixie_qa | pixie_qa/缺__init__.py | 运行pixie init重建 |
sqlite3.OperationalError | 并发run()共享 SQLite 连接 | 在 Runnable 中加asyncio.Semaphore(1) |
文件放置与数据集引用约定
- Runnable 统一放置在
pixie_qa/run_app.py; - 数据集的
"runnable"字段引用格式为"pixie_qa/run_app.py:AppRunnable"; - 项目根目录自动在
sys.path上,因此 Runnable 与数据集文件都使用常规导入即可。
与其他应用形态的边界对照
为了精确理解"无服务器函数"场景的边界,cli-app.md 中有一个值得注意的提示:CLI 应用经子进程运行时,wrap(purpose="input")注入只在应用与评测同进程时生效——子进程场景可能需要改用环境变量或配置文件传递测试数据。而独立函数形态是进程内直接调用,天然没有这个问题,这也是它被称为"最简单情况"的原因之一。
小结
无服务器函数的 Runnable 是 eval-driven-dev 工作流中最薄的一层适配:AppArgs定义用户输入边界,run()直接调用应用真实入口,同步函数用asyncio.to_thread桥接,外部依赖交给 Step 2a 的wrap(purpose="input")自动注入,共享资源才引入setup()/teardown()。如果你发现 Runnable 越写越复杂——开始内置自定义逻辑、重实现应用行为或替换组件——说明哪里出了问题,回到 2b-implement-runnable.md 重新校准。完整的三形态示例与接口定义分别位于 runnable-examples 目录和 wrap-api.md,可继续深入查阅。
【免费下载链接】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
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考