import Foundation import Metal // ───────────────────────────────────────────────────────────── // Async Inference Optimizations // ───────────────────────────────────────────────────────────── /// Async inference result public struct InferenceResult { public let logits: [Float] public let elapsed: TimeInterval public let token: Int public init(logits: [Float], elapsed: TimeInterval, token: Int = 0) { self.logits = logits self.elapsed = elapsed self.token = token } } /// Async inference queue for batching requests public final class AsyncInferenceQueue { private let maxBatchSize: Int private var pending: [(input: Int, position: Int, completion: (Result) -> Void)] = [] private let lock = NSLock() public init(maxBatchSize: Int = 8) { self.maxBatchSize = maxBatchSize } /// Add inference request public func enqueue(input: Int, position: Int, completion: @escaping (Result) -> Void) { lock.lock() pending.append((input, position, completion)) let shouldProcess = pending.count >= maxBatchSize lock.unlock() if shouldProcess { processBatch() } } /// Process batch of requests private func processBatch() { lock.lock() let batch = pending.prefix(maxBatchSize) pending.removeFirst(min(batch.count, maxBatchSize)) lock.unlock() // TODO: Implement actual batch processing // This would require modifications to the model to support batch inference } } /// Async token generator with prefetching public final class AsyncTokenGenerator: @unchecked Sendable { private let model: E4BModel private let tokenizer: Tokenizer private let engine: MarkBaseEngine private let sampler: Sampler private var currentLogits: [Float]? private var currentPosition: Int = 0 private var generatedTokens: [Int] = [] public init(model: E4BModel, tokenizer: Tokenizer, engine: MarkBaseEngine) { self.model = model self.tokenizer = tokenizer self.engine = engine self.sampler = Sampler() } /// Start generation with async streaming public func start(prompt: String, config: GenerationConfig) -> AsyncStream { return AsyncStream { [weak self] continuation in guard let self = self else { continuation.finish() return } Task.detached { do { try await self.generateAsync(prompt: prompt, config: config) { tokenText in continuation.yield(tokenText) } continuation.finish() } catch { continuation.finish() } } } } /// Async generation with callback private func generateAsync( prompt: String, config: GenerationConfig, onToken: (String) -> Void ) async throws { let promptTokens = tokenizer.encode(text: prompt) // Pre-fill KV cache var lastLogits: [Float] = [] for (position, tokenId) in promptTokens.enumerated() { lastLogits = try model.forward(tokenId: tokenId, position: position) } currentLogits = lastLogits currentPosition = promptTokens.count generatedTokens = [] var streamDecoder = StreamingDecoder(tokenizer: tokenizer) // Generate tokens for _ in 0.. [Float] { if let logits = prefetchedLogits, nextToken == token, nextPosition == position { prefetchedLogits = nil return logits } // Compute on demand return try model.forward(tokenId: token, position: position) } }