news 2026/9/13 2:41:35

eval-driven-dev 实战指南:为无服务器 Python 函数构建 pixie Runnable 评测接入

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
eval-driven-dev 实战指南:为无服务器 Python 函数构建 pixie Runnable 评测接入

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的键一一对应。生命周期由评测工具链驱动:

  1. create()— 类方法,构造并返回一个 Runnable 实例;
  2. setup()— 异步,在第一次run()之前仅调用一次,用于初始化共享资源(HTTP 客户端、数据库连接等),默认是 no-op,可选;
  3. run(args)— 异步,对每条数据集条目并发调用(最多 4 条并行),args是由input_data校验构建出的 Pydantic 模型,在这里调用应用的真实入口;
  4. 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)awaitrun()继续执行。这一写法与场景一在结构上完全等价,只是多了一层线程桥接。

场景三:外部服务依赖交给 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)

这里_modelsetup()中一次性加载并保存在实例属性上,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_sidsession_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_qapixie_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),仅供参考

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/9/13 2:39:36

30天259个PR:用Claude Code打造高效AI编程工作流的13个技巧

30天259个PR&#xff0c;这个数字一开始我是怀疑的。平均一天8.6个PR&#xff0c;相当于每个工作日要合入差不多9次代码变更&#xff0c;而且每一笔变更都得经得起审查、测试、合并这一整套流程。如果换成一个喜欢憋大招的开发者&#xff0c;可能一个月下来就一两个巨型PR&…

作者头像 李华
网站建设 2026/9/13 2:38:28

PythonRobotics 如何用 C-GMRES 求解非线性模型预测控制做路径跟踪

PythonRobotics 如何用 C-GMRES 求解非线性模型预测控制做路径跟踪 【免费下载链接】PythonRobotics Python sample codes and textbook for robotics algorithms. 项目地址: https://gitcode.com/GitHub_Trending/py/PythonRobotics 在 PythonRobotics 的 PathTracking…

作者头像 李华
网站建设 2026/9/13 2:37:32

User Information

User Information 【免费下载链接】mastra Mastra is the modern TypeScript framework for AI-powered applications and agents. 项目地址: https://gitcode.com/GitHub_Trending/ma/mastra First Name:Last Name:Location:Occupation:Interests:Goals:Events:Facts:P…

作者头像 李华
网站建设 2026/9/13 2:36:24

CANN Runtime日志分级过滤机制与排障实践:从源码到落盘

CANN Runtime日志系统集成&#xff1a;日志分级过滤输出的实现与源码拆解先说一个我自己调试NPU任务时的典型场景。你写了一个基于CANN的推理程序&#xff0c;在Atlas训练卡上跑起来&#xff0c;结果第一条aclrtLaunch就返回了错误码。这时候大多数人会先怀疑算法写错了&#x…

作者头像 李华