feat(批处理支持): 添加 Batch API 支持，包括文件上传、任务创建、状态查询和结果下载 - #96
Closed
devhunk wants to merge 4 commits into
Closed
Conversation
Reviewer's GuideAdds end-to-end Batch API support for OpenAI-based providers, including protocol extensions, OpenAI adapter implementations, batch lifecycle management service, and a JSON-file-based store for batch job persistence and result handling. Sequence diagram for BatchService submitBatch and polling lifecyclesequenceDiagram
actor User
participant BatchService
participant ChatService
participant APIAdapter
participant BatchJobStore
User->>BatchService: submitBatch(messages, model, sessionID)
BatchService->>ChatService: adapters[model.provider.apiFormat]
BatchService->>APIAdapter: buildChatRequest(for, commonPayload, messages, tools, audioAttachments, imageAttachments, fileAttachments)
BatchService->>APIAdapter: buildBatchFileUploadRequest(for, jsonlData, purpose)
BatchService->>ChatService: fetchData(for: uploadReq, provider)
ChatService-->>BatchService: uploadData
BatchService->>APIAdapter: parseBatchFileUploadResponse(data)
BatchService->>APIAdapter: buildBatchCreateRequest(for, fileId, endpoint, metadata)
BatchService->>ChatService: fetchData(for: createReq, provider)
ChatService-->>BatchService: createData
BatchService->>APIAdapter: parseBatchCreateResponse(data)
BatchService->>BatchJobStore: saveJob(newJob)
BatchService->>BatchService: startPolling(for: newJob)
loop periodic polling
BatchService->>BatchJobStore: getJob(id)
BatchService->>ChatService: providers
BatchService->>APIAdapter: buildBatchStatusRequest(for, batchId)
BatchService->>ChatService: fetchData(for: statusReq, provider)
ChatService-->>BatchService: statusData
BatchService->>APIAdapter: parseBatchStatusResponse(data)
BatchService->>BatchJobStore: saveJob(updatedJob)
alt job completed
BatchService->>APIAdapter: buildBatchResultDownloadRequest(for, fileId)
BatchService->>ChatService: fetchData(for: resultReq, provider)
ChatService-->>BatchService: resultData
BatchService->>APIAdapter: parseBatchResultDownloadResponse(data)
BatchService->>APIAdapter: parseResponse(data)
end
end
File-Level Changes
Possibly linked issues
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 2 issues, and left some high level feedback:
- In
BatchService.submitBatch, failing to build the request body for a specificChatMessagesimply skips that message; consider surfacing this as an error or at least logging which messages were dropped to avoid silent data loss. - The
OpenAIAdapter.parseBatchStatushelper returnsBatchJobobjects with placeholderproviderIDandmodelIDthat are corrected later by callers; you may want to pass these in as parameters or avoid embedding them at all to reduce the risk of accidentally persisting placeholder values.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `BatchService.submitBatch`, failing to build the request body for a specific `ChatMessage` simply skips that message; consider surfacing this as an error or at least logging which messages were dropped to avoid silent data loss.
- The `OpenAIAdapter.parseBatchStatus` helper returns `BatchJob` objects with placeholder `providerID` and `modelID` that are corrected later by callers; you may want to pass these in as parameters or avoid embedding them at all to reduce the risk of accidentally persisting placeholder values.
## Individual Comments
### Comment 1
<location path="ETOSCore/ETOSCore/Batch/BatchService.swift" line_range="43-52" />
<code_context>
+ for (index, msg) in messages.enumerated() {
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Silently skipping messages that fail request-body construction can lead to partial, surprising batch submissions.
When `buildChatRequest` returns `nil` or the body can't be converted to `JSONValue`, the loop just `continue`s, producing fewer `BatchRequestItem`s than input messages and giving no indication of which ones were dropped. Prefer failing fast (throwing) when any message cannot be transformed, or at minimum log skipped messages to make the behavior observable and predictable.
Suggested implementation:
```
// 1. 构建 BatchRequestItems
var batchItems: [BatchRequestItem] = []
for (index, msg) in messages.enumerated() {
let customId = "req-\(sessionID.uuidString)-\(msg.id.uuidString)-\(index)"
// 构造请求体：由于 APIAdapter 没有暴露暴露纯 JSON 构造，
// 我们可以利用 buildChatRequest 并截获其 httpBody
let request = adapter.buildChatRequest(
for: model,
commonPayload: [:],
messages: [msg],
tools: nil,
audioAttachments: [:],
```
```
guard let httpBody = request.httpBody else {
throw NSError(
domain: "BatchService",
code: -2,
userInfo: [
NSLocalizedDescriptionKey: "无法为消息 \(msg.id) 构造请求体，Batch 提交已中止。"
]
)
}
```
```
guard
let jsonObject = try? JSONSerialization.jsonObject(with: httpBody),
let jsonBody = JSONValue(jsonObject)
else {
throw NSError(
domain: "BatchService",
code: -3,
userInfo: [
NSLocalizedDescriptionKey: "无法将消息 \(msg.id) 的请求体转换为 JSON，Batch 提交已中止。"
]
)
}
```
1. If the file defines a custom error type for batch operations (e.g. `BatchError`), prefer throwing that instead of `NSError`, preserving the fail-fast semantics but aligning with existing error handling conventions.
2. Consider adding structured logging where your logging infrastructure exists (e.g. before throwing, log the `sessionID`, `index`, and `msg.id`) so operators can correlate failures with specific messages.
3. If callers rely on partial success, you may want to wrap this function or introduce an alternative API that returns detailed per-message failure information instead of throwing on the first error.
</issue_to_address>
### Comment 2
<location path="ETOSCore/ETOSCore/Batch/BatchJobStore.swift" line_range="21-23" />
<code_context>
+
+ private var fileURL: URL {
+ let docsDir = Persistence.documentsDirectory
+ let batchDir = docsDir.appendingPathComponent("BatchJobs")
+ if !FileManager.default.fileExists(atPath: batchDir.path) {
+ try? FileManager.default.createDirectory(at: batchDir, withIntermediateDirectories: true)
+ }
+ return batchDir.appendingPathComponent("jobs.json")
</code_context>
<issue_to_address>
**issue (bug_risk):** Directory creation failures are silently ignored, which can cause later persistence operations to fail in non-obvious ways.
If `createDirectory` fails (e.g., permissions, disk issues, invalid path), the error is suppressed and later reads/writes to `jobs.json` will likely fail or misbehave. Please handle this failure explicitly (e.g., log via `logger` or propagate the error) so storage issues are visible and diagnosable rather than silently ignored.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+43
to
+52
| for (index, msg) in messages.enumerated() { | ||
| let customId = "req-\(sessionID.uuidString)-\(msg.id.uuidString)-\(index)" | ||
|
|
||
| // 构造请求体：由于 APIAdapter 没有暴露暴露纯 JSON 构造， | ||
| // 我们可以利用 buildChatRequest 并截获其 httpBody | ||
| let request = adapter.buildChatRequest( | ||
| for: model, | ||
| commonPayload: [:], | ||
| messages: [msg], | ||
| tools: nil, |
There was a problem hiding this comment.
suggestion (bug_risk): Silently skipping messages that fail request-body construction can lead to partial, surprising batch submissions.
When buildChatRequest returns nil or the body can't be converted to JSONValue, the loop just continues, producing fewer BatchRequestItems than input messages and giving no indication of which ones were dropped. Prefer failing fast (throwing) when any message cannot be transformed, or at minimum log skipped messages to make the behavior observable and predictable.
Suggested implementation:
// 1. 构建 BatchRequestItems
var batchItems: [BatchRequestItem] = []
for (index, msg) in messages.enumerated() {
let customId = "req-\(sessionID.uuidString)-\(msg.id.uuidString)-\(index)"
// 构造请求体：由于 APIAdapter 没有暴露暴露纯 JSON 构造，
// 我们可以利用 buildChatRequest 并截获其 httpBody
let request = adapter.buildChatRequest(
for: model,
commonPayload: [:],
messages: [msg],
tools: nil,
audioAttachments: [:],
guard let httpBody = request.httpBody else {
throw NSError(
domain: "BatchService",
code: -2,
userInfo: [
NSLocalizedDescriptionKey: "无法为消息 \(msg.id) 构造请求体，Batch 提交已中止。"
]
)
}
guard
let jsonObject = try? JSONSerialization.jsonObject(with: httpBody),
let jsonBody = JSONValue(jsonObject)
else {
throw NSError(
domain: "BatchService",
code: -3,
userInfo: [
NSLocalizedDescriptionKey: "无法将消息 \(msg.id) 的请求体转换为 JSON，Batch 提交已中止。"
]
)
}
- If the file defines a custom error type for batch operations (e.g.
BatchError), prefer throwing that instead ofNSError, preserving the fail-fast semantics but aligning with existing error handling conventions. - Consider adding structured logging where your logging infrastructure exists (e.g. before throwing, log the
sessionID,index, andmsg.id) so operators can correlate failures with specific messages. - If callers rely on partial success, you may want to wrap this function or introduce an alternative API that returns detailed per-message failure information instead of throwing on the first error.
Comment on lines
+21
to
+23
| let batchDir = docsDir.appendingPathComponent("BatchJobs") | ||
| if !FileManager.default.fileExists(atPath: batchDir.path) { | ||
| try? FileManager.default.createDirectory(at: batchDir, withIntermediateDirectories: true) |
There was a problem hiding this comment.
issue (bug_risk): Directory creation failures are silently ignored, which can cause later persistence operations to fail in non-obvious ways.
If createDirectory fails (e.g., permissions, disk issues, invalid path), the error is suppressed and later reads/writes to jobs.json will likely fail or misbehave. Please handle this failure explicitly (e.g., log via logger or propagate the error) so storage issues are visible and diagnosable rather than silently ignored.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This pull request introduces a Batch API processing framework for ETOS LLM Studio, enabling local persistence, lifecycle management, and OpenAI-specific adapter support for batch jobs. The review feedback highlights critical improvements for robustness and extensibility, including: preventing background polling task leaks when jobs are deleted, utilizing the adapter's download response parser for future-proofing, adding error logging for missing configurations or failed batch items, and correctly parsing and using actual API timestamps instead of hardcoding the current date.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
Comment on lines
+141
to
+144
| let currentJob = BatchJobStore.shared.getJob(id: job.id) | ||
| if let status = currentJob?.status, status == .completed || status == .failed || status == .expired || status == .cancelled { | ||
| break | ||
| } |
Contributor
There was a problem hiding this comment.
在轮询任务状态时，如果任务在本地 BatchJobStore 中被删除（例如用户删除了该任务），BatchJobStore.shared.getJob(id:) 将返回 nil。此时 currentJob?.status 为 nil，导致 if let status = ... 条件不成立，循环不会退出。这会导致后台轮询任务无限期地继续运行，造成后台任务泄漏。
建议使用 guard let 提前解包，如果任务已被删除则直接退出循环。
guard let currentJob = BatchJobStore.shared.getJob(id: job.id) else {
break
}
let status = currentJob.status
if status == .completed || status == .failed || status == .expired || status == .cancelled {
break
}
Comment on lines
+195
to
+196
| let data = try await ChatService.shared.fetchData(for: req, provider: model.provider) | ||
| let jsonlString = String(data: data, encoding: .utf8) ?? "" |
Contributor
There was a problem hiding this comment.
在下载并处理 Batch 结果时，代码直接将下载的原始数据转换为字符串，而没有调用适配器的 parseBatchResultDownloadResponse(data:) 方法。
虽然 OpenAI 适配器目前直接返回原始数据，但其他适配器（或未来的自定义适配器）可能会对下载的数据进行解密、解压或格式转换。为了保证协议的完整性和未来的可扩展性，应当调用该方法处理下载的数据。
Suggested change
| let data = try await ChatService.shared.fetchData(for: req, provider: model.provider) | |
| let jsonlString = String(data: data, encoding: .utf8) ?? "" | |
| let rawData = try await ChatService.shared.fetchData(for: req, provider: model.provider) | |
| let data = try adapter.parseBatchResultDownloadResponse(data: rawData) | |
| let jsonlString = String(data: data, encoding: .utf8) ?? "" |
Comment on lines
+161
to
+165
| let providers = ChatService.shared.providers | ||
| guard let provider = providers.first(where: { $0.id == job.providerID }), | ||
| let modelDef = provider.models.first(where: { $0.id.uuidString == job.modelID }) else { | ||
| return | ||
| } |
Contributor
There was a problem hiding this comment.
在 checkStatus 中，如果找不到对应的提供商或模型配置，方法会静默返回。这会导致轮询任务在后台继续运行（因为状态没有更新为终态），但实际上它什么也做不了，且没有任何日志输出，极难排查问题。
建议在返回前记录一条错误日志，以便于调试和维护。
Suggested change
| let providers = ChatService.shared.providers | |
| guard let provider = providers.first(where: { $0.id == job.providerID }), | |
| let modelDef = provider.models.first(where: { $0.id.uuidString == job.modelID }) else { | |
| return | |
| } | |
| let providers = ChatService.shared.providers | |
| guard let provider = providers.first(where: { $0.id == job.providerID }), | |
| let modelDef = provider.models.first(where: { $0.id.uuidString == job.modelID }) else { | |
| logger.error("无法为任务 \(job.id) 找到对应的提供商或模型配置 (ProviderID: \(job.providerID), ModelID: \(job.modelID))。") | |
| return | |
| } |
Comment on lines
+203
to
+210
| let responseItem = try JSONDecoder().decode(BatchResponseItem.self, from: lineData) | ||
| if let payloadBody = responseItem.response?.body { | ||
| // 转回 Data 再丢给原先的 adapter 解析 | ||
| if let rawData = try? JSONEncoder().encode(payloadBody) { | ||
| let msg = try adapter.parseResponse(data: rawData) | ||
| generatedMessages.append(msg) | ||
| } | ||
| } |
Contributor
There was a problem hiding this comment.
在解析 Batch 结果时，如果某个请求项在云端执行失败（即 response 为 nil，但 error 字段有值），当前代码会静默跳过该行，没有任何错误提示。
建议在 response 为 nil 且 error 存在时，记录相应的错误日志，方便用户或开发者排查批处理中具体请求项的失败原因。
Suggested change
| let responseItem = try JSONDecoder().decode(BatchResponseItem.self, from: lineData) | |
| if let payloadBody = responseItem.response?.body { | |
| // 转回 Data 再丢给原先的 adapter 解析 | |
| if let rawData = try? JSONEncoder().encode(payloadBody) { | |
| let msg = try adapter.parseResponse(data: rawData) | |
| generatedMessages.append(msg) | |
| } | |
| } | |
| let responseItem = try JSONDecoder().decode(BatchResponseItem.self, from: lineData) | |
| if let payloadBody = responseItem.response?.body { | |
| if let rawData = try? JSONEncoder().encode(payloadBody) { | |
| let msg = try adapter.parseResponse(data: rawData) | |
| generatedMessages.append(msg) | |
| } | |
| } else if let errorPayload = responseItem.error { | |
| logger.error("Batch 请求项 \(responseItem.customId) 失败: \(String(describing: errorPayload))") | |
| } |
Comment on lines
+110
to
+117
| struct OpenAIBatchJobResponse: Decodable { | ||
| let id: String | ||
| let status: String | ||
| let input_file_id: String? | ||
| let output_file_id: String? | ||
| let error_file_id: String? | ||
| let endpoint: String? | ||
| } |
Contributor
There was a problem hiding this comment.
OpenAI Batch API 响应中实际上包含了 created_at、completed_at 和 failed_at 等 Unix 时间戳。当前的数据模型 OpenAIBatchJobResponse 遗漏了这些字段，导致在解析状态时只能使用当前时间 Date() 作为替代，这会导致任务的创建、完成和失败时间极不准确（每次轮询都会被更新为当前时间）。
建议在模型中补充这些时间戳字段。
struct OpenAIBatchJobResponse: Decodable {
let id: String
let status: String
let input_file_id: String?
let output_file_id: String?
let error_file_id: String?
let endpoint: String?
let created_at: Int?
let completed_at: Int?
let failed_at: Int?
}
Comment on lines
+134
to
+146
| return BatchJob( | ||
| id: response.id, | ||
| providerID: UUID(), // Caller should correct this | ||
| modelID: "", // Caller should correct this | ||
| status: status, | ||
| createdAt: Date(), | ||
| completedAt: status == .completed ? Date() : nil, | ||
| failedAt: status == .failed ? Date() : nil, | ||
| inputFileId: response.input_file_id, | ||
| outputFileId: response.output_file_id, | ||
| errorFileId: response.error_file_id, | ||
| endpoint: response.endpoint | ||
| ) |
Contributor
There was a problem hiding this comment.
配合 OpenAIBatchJobResponse 中新增的时间戳字段，在解析 BatchJob 时应当使用 API 返回的真实时间戳，而不是硬编码为当前时间 Date()。
Suggested change
| return BatchJob( | |
| id: response.id, | |
| providerID: UUID(), // Caller should correct this | |
| modelID: "", // Caller should correct this | |
| status: status, | |
| createdAt: Date(), | |
| completedAt: status == .completed ? Date() : nil, | |
| failedAt: status == .failed ? Date() : nil, | |
| inputFileId: response.input_file_id, | |
| outputFileId: response.output_file_id, | |
| errorFileId: response.error_file_id, | |
| endpoint: response.endpoint | |
| ) | |
| return BatchJob( | |
| id: response.id, | |
| providerID: UUID(), // Caller should correct this | |
| modelID: "", // Caller should correct this | |
| status: status, | |
| createdAt: response.created_at.map { Date(timeIntervalSince1970: TimeInterval($0)) } ?? Date(), | |
| completedAt: response.completed_at.map { Date(timeIntervalSince1970: TimeInterval($0)) }, | |
| failedAt: response.failed_at.map { Date(timeIntervalSince1970: TimeInterval($0)) }, | |
| inputFileId: response.input_file_id, | |
| outputFileId: response.output_file_id, | |
| errorFileId: response.error_file_id, | |
| endpoint: response.endpoint | |
| ) |
本次提交为应用底层核心增加了对服务端大批量离线异步生成（Batch API）的支持，大幅降低大批量文本处理场景的调用成本（享官方折扣），并在后台实现全自动化管理与轮询。 主要改动： - feat(Protocol): 在 `APIAdapter` 协议中新增 Batch 生命周期的 4 个必须接口（上传、创建、轮询、下载）。 - feat(OpenAI): 增加针对 OpenAI `/v1/batches` 与 `/v1/files` 的全套适配及 `Multipart/form-data` JSONL 的上传构建逻辑。 - feat(Service): 新增 `BatchService` 作为核心处理器，全自动化完成普通 `[ChatMessage]` 转 JSONL 的打包、上传、静默轮询挂起，并在云端完结后自动拉取解析。 - feat(Persistence): 新增基于轻量级 JSON 的持久化 `BatchJobStore`，确保应用重新启动后后台未完结的 Batch 任务仍能恢复追踪。 - fix(CI): 修复 `.github/workflows/pr-build-check.yml` 中 `xcodebuild` 使用 `-workspace` 解析依赖时缺少 `-scheme` 导致工作流失败的问题。 - fix(Build): 解决 Swift 6 严格并发检查对于后台异步状态挂起所引发的 `@Sendable` 闭包隔离报错；并将新编写的辅助逻辑代码合并入已有的工程追踪文件（`APIAdapterSupport.swift`等）中，规避 CI 缺失 `.pbxproj` 引用的编译异常。
Owner
|
我自己来吧 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
改动内容
APIAdapterSupport.swift中扩展了APIAdapter协议，新增了与 Batch 生命周期相关的 4 个标准接口（包含文件上传、创建批处理、查询状态、下载结果）。BatchModels.swift，定义了应用级的BatchJob、BatchJobStatus以及 JSONL 请求响应相关的BatchRequestItem、BatchResponseItem。OpenAIAdapterModels.swift中补充了针对 OpenAI 响应的OpenAIFileUploadResponse和OpenAIBatchJobResponse。OpenAIAdapterBatchSupport.swift，基于 OpenAI Batch API 的标准完全实现了上传 JSONL、创建 Batch 任务、状态解析与结果下载的逻辑。BatchJobStore.swift，采用基于本地 JSON (jobs.json) 的轻量级存储，管理正在运行和已完成的 Batch 任务状态，确保应用重启不丢失状态。BatchService.swift。该服务负责将用户的常规聊天消息 ([ChatMessage]) 自动打包为临时 JSONL 格式发送上传，发起 Batch 调用后使用异步任务静默轮询更新状态，并在任务最终完成（或失败）时自动下载解析生成结果。验证
BatchService.shared.submitBatch()，验证能否正确生成符合{"custom_id":..., "method": "POST", "url": "/v1/chat/completions", "body": ...}规范的临时 JSONL 文件。in_progress状态的 Batch 任务可以通过BatchJobStore被正确加载并恢复轮询。截图或录屏
CLA
Summary by Sourcery
Add end-to-end batch processing support, including protocol extensions, models, service layer, OpenAI adapter integration, and persistence of batch job state.
New Features:
Enhancements: