AIBrix OpenAI Batch API 端到端测试实战：从环境搭建到完整流程验证

AIBrix OpenAI Batch API 端到端测试实战从环境搭建到完整流程验证【免费下载链接】aibrixCost-efficient and pluggable Infrastructure components for GenAI inference项目地址: https://gitcode.com/GitHub_Trending/ai/aibrix本指南以 AIBrix 仓库中的端到端E2E测试文档为主体讲解如何针对真实运行的 AIBrix 元数据服务Metadata Service验证 OpenAI Batch API 的完整批量推理工作流。文章覆盖测试环境准备服务端口转发、S3/TOS 对象存储凭据生成、pytest 运行方式、测试用例结构与断言逻辑并结合 python/aibrix/aibrix/metadata/api/v1/batch.py 与 python/aibrix/aibrix/metadata/api/v1/files.py 等源码深入剖析测试背后对应的服务端实现。读完本文你将能够独立搭建 E2E 测试环境、运行并解读 Batch API 测试结果并理解上传文件 → 创建批量任务 → 轮询状态 → 下载结果这一 OpenAI 兼容批量推理闭环的服务端状态机。一、E2E 测试是什么AIBrix 的端到端测试位于 python/aibrix/tests/e2e/其特点是针对真实运行的服务实例进行验证而非使用 mock 对象。当前该目录下的测试主体是 python/aibrix/tests/e2e/test_batch_api.pytest_batch_api.py—— OpenAI Batch API 端点的端到端测试该测试文件会向一个真实启动的 AIBrix 元数据服务发送 HTTP 请求依次执行文件上传、批量任务创建、状态轮询、结果下载与批量列表校验从而验证整套批量推理链路在真实存储后端与真实推理引擎下的可用性。这是文档明确说明的测试定位This directory contains end-to-end tests for Aibrix services that run against real service instances.说明原 README 中提及的test_batch_api_service_availability()与test_batch_api_error_handling_real_service()属于文档描述的历史测试形态从当前源码结构看test_batch_api.py 实际包含test_batch_api_e2e_real_servicehttpx 直连版与test_openai_batch_apiOpenAI Python SDK 版两个用例下文均以实际源码为准展开。二、前置条件一个可访问的服务与一套存储凭据2.1 运行中的 AIBrix 服务测试默认连接http://localhost:8888/。如果你在 Kubernetes 集群中部署了元数据服务最简单的方式是使用kubectl port-forward将服务端口暴露到本机kubectl -n envoy-gateway-system port-forward service/envoy-aibrix-system-aibrix-eg-903790dc 8888:80该命令把集群内envoy-gateway-system命名空间下的网关服务映射到本机8888端口。作为参考AIBrix 元数据服务的实际部署清单位于 config/metadata/metadata.yaml其中 Service 暴露端口为8090Deployment 以--host 0.0.0.0 --port 8090启动aibrix_metadata进程并通过--enable-k8s-support开启 Kubernetes 任务执行能力——生产环境通常还需要在网关层做路由与端口转发。2.2 生成对象存储凭据批量推理的输入文件与输出文件都需要持久化到对象存储S3、TOS 或本地存储因此测试前必须确保对象存储可访问、且元数据服务能读取相应凭据。以 S3 为例在仓库的 Python 包根目录执行cd /path/to/aibrix/python/aibrix python -m scripts.generate_secrets s3 --bucket your-bucket-name该命令会读取你通过aws configure配置好的 AWS 凭据访问密钥、区域并在 Kubernetes 集群中创建对应的 Secret。其 CLI 实现在 python/aibrix/scripts/generate_secrets.py支持四个子命令子命令作用关键参数s3创建 S3 凭据 Secret--bucket/-b必填、--name、--namespace/-ntos创建 TOS 凭据 Secret--bucket/-b必填、--name、--namespace/-ndelete删除指定 Secretsecret_name位置参数、--namespacelist列出命名空间内所有 Secret--namespace命令示例摘自 generate_secrets.py 的 help 文本# 使用默认名称创建 S3 Secret python -m scripts.generate_secrets s3 --bucket my-bucket # 自定义 Secret 名称 python -m scripts.generate_secrets s3 --bucket my-bucket --name my-s3-creds # 创建 TOS Secret需要 TOS_* 环境变量 python -m scripts.generate_secrets tos --bucket my-tos-bucket # 删除与列出 python -m scripts.generate_secrets delete my-secret-name python -m scripts.generate_secrets list # --namespace 既可以放在子命令前也可以放在子命令后 python -m scripts.generate_secrets --namespace my-namespace s3 --bucket my-bucket python -m scripts.generate_secrets s3 --bucket my-bucket --namespace my-namespace底层逻辑位于 python/aibrix/aibrix/metadata/secret_gen.pySecretGenerator类通过boto3.Session().get_credentials()获取 AWS 凭据区域缺省为us-east-1将access_key、secret_key、region、bucket-name写入基于模板的 Kubernetes SecretS3 模板 python/aibrix/aibrix/metadata/setting/s3_secret_template.yamltype: Opaque默认名称aibrix-s3-credentialsTOS 模板 python/aibrix/aibrix/metadata/setting/tos_secret_template.yaml需要TOS_ACCESS_KEY、TOS_SECRET_KEY、TOS_ENDPOINT、TOS_REGION四个环境变量。_encode_data()会将全部值做 Base64 编码后写入data字段符合 Kubernetes Secret 的存储规范创建前若同名 Secret 已存在会先删除再重建避免残留冲突。三、运行测试四种 pytest 姿势所有命令都需在python/aibrix目录下执行即pyproject.toml所在目录。测试依赖httpx、pytest、pytest-asyncio、openai等包均已在 python/aibrix/pyproject.toml 中声明如pytest ^8.3.2、pytest-asyncio ^1.1.0、openai ^2.32.0。3.1 运行全部 E2E 测试cd /path/to/aibrix/python/aibrix pytest tests/e2e/ -v3.2 仅运行 Batch API 测试cd /path/to/aibrix/python/aibrix pytest tests/e2e/test_batch_api.py -v3.3 运行单个指定测试cd /path/to/aibrix/python/aibrix pytest tests/e2e/test_batch_api.py::test_batch_api_e2e_real_service -v3.4 输出详细日志cd /path/to/aibrix/python/aibrix pytest tests/e2e/test_batch_api.py -v -s-s会关闭 pytest 对 stdout 的捕获测试中用print()输出的进度信息如Step 1: Uploading batch input file...、Attempt 3: Status in_progress会实时显示便于观察批量任务的状态迁移过程。四、测试结构拆解Fixture、用例与断言4.1service_health会话级健康检查 Fixturetest_batch_api.py 定义了一个scopesession的 fixturepytest.fixture(scopesession) def service_health(): Fixture to check service health and skip tests if service is not available. base_url http://localhost:8888 print(f Checking service health at {base_url}...) is_healthy asyncio.run(check_service_health(base_url)) if not is_healthy: pytest.skip(fService at {base_url} is not available or healthy) print(f✅ Service at {base_url} is healthy) return base_url要点会话级session-scoped整个测试会话只检查一次服务健康状态避免每个用例都重复探测自动跳过若服务不可用直接pytest.skip()全部依赖该 fixture 的用例都会被标记为 SKIPPED 而非 FAILED这正是针对真实服务的测试应有的容错行为健康检查方式check_service_health()通过httpx.AsyncClient(timeout10.0)发起GET {base_url}/v1/batches只要返回200即认为服务健康见 test_batch_api.py。这里借用了列出所有批量任务这个读接口做通用可用性探测——README 的 API Endpoints 一节对此有明确说明/v1/batches用于General service availability check by list all batches。注意README 描述 fixture 时提到Tests/healthzendpoint但实际代码以GET /v1/batches作为健康检查路径请以源码行为为准。4.2 完整工作流测试test_batch_api_e2e_real_service这是核心用例test_batch_api.py用httpx.AsyncClient(timeout60.0)直连服务完整走一遍 OpenAI Batch API 的五个步骤上传输入文件Files API构造包含 3 条请求的 JSONL 数据以multipart/form-data形式POST /v1/filespurpose固定为batch。断言返回的object file、purpose batch、status uploaded并取出id作为input_file_id创建批量任务Batch APIPOST /v1/batches请求体为{input_file_id: ..., endpoint: /v1/chat/completions, completion_window: 24h}断言返回的object batch且input_file_id、endpoint回显一致取得batch_id轮询任务状态每 5 秒GET /v1/batches/{batch_id}一次最多轮询 60 次即最长等待 300 秒。根据状态分流处理completed→ 取出output_file_id校验request_counts为total 3、completed 3、failed 0failed→ 读取errors字段并pytest.failcancelled/expired→ 直接判失败scheduling、validating、in_progress、finalizing→ 预期的中间状态继续等待未知状态 → 打印告警继续轮询下载并校验输出Files APIGET /v1/files/{output_file_id}/content将响应内容按 UTF-8 解码后交给verify_batch_output_content()校验验证批量列表接口GET /v1/batches断言object list、data非空且能在列表中找回本测试创建的batch_id且状态为completed。4.3 输出内容校验verify_batch_output_content该函数test_batch_api.py逐行解析输出 JSONL验证 OpenAI Batch 输出格式输出行数必须等于预期的请求数此处为 3每条输出必须包含顶层字段id、custom_id、responsecustom_id必须按序匹配request-1、request-2、request-3response内必须包含status_code、request_id、body且status_code 200body内必须包含model、choices。这一断言与 OpenAI Batch API 的官方输出契约保持一致批量输出文件中的每一行都携带原始请求的custom_id便于用户将结果与输入一一对应。4.4 OpenAI SDK 版本test_openai_batch_apitest_batch_api.py 提供了使用官方openaiPython SDK 的等价实现逻辑与 httpx 版完全一致但调用方式更贴近真实用户with OpenAI(base_urlf{base_url}/v1, api_keyaibrix) as client: upload_result client.files.create(filef, purposebatch) batch_result client.batches.create( input_file_idinput_file_id, endpoint/v1/chat/completions, completion_window24h, ) status_result client.batches.retrieve(batch_idbatch_id) output client.files.content(output_file_id) list_result client.batches.list()注意 SDK 客户端的base_url指向http://localhost:8888/v1api_key任意填写此处为aibrix。该用例验证了 AIBrix 服务与 OpenAI SDK 的兼容性——用户可以直接用官方 SDK 驱动 AIBrix 的批量推理。4.5 测试输入数据的生成generate_batch_input_data()test_batch_api.py用于构造 JSONL 批量请求其中ENDPOINT_SAMPLE_BODIES定义了四类端点的样例请求体端点样例模型关键字段/v1/chat/completionsgpt-3.5-turbo-0125messages、max_tokens: 1000/v1/completionsgpt-3.5-turbo-0125prompt、max_tokens: 100/v1/embeddingstext-embedding-ada-002input/v1/rerankreranker-v1query、documents生成的数据格式为 OpenAI Batch 标准的 JSONL{custom_id: request-1, method: POST, url: /v1/chat/completions, body: {model: gpt-3.5-turbo-0125, messages: [{role: system, content: You are a helpful assistant.}, {role: user, content: Hello world!}], max_tokens: 1000}}每条记录包含custom_id、method、url、body四个字段——这正是批量任务中每条独立请求的完整描述。五、服务端实现测试背后的 Batch API 与 Files APIE2E 测试并非孤立存在它与元数据服务的实现一一对应。以下是从源码角度对测试覆盖面的印证。5.1 Files API/v1/files系列端点实现在 python/aibrix/aibrix/metadata/api/v1/files.pyPOST /v1/filescreate_file校验文件扩展名仅支持json、jsonl通过Reader包裹上传内容并受settings.MAX_FILE_SIZE大小限制生成 UUID 作为file_id调用request.app.state.storage.put_object()写入对象存储同时记录filename、purpose、created_at元数据。超限返回 413content_size_limit_exceededGET /v1/files/{file_id}/contentretrieve_file_content从存储读取原始内容并以附件形式返回未找到返回 404GET /v1/files/{file_id}retrieve_file_metadata仅读取元数据大小、类型、创建时间等不下载内容HEAD /v1/files/{file_id}以 HTTP 头形式返回元数据X-File-ID、X-File-Name、ETag等DELETE /v1/files/{file_id}删除文件先head_object探测不存在则返回 404GET /v1/fileslist_files支持purpose过滤、limit1–100默认 20与基于file_id的游标分页。注意files.py源码注释明确说明The implementation is for e2e test only for now, and can upload batch input file only—— 即当前 Files API 以支撑批量测试为主要目标。5.2 Batch API/v1/batches系列端点实现在 python/aibrix/aibrix/metadata/api/v1/batch.pyPOST /v1/batchescreate_batch接收 OpenAI 形状的BatchSpecinput_file_id、endpoint、completion_window等通过BatchSpec.newBatchJobSpec()转换为内部BatchJobSpec交给request.app.state.batch_driver.create_job()创建任务并返回 OpenAI 格式的BatchResponseGET /v1/batches/{batch_id}get_batch查询任务详情未找到返回 404POST /v1/batches/{batch_id}/cancelcancel_batch取消任务已处于终态时返回 409GET /v1/batcheslist_batches支持after游标与limit1–100默认 20分页返回object list的结构与测试断言一致。5.3 任务状态机测试轮询逻辑的底层依据测试中轮询所依赖的状态串scheduling → validating → in_progress → finalizing → completed/failed/...在服务端有明确映射见_batch_job_to_openai_response()batch.py中的注释说明内部CREATED状态对外呈现为scheduling任务已接受、等待资源准入VALIDATING到IN_PROGRESS几乎是瞬时的在admit()内完成FINALIZED是终态伞形状态实际结果由status.condition决定completed/failed/expired/cancelledoutput_file_id仅在任务完成且有成功结果rc.completed 0时才对外暴露error_file_id仅在存在失败请求时暴露——这正是 OpenAI 的契约行为测试在completed分支中对output_file_id与request_counts的断言与此一一对应。5.4 批处理系统架构从 python/aibrix/aibrix/batch/README.md 的架构说明可以了解测试背后的完整链路BatchDriver任务生命周期的主编排器JobManager任务状态管理与追踪JobDriver连接任务管理与推理执行如 Kubernetes Job 运行时BatchWorker以 sidecar 模式运行在 Kubernetes Job 中的 worker 脚本等待同 Pod 内 vLLM 引擎在localhost:8000就绪后执行任务直到FINALIZING状态后以退出码 0 结束。元数据服务需要相应 RBAC 权限才能创建/轮询/删除 Kubernetes Job见 config/metadata/metadata.yaml 中aibrix-metadata-service-readerClusterRole 对batch/jobs资源的get/list/create/patch/delete授权。六、配置说明测试的服务地址是硬编码在测试函数中的service_healthfixture 中base_url http://localhost:8888这是 README 明确指出的当前行为The service URL is hardcoded in the test functions。如果你希望指向其他地址或端口需要修改 test_batch_api.py 中的base_url后重新运行。七、预期输出解读7.1 服务可用时的成功输出tests/e2e/test_batch_api.py::test_batch_api_service_availability PASSED tests/e2e/test_batch_api.py::test_batch_api_e2e_real_service PASSED tests/e2e/test_batch_api.py::test_batch_api_error_handling_real_service PASSED以上是 README 记录的历史用例输出形态当前源码对应的实际输出为test_batch_api_e2e_real_service与test_batch_api_openai_batch等用例的 PASSED 结果。使用-v -s运行时你还会看到诸如Step 1: Uploading batch input file...、✅ File uploaded successfully with ID: ...、Attempt N: Status ...、 E2E test completed successfully!等逐步日志。7.2 服务不可用时的跳过输出当localhost:8888上没有可用服务时service_healthfixture 会跳过全部依赖它的用例tests/e2e/test_batch_api.py::test_batch_api_service_availability SKIPPED tests/e2e/test_batch_api.py::test_batch_api_e2e_real_service SKIPPED tests/e2e/test_batch_api.py::test_batch_api_error_handling_real_service SKIPPED这种服务不可用即跳过而非失败的设计保证了 CI 或本机环境未启动服务时测试套件仍能快速、无害地退出也提醒使用者E2E 测试的结果有效性完全依赖于前置条件是否就绪。八、排障建议结合源码可以给出几个常见的排查方向服务不通确认kubectl port-forward仍在运行且目标 Service 名称与命名空间正确可用curl http://localhost:8888/v1/batches手工验证存储凭据缺失运行python -m scripts.generate_secrets list检查 Secret 是否已创建S3 场景先确认aws configure已配置有效凭据批量任务长时间停留在scheduling从源码注释看scheduling表示任务在等待资源准入应检查集群资源与 Job 运行时是否正常参考 batch.py 的状态说明任务failedGET /v1/batches/{batch_id}返回的errors字段会给出具体错误可结合 python/aibrix/aibrix/batch/README.md 中的调试章节查看 worker 与 vLLM 的 Pod 日志。通过本文介绍的环境准备、运行方式与源码对应关系你可以快速上手并深入理解 AIBrix 的 OpenAI Batch API 端到端测试进而将其复用到自己的部署验证流程中。【免费下载链接】aibrixCost-efficient and pluggable Infrastructure components for GenAI inference项目地址: https://gitcode.com/GitHub_Trending/ai/aibrix创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考