ARTICLE DETAIL

资讯详情

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

为 FastAPI 网页服务编写 pixie Runnable:基于 ASGI 进程内传输的评估驱动开发实战指南

为 FastAPI 网页服务编写 pixie Runnable:基于 ASGI 进程内传输的评估驱动开发实战指南 为 FastAPI 网页服务编写 pixie Runnable基于 ASGI 进程内传输的评估驱动开发实战指南【免费下载链接】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本文是 eval-driven-dev 技能体系中可运行示例Runnable Example专题的一篇实战指南聚焦于被测应用是 FastAPI / Flask / Starlette 等网页服务器的场景。你将学会如何在pixie_qa/run_app.py中编写一个pixie.Runnable通过httpx.AsyncClientASGITransport在进程内驱动完整的 HTTP 请求管线并掌握 lifespan 事件缺失、并发共享状态、外部子进程服务器这三种关键情形的标准解法从而为评估驱动开发eval-driven dev流水线接入任意 Web 形态的 LLM 应用。完整方法论与流水线背景可参考 skills/eval-driven-dev/SKILL.md。背景Runnable 在 eval 流水线中的角色在 eval-driven-dev 的六步工作流中SKILL.md 定义了 Step 1–6Step 2b 的产物就是一个 Runnable 类文件pixie_qa/run_app.py。Runnable 是评估驱动开发流水线pixie与真实应用之间的桥梁The Runnable is howpixie testandpixie tracerun your application. Think of it as a programmatic stand-in for a real user: it starts the app, sends it a request, and lets the app do its thing.它必须满足四条硬性要求详见 references/2b-implement-runnable.md运行真实的生产代码调用应用真正的入口函数、类或 HTTP 端点绝不 mock、stub 或替换任何组件——包括 LLM 调用本身因为 LLM 输出的非确定性正是使用 evaluator 而非assertEqual评分的原因用 PydanticBaseModel表示启动参数run()接收的模型字段与数据集的input_data键一一对应并发安全run()会被并发调用最多 4 个 entry 并行共享可变状态需要保护遵循pixie.Runnable协议接口。Runnable 协议的生命周期根据 references/wrap-api.md 中自动生成的 API 参考Runnable是一个Protocol[T]其生命周期为class pixie.Runnable(Protocol[T]): classmethod def create(cls) - Runnable[Any]: ... # 类方法构造并返回实例 async def setup(self) - None: ... # async仅在第一次 run() 前调用一次 async def run(self, args: T) - None: ... # async每个数据集条目调用一次可并发 async def teardown(self) - None: ... # async最后一次 run() 后调用一次其中T是pydantic.BaseModel子类字段与数据集 JSON 中的input_data键匹配。setup()与teardown()有默认空实现仅在需要初始化/释放共享资源HTTP 客户端、数据库连接、服务器等时才重写。方案一进程内 ASGI 传输推荐当应用是一个 Web 服务器FastAPI、Flask、Starlette需要锻炼完整的 HTTP 请求管线时优先使用httpx.AsyncClientASGITransport在进程内驱动 ASGI 应用references/runnable-examples/fastapi-web-server.md。这是最快、最可靠的方案——没有子进程没有端口管理。# 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 # your FastAPI/Starlette app instance 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()要点拆解create()类方法返回实例返回类型使用带引号的- AppRunnable以避免前向引用错误这是本技能明确规定的写法见 2b-implement-runnable.md 的技术说明。setup()在第一次run()前执行一次这里从应用模块导入 ASGIapp实例构建httpx.ASGITransport与AsyncClient。base_urlhttp://test只是占位——请求不会真的离开进程因此无需真实域名。run()每个数据集条目调用一次向/chat端点发送 POST 请求请求体{message: args.user_message}来自input_data的user_message字段。teardown()所有条目跑完后调用一次关闭异步客户端释放连接资源。这一模式与 wrap-api.md 中给出的 Web 服务器示例使用httpx.AsyncClient(base_urlhttp://localhost:8000)的变体互为补充进程内 ASGI 方案不需要真实端口是 eval 流水线默认推荐的形式。为什么进程内 ASGI 更优速度请求在事件循环内直接分发到 ASGI 应用省去 socket 建立、端口监听、进程调度开销可靠性不存在端口冲突、服务未就绪、子进程崩溃等外部因素wrap 注入生效由于应用与评估进程运行在同一进程Step 2a 添加的wrap(purposeinput)函数式注入可以直接替换外部依赖返回值。这一点与 CLI 子进程方案形成鲜明对比——cli-app.md 明确提示For CLI apps,wrap(purposeinput)injection only works when the app runs in the same process子进程方案下只能改用环境变量或配置文件传测试数据。关键陷阱ASGITransport 跳过 lifespan 事件httpx.ASGITransport不会触发 ASGI lifespan 事件startup/shutdown。如果应用在 lifespan 中初始化资源——数据库连接、缓存、服务客户端——你必须在setup()中手动复现这段初始化逻辑否则应用会以未初始化的状态处理请求async def setup(self) - None: # Manually replicate what the apps lifespan does 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 # set the module-level global the app expects transport httpx.ASGITransport(appapp_module.app) self._client httpx.AsyncClient(transporttransport, base_urlhttp://test) async def teardown(self) - None: await self._client.aclose() # Clean up the manually-initialized resources import myapp.main as app_module if hasattr(app_module, db_conn) and app_module.db_conn: app_module.db_conn.close()两个实践要点逐条对照应用 lifespan 代码把startup里做的每一件事建连接、建表、灌种子数据、注册客户端在setup()中复刻把shutdown里做的清理在teardown()中复刻。通过模块级全局接入示例中app_module.db_conn conn直接把连接写进应用模块期望读取的全局变量这与应用自身 lifespan 的行为保持一致teardown()中使用hasattr防御式清理避免重复关闭或清理不存在的资源。并发与共享可变状态何时加 Semaphorerun()会被asyncio.gather并发调用最多 4 个数据集条目并行见 wrap-api.md 与 5-run-tests.md。如果应用使用共享可变状态——内存型 SQLite、基于文件的数据库、全局缓存——需要用信号量串行化访问import asyncio class AppRunnable(pixie.Runnable[AppArgs]): _client: httpx.AsyncClient _sem: asyncio.Semaphore classmethod def create(cls) - AppRunnable: inst cls() inst._sem asyncio.Semaphore(1) return inst async def setup(self) - None: from myapp.main import app transport httpx.ASGITransport(appapp) self._client httpx.AsyncClient(transporttransport, base_urlhttp://test) async def run(self, args: AppArgs) - None: async with self._sem: await self._client.post(/chat, json{message: args.user_message}) async def teardown(self) - None: await self._client.aclose()只在确有必要时才加信号量——如果应用使用按唯一 ID如call_sid、session_id键控的会话态并发调用天然隔离无需加锁。并发陷阱在 wrap-api.md 中有更系统的归纳SQLite不支持并发写需要Semaphore(1)或使用 WAL 模式的aiosqlite模块级全局可变状态在run()中修改的模块级 dict/list 需要保护速率受限 API加信号量可避免 429 错误。典型故障信号是sqlite3.OperationalError/database is locked——这正是 5-run-tests.md 机械错误排查表中的已知条目修复方式就是在 Runnable 中加asyncio.Semaphore(1)。方案二外部服务器 httpx子进程模式当应用无法被直接导入时启动过程复杂、__main__里调用uvicorn.run()采用子进程方案先把服务器作为独立进程启动再用 HTTP 请求它class AppRunnable(pixie.Runnable[AppArgs]): _client: httpx.AsyncClient classmethod def create(cls) - AppRunnable: return cls() async def setup(self) - None: # Assumes the server is already running (started via run-with-timeout.sh) self._client httpx.AsyncClient(base_urlhttp://localhost:8000) 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()启动服务器并等待就绪必须在运行pixie trace或pixie test之前完成bash resources/run-with-timeout.sh 120 uv run python -m myapp.server sleep 3 # wait for readiness该方案的取舍优点能覆盖进程内方案无法处理的复杂启动流程最接近真实部署形态代价需要端口管理、就绪等待且wrap(purposeinput)注入在子进程模式下不生效不同进程注册表无法注入测试数据需通过环境变量或配置文件传递参见 cli-app.md 中关于子进程方案的同一限制说明。Runnable 的放置、接线与验证文件放置与引用文件固定放在pixie_qa/run_app.py数据集 JSON 的runnable字段引用格式为pixie_qa/run_app.py:AppRunnable见 testing-api.md 的 Dataset JSON Format项目根目录会自动加入sys.path因此 Runnable 内可直接使用普通 importfrom app import service。技术注意不要在 Runnable 文件中使用from __future__ import annotations——它会破坏 Pydantic 对嵌套模型的解析需要用到前向引用时改用带引号的返回类型- AppRunnable。验证流水线先用pixie trace捕获参考 trace确认 Runnable 与插桩工作正常2c-capture-and-verify-trace.mdecho {user_message: 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--input接收的是文件路径不是内联 JSONJSON 的键会成为 Pydantic 模型的 kwargs。随后运行完整评估uv run pixie test # 端到端跑完整流水线 uv run pixie test -v # 冗长输出显示每个用例的分数与 evaluator 推理注意 testing-api.md 与 5-run-tests.md 强调的测试期映射关系应用中的wrap(purposeoutput)与wrap(purposestate)调用都会进入Evaluable.eval_outputlist[NamedData]自定义 evaluator 通过名称查找输出值wrap(purposeinput)则从数据集 entry 的eval_input注册表中取注入值。常见机械错误速查如果pixie test报错先对照 5-run-tests.md 的机械错误排查表不要贸然去改应用代码错误原因修复WrapRegistryMissError: namekey数据集 entry 缺少应用wrap(purposeinput, namekey)期望的eval_input项给每个受影响 entry 补充{name: key, value: ...}WrapTypeMismatchError反序列化类型与应用期望不符修正数据集中的 valueRunnable 解析失败runnable路径或类名错误或类未实现协议修正filepath:ClassName确保类有create()与run()ModuleNotFoundError: pixie_qapixie_qa/缺少__init__.py重新运行pixie initsqlite3.OperationalError并发run()共享 SQLite 连接在 Runnable 中加asyncio.Semaphore(1)三种应用形态的 Runnable 对比本示例文件是 eval-driven-dev 技能中三个 Runnable 示例之一见 2b-implement-runnable.md 的架构示例表格应用类型入口点推荐方案示例文件Web 服务器FastAPI、Flask、StarletteHTTP/WebSocket 端点httpx.AsyncClientASGITransport进程内驱动本文方案一或外部子进程 HTTP方案二fastapi-web-server.md独立函数无服务器Python 函数直接导入并调用同步函数用asyncio.to_thread包装standalone-function.mdCLI 应用命令行调用asyncio.create_subprocess_exec子进程调用并捕获输出cli-app.md小结为 FastAPI 等 Web 服务器编写 Runnable 时优先选择httpx.ASGITransport进程内方案——它速度快、无端口管理、能让wrap(purposeinput)注入在评估进程中直接生效。三个必须牢记的要点lifespan 不会自动触发在setup()中手动复现应用的startup初始化在teardown()中清理并发安全run()最多 4 路并行共享可变状态SQLite、全局缓存必须用asyncio.Semaphore(1)串行化无法直接导入时切子进程方案先用run-with-timeout.sh启动服务器、等待就绪再让 Runnable 通过 HTTP 访问并接受 wrap 注入失效的局限。坚持Runnable 只做接线、绝不重实现应用逻辑的原则你的 Web 应用就能被纳入端到端、基于真实 LLM 调用、由真实数据驱动的评估流水线。更完整的 API 细节可继续查阅 wrap-api.md 与 testing-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),仅供参考
返回列表