ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

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

eval-driven-dev 技能指南:用 pixie.Runnable 驱动真实应用完成 LLM 评测(Step 2b 实战解析) 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→ 插桩、运行应用并捕获参考 traceStep 2→ 定义评估器Step 3→ 构建数据集Step 4→ 运行pixie testStep 5→ 分析结果Step 6。其中Step 2b 的产出物pixie_qa/run_app.py是整个管线的心脏Step 2a通过wrap()在数据边界注入受控输入、捕获输出详见 2a-instrumentation.mdStep 2b编写的 Runnable 负责按真实用户的方式调用应用让pixie trace和pixie test能把每个测试用例真正跑起来Step 2c 通过它捕获参考 tracepixie_qa/reference-trace.jsonlStep 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.mdpixie trace时写入 trace 文件并发出 OTel 事件pixie test的 eval 模式下purposeinput会注入数据集中的依赖数据purposeoutput/purposestate则捕获输出与中间状态。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.mdStep 1a 的产出见 1-a-project-analysis.md中的 Realistic input characteristics 一节——它描述了真实输入的复杂度、规模与多样性。模型要按这种真实度设计而不是简化的玩具版本。用户参数与世界数据的边界这是最容易踩坑的设计决策。需要分清两类数据用户提供的参数BaseModel 上的字段真实用户输入或配置的内容——prompt、查询、配置标志、URL、schema 定义世界数据由 Step 2a 的wrap(purposeinput)处理应用执行期间从外部来源获取的内容——网页、数据库记录、API 响应。这不属于 BaseModel。原文档给出的典型映射表应用类型BaseModel 字段用户提供世界数据wrap 提供Web 爬虫URL prompt schema 定义HTML 页面内容研究型 Agent研究问题 范围约束源文档、搜索结果客服机器人客户的语音消息CRM 中的客户档案、会话存储中的历史记录代码审查工具PR URL 审查标准实际 diff、文件内容、CI 结果判断准则如果某个字段最终存放的是应用本来会自行获取的数据那么它很可能应该放在wrap(purposeinput)调用中而不是 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 目录下应用类型入口点示例文件独立函数无服务器Python 函数standalone-function.mdWeb 服务器FastAPI、FlaskHTTP/WebSocket 端点fastapi-web-server.mdCLI 应用命令行调用cli-app.md5.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(purposeinput)会自动处理——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, modelself._model)5.2 FastAPI / Web 服务器推荐方案用httpx.AsyncClientASGITransport在进程内运行 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(appapp) self._client httpx.AsyncClient(transporttransport, base_urlhttp://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(appapp_module.app) self._client httpx.AsyncClient(transporttransport, base_urlhttp://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, stdoutasyncio.subprocess.PIPE, stderrasyncio.subprocess.PIPE, ) stdout, stderr await asyncio.wait_for(proc.communicate(), timeout120) if proc.returncode ! 0: raise RuntimeError(fApp failed (exit {proc.returncode}): {stderr.decode()})CLI 需要补丁依赖时如果 CLI 读取外部服务可创建一个先补丁依赖再运行真实 CLI 的包装入口点如pixie_qa/patched_app.py再让 Runnable 指向该包装。重要限制对 CLI 应用wrap(purposeinput)的注入只在应用与 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 后可对照以下清单自检运行真实代码run()是否调用了应用的真实入口点是否存在 mock、fake 或组件替换违者直接判废参数模型真实AppArgs的字段是否与数据集input_data一一对应字段是否代表用户真实提供的参数而非应用自行获取的世界数据并发安全应用有共享可变状态时是否用asyncio.Semaphore保护无状态/按请求隔离的应用是否避免了不必要的加锁接口完整四个生命周期方法签名是否正确create()是否使用带引号的返回类型文件位置与引用是否位于pixie_qa/run_app.py数据集runnable字段是否写成pixie_qa/run_app.py:AppRunnable没有未来的注解是否避免使用from __future__ import annotations9. 与后续步骤的衔接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(purposeinput)的世界数据来自eval_inputinput_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创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表