Merge C-1: ASR provider abstraction + Volcengine wrap

Phase C-1 of provider abstraction. Adds ASRProvider / ASRStreamingSession
protocols + ASRError / ASRMode / ASRProviderInfo in OpenLessCore.
VolcengineASRProvider wraps existing VolcengineStreamingASR via adapter
without modifying the Volcengine class itself (zero edits to
VolcengineStreamingASR/Frame/Credentials). DictationCoordinator now
routes ASR through the protocol via a hardcoded makeASRProvider()
factory; vault.activeASRProviderId plumbing comes in Wave 3c.

AudioConsumer bridged via default-impl forwarder so existing
Recorder→Buffering→session.consume(pcmChunk:) flow is unchanged.

partialResults stream exposed but always finishes immediately —
VolcengineStreamingASR doesn't expose partial callbacks and we promised
zero edits to it. Confirmed no UI surface used partial transcripts, so
this is a documentation-only difference, not a regression.

Branch: feature/asr-provider-abstraction
This commit is contained in:
baiqing committed 2026-04-29 12:33:59 +08:00
commit b59c578b55
5 files changed
+418 -22

No files matched your search

+5
View File
@@ -95,6 +95,11 @@ let package = Package(
dependencies: ["OpenLessPersistence", "OpenLessCore"],
path: "Tests/OpenLessPersistenceTests"
),
.testTarget(
name: "OpenLessASRTests",
dependencies: ["OpenLessASR", "OpenLessCore"],
path: "Tests/OpenLessASRTests"
),
],
// 暂留 Swift 5 语言模式,避免一次性吞掉 Swift 6 严格并发的全部改造。
// 升 .v6 时需要重新审视 Recorder / VolcengineStreamingASR 等 @unchecked Sendable 类。
@@ -0,0 +1,119 @@
import Foundation
import OpenLessCore
// VolcengineASRProvider 把现有的 `VolcengineStreamingASR` 包装成符合
// `ASRProvider` / `ASRStreamingSession` 协议的对象。
//
// 这一层是「纯外包装」——不修改 VolcengineStreamingASR 任何内部逻辑,
// 仅把它的 openSession / consume / sendLastFrame / awaitFinalResult / cancel
// 暴露给 coordinator 通过 protocol 调用。
//
// 现有 `VolcengineStreamingASR` 的几个特性决定了这里的妥协:
// 1. 它在 `init` 时一次性接收 `dictionaryEntries`(即热词),没有运行时换词的接口。
// 所以 hotwords 在 openStreamingSession 阶段被构造成 DictionaryEntry,再
// 传进 VolcengineStreamingASR.init —— 我们不动它。
// 2. 它的 `cancel()` 是同步方法;我们在 `ASRStreamingSession.cancel()` 里 await
// 没有意义但符合协议(async 包同步是无害的)。
// 3. 它没有暴露 partial 回调。火山协议本身有增量结果,但旧实现把它们累到
// `rawText` 里直到拿到 final 才 fire continuation;本次 C-1 不动这个逻辑,
// `partialResults` 返回空 stream。需要 partial UI 时再回头改 ASR 内部。
public final class VolcengineASRProvider: ASRProvider {
public let info: ASRProviderInfo
private let credentials: VolcengineCredentials
private let logger: (@Sendable (String) -> Void)?
public init(
credentials: VolcengineCredentials,
logger: (@Sendable (String) -> Void)? = nil
) {
self.credentials = credentials
self.logger = logger
self.info = ASRProviderInfo(
providerId: "volcengine",
displayName: "火山引擎",
mode: .streaming,
supportsHotwords: true,
supportsLanguageHint: false,
supportsPartialResults: false
)
}
public func openStreamingSession(
language: String,
hotwords: [String]
) async throws -> ASRStreamingSession {
// 把 hotwords[String] 适配回 DictionaryEntry(VolcengineStreamingASR 的入参形态)。
// language 暂不消费——VolcengineStreamingASR 当前没有暴露 language 入参。
// C-3/C-4 接入多 ASR provider 时再回来扩展。
let entries = hotwords
.filter { !$0.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty }
.map { phrase in
DictionaryEntry(phrase: phrase)
}
let asr = VolcengineStreamingASR(
credentials: credentials,
dictionaryEntries: entries,
logger: logger
)
try await asr.openSession()
return VolcengineStreamingSessionAdapter(asr: asr)
}
public func transcribeBatch(
pcm: Data,
sampleRate: Int,
channels: Int,
language: String,
hotwords: [String]
) async throws -> RawTranscript {
throw ASRError.unsupportedMode
}
}
/// 适配 `VolcengineStreamingASR` 到 `ASRStreamingSession`。
///
/// 它本身是 `AudioConsumer`:`consume(pcmChunk:)` 直接转发给底层 ASR 实例。
/// `partialResults` 暂时是个空 stream(在 init 里立即 finish),UI 即便订阅也不会拿到东西。
final class VolcengineStreamingSessionAdapter: ASRStreamingSession, @unchecked Sendable {
private let asr: VolcengineStreamingASR
let partialResults: AsyncStream<String>
private let partialContinuation: AsyncStream<String>.Continuation
init(asr: VolcengineStreamingASR) {
self.asr = asr
var capturedContinuation: AsyncStream<String>.Continuation!
self.partialResults = AsyncStream<String> { continuation in
capturedContinuation = continuation
}
self.partialContinuation = capturedContinuation
// 当前 VolcengineStreamingASR 不暴露 partial 回调,立即关掉 stream,避免订阅方挂等。
// 等火山 partial 真接入时,把这一行删掉,改成 onPartial { text in cont.yield(text) }。
capturedContinuation.finish()
}
// MARK: - AudioConsumer
func consume(pcmChunk: Data) {
asr.consume(pcmChunk: pcmChunk)
}
// MARK: - ASRStreamingSession
func sendAudio(_ pcm: Data) async throws {
asr.consume(pcmChunk: pcm)
}
func endStream() async throws {
try await asr.sendLastFrame()
}
func awaitFinalResult() async throws -> RawTranscript {
return try await asr.awaitFinalResult()
}
func cancel() async {
asr.cancel()
}
}
+44 -22
View File
@@ -30,7 +30,10 @@ final class DictationCoordinator {
/// Toggle 状态:idle → 开始;listening → 结束;starting/processing 阶段忽略重复触发。
private var sessionPhase: SessionPhase = .idle
private var asr: VolcengineStreamingASR?
/// 当前 ASR session(通过 ASRProvider.openStreamingSession 拿到)。
/// 抽象化后 coordinator 不再直接持有 VolcengineStreamingASR;具体 provider 由
/// `makeASRProvider()` 选择,C-3 起会从 vault.activeASRProviderId 路由到不同实现。
private var asrSession: ASRStreamingSession?
private var audioConsumer: BufferingAudioConsumer?
private var sessionStartedAt: Date = Date()
/// hold 模式下,Esc 取消后下一次 .released 应被忽略(否则会再次触发结束流程)。
@@ -211,8 +214,11 @@ final class DictationCoordinator {
guard sessionPhase == .starting || sessionPhase == .listening else { return }
Log.write("用户取消")
sessionPhase = .idle
asr?.cancel()
asr = nil
if let session = asrSession {
asrSession = nil
// session.cancel() 是 async;让它在后台 detach 即可,不阻塞 UI/cancel 路径。
Task { await session.cancel() }
}
recorder.stop()
audioConsumer?.clear()
audioConsumer = nil
@@ -261,28 +267,27 @@ final class DictationCoordinator {
}
guard sessionPhase == .starting else { return }
guard let credentials = loadVolcengineCredentials() else {
Log.write("缺少火山引擎凭据;麦克风已启动,结束后走 mock 模式")
guard let provider = makeASRProvider() else {
Log.write("缺少 ASR 凭据;麦克风已启动,结束后走 mock 模式")
sessionPhase = .listening
return
}
let dictionaryEntries = dictionary.enabledEntries()
let asr = VolcengineStreamingASR(
credentials: credentials,
dictionaryEntries: dictionaryEntries,
logger: { msg in Log.write(msg) }
)
self.asr = asr
let hotwords = dictionaryEntries.map(\.trimmedPhrase).filter { !$0.isEmpty }
do {
try await asr.openSession()
Log.write("[asr] WebSocket 已连接")
let session = try await provider.openStreamingSession(
language: "zh-CN",
hotwords: hotwords
)
Log.write("[asr] streaming session opened (provider=\(provider.info.providerId))")
guard sessionPhase == .starting else {
asr.cancel()
await session.cancel()
return
}
audioConsumer.attach(asr)
self.asrSession = session
audioConsumer.attach(session)
Log.write("[asr] 音频已接入 ASR")
sessionPhase = .listening
} catch {
@@ -310,17 +315,17 @@ final class DictationCoordinator {
audioConsumer = nil
Log.write("[session] 录音停止(含 250ms 末尾 padding),等待 ASR 终态")
guard let asr = self.asr else {
guard let session = self.asrSession else {
// mock pipeline:无凭据时给一段提示
await runMockPipeline()
return
}
do {
try await asr.sendLastFrame()
let raw = try await asr.awaitFinalResult()
try await session.endStream()
let raw = try await session.awaitFinalResult()
Log.write("[asr] final: \(raw.text)")
self.asr = nil
self.asrSession = nil
let trimmed = raw.text.trimmingCharacters(in: .whitespacesAndNewlines)
if trimmed.isEmpty {
Log.write("[asr] 终态文本为空 — 跳过润色与插入")
@@ -333,7 +338,7 @@ final class DictationCoordinator {
)
} catch {
Log.write("[asr] final 失败: \(error)")
self.asr = nil
self.asrSession = nil
await failSession(reason: "识别失败")
}
}
@@ -517,8 +522,10 @@ final class DictationCoordinator {
recorder.stop()
audioConsumer?.clear()
audioConsumer = nil
asr?.cancel()
asr = nil
if let session = asrSession {
asrSession = nil
await session.cancel()
}
sessionPhase = .idle
capsule.update(state: .error(reason))
DispatchQueue.main.asyncAfter(deadline: .now() + 1.6) { [weak self] in
@@ -568,4 +575,19 @@ final class DictationCoordinator {
// loadArkCredentials 已移除:润色路径改走 vault.llmProviderConfig(for:) +
// OpenAICompatibleLLMProvider,不再单独构造 ArkCredentials。
// MARK: - ASR provider 工厂
/// 构造当前会话使用的 `ASRProvider`。
///
/// C-1 阶段硬编码到 `VolcengineASRProvider`;C-3(Settings UI)会改成从
/// `vault.activeASRProviderId` 路由。返回 nil 时 coordinator 走 mock 流程
/// (沿用旧版「凭据缺失 → 提示性占位」的契约,不向用户报硬错误)。
private func makeASRProvider() -> ASRProvider? {
guard let creds = loadVolcengineCredentials() else { return nil }
return VolcengineASRProvider(
credentials: creds,
logger: { msg in Log.write(msg) }
)
}
}
+134
View File
@@ -0,0 +1,134 @@
import Foundation
// ASRProvider 是录音层与具体 ASR 供应商之间的抽象。
//
// 与 LLMProvider 的差异:ASR 有「流式」和「批量」两种交互形态。
// - 流式(火山 / 阿里 paraformer / Apple Speech):边录边推 PCM,partial → final
// - 批量(OpenAI Whisper):录完整体上传,等几秒拿 final
//
// 我们在 protocol 层一并暴露两种入口;具体 provider 用 `info.mode` 自报家门,
// 调到错误入口时抛 `.unsupportedMode`。这样 coordinator 调用的代码在两种模式下
// 的「形状」是一致的——选择由配置决定,业务流程不分叉。
//
// AudioConsumer 桥接策略:
// `AudioConsumer.consume(pcmChunk:)` 是同步 + 不抛错的现存协议(OpenLessRecorder /
// BufferingAudioConsumer 都依赖它)。`ASRStreamingSession` 同时继承 AudioConsumer,
// provider 自己实现 `consume(pcmChunk:)`(通常委托给内部异步发送任务)。这是最小侵入
// 方案——不改 AudioConsumer,不改任何现有 conformer 的签名。
/// ASR 提供两种交互形态。
public enum ASRMode: String, Codable, Sendable {
/// 边录边推 PCM,边拿 partial 边等 final。火山 / 阿里 / Apple Speech 都属此类。
case streaming
/// 录完整体上传,等几秒拿 final。OpenAI Whisper 属此类。
case batch
}
/// Provider 元数据:用于 UI 选择列表 + coordinator 路由决策。
public struct ASRProviderInfo: Sendable, Hashable {
/// 稳定 slug,例如 "volcengine" / "openai-whisper" / "aliyun-paraformer" / "apple-speech"。
public let providerId: String
/// 给 UI 显示的中文名,例如「火山引擎」。
public let displayName: String
public let mode: ASRMode
/// 是否支持「真热词」(服务器端生效),不是 prompt 软提示。
public let supportsHotwords: Bool
public let supportsLanguageHint: Bool
/// 流式 provider 是否会推送 partial(增量识别结果)。批量 provider 应为 false。
public let supportsPartialResults: Bool
public init(
providerId: String,
displayName: String,
mode: ASRMode,
supportsHotwords: Bool,
supportsLanguageHint: Bool,
supportsPartialResults: Bool
) {
self.providerId = providerId
self.displayName = displayName
self.mode = mode
self.supportsHotwords = supportsHotwords
self.supportsLanguageHint = supportsLanguageHint
self.supportsPartialResults = supportsPartialResults
}
}
/// ASR provider 抽象。
///
/// 同一份 protocol 同时承载流式 / 批量两类入口;非自家形态直接抛 `.unsupportedMode`。
public protocol ASRProvider: Sendable {
var info: ASRProviderInfo { get }
/// 流式入口:开 session,coordinator 后续把 PCM 推进 session(session 是 AudioConsumer)。
/// - Parameters:
/// - language: BCP-47 语言提示,例如 "zh-CN" / "en-US"。不支持的 provider 可忽略。
/// - hotwords: 用户词典中启用的词条;服务器端真热词。
/// - Throws: 批量 provider 应抛 `ASRError.unsupportedMode`;连接失败抛对应错误。
func openStreamingSession(language: String, hotwords: [String]) async throws -> ASRStreamingSession
/// 批量入口:传完整 PCM,返回 final transcript。
/// - Throws: 流式 provider 应抛 `ASRError.unsupportedMode`。
func transcribeBatch(
pcm: Data,
sampleRate: Int,
channels: Int,
language: String,
hotwords: [String]
) async throws -> RawTranscript
}
/// 流式 session 抽象。
///
/// 同时继承 `AudioConsumer`:可以直接 `BufferingAudioConsumer.attach(session)`,
/// recorder 推 PCM → session.consume(pcmChunk:) → provider 内部发送队列。
public protocol ASRStreamingSession: AudioConsumer, Sendable {
/// 显式异步发送一段 PCM(16 kHz / 16-bit LE / mono);framing / 编码 / gzip 由 provider 自负。
/// 默认实现转发到 `AudioConsumer.consume(pcmChunk:)`,供 provider 选择性覆盖。
func sendAudio(_ pcm: Data) async throws
/// 用户停止说话,告诉 provider 收尾(不是 cancel)。
func endStream() async throws
/// 可选 partial:UI 想显示就订阅,不订阅也不影响 final。批量 / 不支持的 provider 应返回空 stream。
var partialResults: AsyncStream<String> { get }
/// 阻塞等 final transcript。`endStream()` 之后调用。
func awaitFinalResult() async throws -> RawTranscript
/// 用户按 Esc / 取消。drop 所有内部状态。
func cancel() async
}
/// `sendAudio(_:)` 的默认实现:直接走 `consume(pcmChunk:)`。Provider 若有更精确的
/// 异步控制(背压 / 错误回流)可自己覆盖。
public extension ASRStreamingSession {
func sendAudio(_ pcm: Data) async throws {
consume(pcmChunk: pcm)
}
}
/// ASR provider 抛出的错误。
///
/// 与 `LLMError` 风格保持一致:URLError 不直接暴露(不 Sendable / 不稳定 Equatable),
/// 网络错误统一用 NSError 包装。
public enum ASRError: Error, Sendable, Equatable {
/// 凭据缺失(apiKey / appId 等)。
case missingCredentials
/// 流式 provider 调批量入口或反之。
case unsupportedMode
/// 鉴权失败。`statusCode` 可空——某些协议在 WebSocket 握手后才报错。
case authFailed(statusCode: Int?)
/// 配额或限流。
case quotaExceeded
/// 输入 PCM 不合法(采样率 / 声道数错等)。
case invalidAudio(String)
/// 服务端返回业务错误(带 code + message)。
case providerError(code: String, message: String)
/// 操作超时。
case timeout
/// 其他网络错误。
case network(NSError)
/// 用户主动取消。
case cancelled
}
@@ -0,0 +1,116 @@
import XCTest
@testable import OpenLessASR
import OpenLessCore
final class VolcengineASRProviderTests: XCTestCase {
private func makeProvider() -> VolcengineASRProvider {
let creds = VolcengineCredentials(
appID: "test-app",
accessToken: "test-token",
resourceID: VolcengineCredentials.defaultResourceId
)
return VolcengineASRProvider(credentials: creds)
}
// MARK: - info 元数据
func test_info_reports_volcengine_streaming_provider() {
// Arrange
let provider = makeProvider()
// Act
let info = provider.info
// Assert
XCTAssertEqual(info.providerId, "volcengine")
XCTAssertEqual(info.mode, .streaming)
XCTAssertTrue(info.supportsHotwords)
}
func test_info_displayName_is_localized_chinese() {
// Arrange
let provider = makeProvider()
// Act
let displayName = provider.info.displayName
// Assert
XCTAssertEqual(displayName, "火山引擎")
}
// MARK: - 批量入口在流式 provider 上必须明确不支持
func test_transcribeBatch_throws_unsupportedMode() async {
// Arrange
let provider = makeProvider()
let pcm = Data(repeating: 0, count: 32_000) // 1 秒静音
// Act / Assert
do {
_ = try await provider.transcribeBatch(
pcm: pcm,
sampleRate: 16_000,
channels: 1,
language: "zh-CN",
hotwords: []
)
XCTFail("transcribeBatch 应抛 ASRError.unsupportedMode")
} catch let error as ASRError {
XCTAssertEqual(error, .unsupportedMode)
} catch {
XCTFail("应抛 ASRError,实际抛: \(error)")
}
}
}
// MARK: - ASRProviderInfo / ASRError 基础形状
final class ASRProviderInfoTests: XCTestCase {
func test_info_struct_is_value_equal() {
// Arrange
let a = ASRProviderInfo(
providerId: "x",
displayName: "X",
mode: .streaming,
supportsHotwords: true,
supportsLanguageHint: true,
supportsPartialResults: true
)
let b = ASRProviderInfo(
providerId: "x",
displayName: "X",
mode: .streaming,
supportsHotwords: true,
supportsLanguageHint: true,
supportsPartialResults: true
)
// Act / Assert
XCTAssertEqual(a, b)
}
func test_asrError_unsupportedMode_is_equatable() {
// Arrange
let lhs: ASRError = .unsupportedMode
let rhs: ASRError = .unsupportedMode
// Act / Assert
XCTAssertEqual(lhs, rhs)
XCTAssertNotEqual(lhs, .missingCredentials)
XCTAssertNotEqual(lhs, .timeout)
}
func test_asrMode_codable_roundtrips() throws {
// Arrange
let original: ASRMode = .batch
// Act
let encoded = try JSONEncoder().encode(original)
let decoded = try JSONDecoder().decode(ASRMode.self, from: encoded)
// Assert
XCTAssertEqual(decoded, original)
}
}