diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 36b5fab1e..5174d4485 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -24,9 +24,8 @@ Thank you for your interest in contributing to Cactus! This document covers the 1. Fork the repository and create a branch from `main`. 2. Make your changes, keeping the scope focused on the relevant GitHub issue. 3. Run `cactus test` to verify your changes build and pass all tests. -4. Run `cactus test --suite performance` if your changes affect performance-critical paths. -5. Update documentation if your changes affect the public API or user-facing behavior. -6. Submit a pull request with a clear description of what you changed and why. +4. Update documentation if your changes affect the public API or user-facing behavior. +5. Submit a pull request with a clear description of what you changed and why. ## Testing @@ -40,9 +39,6 @@ cactus test --ios # Run tests on a connected Android device cactus test --android -# Run performance tests -cactus test --suite performance - # Test a specific model cactus test --model LiquidAI/LFM2-VL-450M ``` diff --git a/README.md b/README.md index e3bfd185c..802b2606f 100644 --- a/README.md +++ b/README.md @@ -87,7 +87,9 @@ Example response from Gemma3-270m "cloud_handoff": false, // true if cloud model used "response": "Hi there!", "function_calls": [], // parsed tool calls + "segments": [], // transcription segments (empty for chat) "confidence": 0.8193, // model confidence + "confidence_threshold": 0.7, // resolved handoff threshold (model-dependent) "time_to_first_token_ms": 45.23, "total_time_ms": 163.67, "prefill_tps": 1621.89, @@ -192,9 +194,9 @@ graph.hard_reset(); │ --status show key status │ │ --clear remove saved key │ │ │ -│ cactus run run a model (downloads if needed) │ +│ cactus run [model|path] run a model (downloads if needed) │ │ --bits 1|2|3|4 CQ quantization (default: 4) │ -│ --platform cpu|apple target accelerator (default: cpu) │ +│ --platform auto|cpu|apple target accelerator (default: auto) │ │ --image image file for VLM inference │ │ --audio audio file for audio chat │ │ --system system prompt │ @@ -203,15 +205,15 @@ graph.hard_reset(); │ --token HuggingFace token (gated models) │ │ --reconvert force local convert+transpile fallback │ │ │ -│ cactus transcribe [model] transcribe audio with a model │ -│ --file audio file to transcribe (required) │ +│ cactus transcribe [model] live microphone transcription with a model│ +│ --file audio file to transcribe (WAV) │ │ --language language code (default: en) │ │ --token HuggingFace token (gated models) │ │ --reconvert force reconversion from source │ │ │ -│ cactus download download a pre-built bundle │ +│ cactus download [model] download a pre-built bundle │ │ --bits 1|2|3|4 CQ quantization (default: 4) │ -│ --platform cpu|apple target accelerator (default: cpu) │ +│ --platform auto|cpu|apple target accelerator (default: auto) │ │ --token HuggingFace token │ │ │ │ cactus convert [dir] convert HuggingFace weights to CQ │ @@ -241,15 +243,15 @@ graph.hard_reset(); │ (default: all) │ │ --model default: LiquidAI/LFM2-VL-450M │ │ --transcription-model default: openai/whisper-base │ -│ --suite run a single test suite (resolved │ -│ across components; e.g. performance │ -│ → kernels + graph, llm → engine) │ +│ --suite run a single test suite by name │ +│ (resolved across components, │ +│ e.g. llm → engine) │ │ --list list components and suites │ │ --ios run on connected iPhone │ │ --android run on connected Android │ │ --enable-telemetry send cloud telemetry (off by default) │ │ │ -│ cactus clean delete build artifacts │ +│ cactus clean delete build artifacts, weights, venv │ │ cactus --help show this help │ │ │ └────────────────────────────────────────────────────────────────────────────────┘ diff --git a/android/cactus_jni.cpp b/android/cactus_jni.cpp index 31371807a..44d06d188 100644 --- a/android/cactus_jni.cpp +++ b/android/cactus_jni.cpp @@ -264,6 +264,54 @@ Java_com_cactus_CactusJNI_nativeTranscribe(JNIEnv* env, jobject, jlong handle, return result; } +JNIEXPORT jlong JNICALL +Java_com_cactus_CactusJNI_nativeStreamTranscribeStart(JNIEnv* env, jobject, jlong handle, + jstring optionsJson) { + const char* options = optionsJson ? env->GetStringUTFChars(optionsJson, nullptr) : nullptr; + cactus_stream_transcribe_t stream = cactus_stream_transcribe_start( + reinterpret_cast(handle), options); + if (options) env->ReleaseStringUTFChars(optionsJson, options); + return reinterpret_cast(stream); +} + +JNIEXPORT jint JNICALL +Java_com_cactus_CactusJNI_nativeStreamTranscribeProcess(JNIEnv* env, jobject, jlong stream, + jbyteArray pcmData, + jbyteArray responseBuffer) { + jbyte* pcmBytes = pcmData ? env->GetByteArrayElements(pcmData, nullptr) : nullptr; + size_t pcmSize = pcmData ? static_cast(env->GetArrayLength(pcmData)) : 0; + + jsize bufSize = env->GetArrayLength(responseBuffer); + jbyte* buf = env->GetByteArrayElements(responseBuffer, nullptr); + + int result = cactus_stream_transcribe_process( + reinterpret_cast(stream), + reinterpret_cast(pcmBytes), pcmSize, + reinterpret_cast(buf), static_cast(bufSize) + ); + + env->ReleaseByteArrayElements(responseBuffer, buf, 0); + if (pcmBytes) env->ReleaseByteArrayElements(pcmData, pcmBytes, JNI_ABORT); + + return result; +} + +JNIEXPORT jint JNICALL +Java_com_cactus_CactusJNI_nativeStreamTranscribeStop(JNIEnv* env, jobject, jlong stream, + jbyteArray responseBuffer) { + jsize bufSize = env->GetArrayLength(responseBuffer); + jbyte* buf = env->GetByteArrayElements(responseBuffer, nullptr); + + int result = cactus_stream_transcribe_stop( + reinterpret_cast(stream), + reinterpret_cast(buf), static_cast(bufSize) + ); + + env->ReleaseByteArrayElements(responseBuffer, buf, 0); + + return result; +} + JNIEXPORT jint JNICALL Java_com_cactus_CactusJNI_nativeEmbed(JNIEnv* env, jobject, jlong handle, jstring text, jfloatArray embeddingsBuffer, diff --git a/bindings/flutter/README.md b/bindings/flutter/README.md index 077219048..1e36bc559 100644 --- a/bindings/flutter/README.md +++ b/bindings/flutter/README.md @@ -23,6 +23,7 @@ cactus build --android ```dart +import 'dart:ffi'; import 'cactus.dart'; import 'package:ffi/ffi.dart'; @@ -40,3 +41,22 @@ calloc.free(buf); cactusDestroy(model); ``` + +## Streaming transcription + +Push 16 kHz mono PCM16 as it arrives; each call returns `{"success":true,"confirmed":...,"pending":...}` (`confirmed` is final, `pending` is the volatile tail). + +```dart +final opts = '{"language":"en"}'.toNativeUtf8(); +final stream = cactusStreamTranscribeStart(model, opts); +final out = calloc(65536); +for (final chunk in pcmChunks) { // each: 16 kHz mono PCM16 bytes + final p = calloc(chunk.length)..asTypedList(chunk.length).setAll(0, chunk); + cactusStreamTranscribeProcess(stream, p, chunk.length, out.cast(), 65536); + calloc.free(p); + // parse {"confirmed":...,"pending":...} from out.cast().toDartString() +} +cactusStreamTranscribeStop(stream, out.cast(), 65536); +calloc.free(opts); +calloc.free(out); +``` diff --git a/bindings/flutter/cactus.dart b/bindings/flutter/cactus.dart index 566580002..f45fb6d25 100644 --- a/bindings/flutter/cactus.dart +++ b/bindings/flutter/cactus.dart @@ -9,6 +9,7 @@ final DynamicLibrary _lib = (Platform.isAndroid || Platform.isLinux) typedef CactusModelT = Pointer; typedef CactusIndexT = Pointer; +typedef CactusStreamTranscribeT = Pointer; typedef TokenCallbackNative = Void Function( Pointer token, Uint32 tokenId, Pointer userData); @@ -146,6 +147,38 @@ final cactusTranscribe = _lib.lookupFunction<_CactusTranscribeN, _CactusTranscribeD>( 'cactus_transcribe'); +typedef _CactusStreamTranscribeStartN = Pointer Function( + Pointer model, Pointer optionsJson); +typedef _CactusStreamTranscribeStartD = Pointer Function( + Pointer model, Pointer optionsJson); +final cactusStreamTranscribeStart = _lib.lookupFunction< + _CactusStreamTranscribeStartN, + _CactusStreamTranscribeStartD>('cactus_stream_transcribe_start'); + +typedef _CactusStreamTranscribeProcessN = Int32 Function( + Pointer stream, + Pointer pcmBuffer, + IntPtr pcmBufferSize, + Pointer responseBuffer, + IntPtr bufferSize); +typedef _CactusStreamTranscribeProcessD = int Function( + Pointer stream, + Pointer pcmBuffer, + int pcmBufferSize, + Pointer responseBuffer, + int bufferSize); +final cactusStreamTranscribeProcess = _lib.lookupFunction< + _CactusStreamTranscribeProcessN, + _CactusStreamTranscribeProcessD>('cactus_stream_transcribe_process'); + +typedef _CactusStreamTranscribeStopN = Int32 Function( + Pointer stream, Pointer responseBuffer, IntPtr bufferSize); +typedef _CactusStreamTranscribeStopD = int Function( + Pointer stream, Pointer responseBuffer, int bufferSize); +final cactusStreamTranscribeStop = _lib.lookupFunction< + _CactusStreamTranscribeStopN, + _CactusStreamTranscribeStopD>('cactus_stream_transcribe_stop'); + typedef _CactusEmbedN = Int32 Function( Pointer model, Pointer text, diff --git a/bindings/kotlin/Cactus.android.kt b/bindings/kotlin/Cactus.android.kt index 553a81f55..b51eea024 100644 --- a/bindings/kotlin/Cactus.android.kt +++ b/bindings/kotlin/Cactus.android.kt @@ -45,6 +45,24 @@ actual fun cactusTranscribe(handle: Long, audioPath: String?, prompt: String, op return buffer.decodeToString().trimEnd('\u0000') } +actual fun cactusStreamTranscribeStart(handle: Long, optionsJson: String?): Long { + val stream = CactusJNI.nativeStreamTranscribeStart(handle, optionsJson) + if (stream == 0L) throw RuntimeException(CactusJNI.nativeGetLastError().ifEmpty { "Failed to start streaming transcription" }) + return stream +} + +actual fun cactusStreamTranscribeProcess(stream: Long, pcmData: ByteArray?): String { + val buffer = ByteArray(65536) + check(CactusJNI.nativeStreamTranscribeProcess(stream, pcmData, buffer)) + return buffer.decodeToString().trimEnd('\u0000') +} + +actual fun cactusStreamTranscribeStop(stream: Long): String { + val buffer = ByteArray(65536) + check(CactusJNI.nativeStreamTranscribeStop(stream, buffer)) + return buffer.decodeToString().trimEnd('\u0000') +} + actual fun cactusEmbed(handle: Long, text: String, normalize: Boolean): FloatArray { val buffer = FloatArray(4096) val outDim = LongArray(1) diff --git a/bindings/kotlin/Cactus.common.kt b/bindings/kotlin/Cactus.common.kt index 42583dd85..dcc9be275 100644 --- a/bindings/kotlin/Cactus.common.kt +++ b/bindings/kotlin/Cactus.common.kt @@ -9,6 +9,9 @@ expect fun cactusPrefill(handle: Long, messagesJson: String, optionsJson: String expect fun cactusTokenize(handle: Long, text: String): IntArray expect fun cactusScoreWindow(handle: Long, tokens: IntArray, start: Long, end: Long, context: Long): String expect fun cactusTranscribe(handle: Long, audioPath: String?, prompt: String, optionsJson: String?, callback: CactusTokenCallback?, pcmData: ByteArray?): String +expect fun cactusStreamTranscribeStart(handle: Long, optionsJson: String?): Long +expect fun cactusStreamTranscribeProcess(stream: Long, pcmData: ByteArray?): String +expect fun cactusStreamTranscribeStop(stream: Long): String expect fun cactusEmbed(handle: Long, text: String, normalize: Boolean): FloatArray expect fun cactusImageEmbed(handle: Long, imagePath: String): FloatArray expect fun cactusAudioEmbed(handle: Long, audioPath: String): FloatArray diff --git a/bindings/kotlin/Cactus.ios.kt b/bindings/kotlin/Cactus.ios.kt index 17f500c8a..98ad5ec07 100644 --- a/bindings/kotlin/Cactus.ios.kt +++ b/bindings/kotlin/Cactus.ios.kt @@ -134,6 +134,40 @@ actual fun cactusTranscribe(handle: Long, audioPath: String?, prompt: String, op } } +@OptIn(ExperimentalForeignApi::class) +actual fun cactusStreamTranscribeStart(handle: Long, optionsJson: String?): Long { + val stream = cactus_stream_transcribe_start(handle.toCPointer(), optionsJson) + ?: throw RuntimeException(cactus_get_last_error()?.toKString() ?: "Failed to start streaming transcription") + return stream.rawValue.toLong() +} + +@OptIn(ExperimentalForeignApi::class) +actual fun cactusStreamTranscribeProcess(stream: Long, pcmData: ByteArray?): String { + memScoped { + val buffer = allocArray(65536) + val pcmPtr = pcmData?.refTo(0)?.getPointer(this) + val result = cactus_stream_transcribe_process( + stream.toCPointer(), + pcmPtr?.reinterpret(), + pcmData?.size?.toULong() ?: 0u, + buffer, + 65536u + ) + if (result < 0) throw RuntimeException(cactus_get_last_error()?.toKString() ?: "Unknown error") + return buffer.toKString() + } +} + +@OptIn(ExperimentalForeignApi::class) +actual fun cactusStreamTranscribeStop(stream: Long): String { + memScoped { + val buffer = allocArray(65536) + val result = cactus_stream_transcribe_stop(stream.toCPointer(), buffer, 65536u) + if (result < 0) throw RuntimeException(cactus_get_last_error()?.toKString() ?: "Unknown error") + return buffer.toKString() + } +} + @OptIn(ExperimentalForeignApi::class) actual fun cactusEmbed(handle: Long, text: String, normalize: Boolean): FloatArray { memScoped { diff --git a/bindings/kotlin/Cactus.kt b/bindings/kotlin/Cactus.kt index afc4e2178..875076ade 100644 --- a/bindings/kotlin/Cactus.kt +++ b/bindings/kotlin/Cactus.kt @@ -14,6 +14,9 @@ object CactusJNI { @JvmStatic external fun nativeTokenize(handle: Long, text: String, tokenBuffer: IntArray, outTokenLen: LongArray): Int @JvmStatic external fun nativeScoreWindow(handle: Long, tokens: IntArray, start: Long, end: Long, context: Long, responseBuffer: ByteArray): Int @JvmStatic external fun nativeTranscribe(handle: Long, audioPath: String?, prompt: String, responseBuffer: ByteArray, optionsJson: String?, callback: CactusTokenCallback?, pcmData: ByteArray?): Int + @JvmStatic external fun nativeStreamTranscribeStart(handle: Long, optionsJson: String?): Long + @JvmStatic external fun nativeStreamTranscribeProcess(stream: Long, pcmData: ByteArray?, responseBuffer: ByteArray): Int + @JvmStatic external fun nativeStreamTranscribeStop(stream: Long, responseBuffer: ByteArray): Int @JvmStatic external fun nativeEmbed(handle: Long, text: String, embeddingsBuffer: FloatArray, outEmbeddingDim: LongArray, normalize: Boolean): Int @JvmStatic external fun nativeImageEmbed(handle: Long, imagePath: String, embeddingsBuffer: FloatArray, outEmbeddingDim: LongArray): Int @JvmStatic external fun nativeAudioEmbed(handle: Long, audioPath: String, embeddingsBuffer: FloatArray, outEmbeddingDim: LongArray): Int diff --git a/bindings/kotlin/README.md b/bindings/kotlin/README.md index a31c1cf55..b0301543b 100644 --- a/bindings/kotlin/README.md +++ b/bindings/kotlin/README.md @@ -27,6 +27,18 @@ CactusJNI.nativeDestroy(handle) ``` +Streaming transcription — push 16 kHz mono PCM16, read `{"success":true,"confirmed":...,"pending":...}` back each call: + +```kotlin +val stream = CactusJNI.nativeStreamTranscribeStart(handle, "{\"language\":\"en\"}") +val out = ByteArray(65536) +for (chunk in pcmChunks) { // each: 16 kHz mono PCM16 bytes + CactusJNI.nativeStreamTranscribeProcess(stream, chunk, out) + // parse confirmed/pending from String(out, 0, out.indexOf(0)) +} +CactusJNI.nativeStreamTranscribeStop(stream, out) +``` + ## Kotlin Multiplatform ``` @@ -64,5 +76,11 @@ kotlin { ```kotlin val model = cactusInit("/path/to/model", null, false) val result = cactusComplete(model, messagesJson, null, null, null, null) + +// streaming transcription: start, feed PCM16 chunks, stop +val stream = cactusStreamTranscribeStart(model, "{\"language\":\"en\"}") +val out = cactusStreamTranscribeProcess(stream, pcmChunk) // {"success":true,"confirmed":...,"pending":...} +val tail = cactusStreamTranscribeStop(stream) + cactusDestroy(model) ``` diff --git a/bindings/python/README.md b/bindings/python/README.md index d26e2d0c7..fbb1f5fa2 100644 --- a/bindings/python/README.md +++ b/bindings/python/README.md @@ -3,8 +3,8 @@ ctypes FFI to `cactus_engine.h`. The binding module lives at `/python/cactus/bindings/cactus.py`. -It loads `libcactus_engine.{so,dylib}` from `/python/cactus/bin/`, populated by -`cactus build --python`. +It loads `libcactus_engine.{so,dylib}` from `/python/cactus/bindings/lib/` (bundled) +or `cactus-engine/build/` (dev build), populated by `cactus build --python`. For installation, the high-level API, the CLI, the server, and the transpiler, see [`/python/README.md`](/python/). diff --git a/bindings/react-native/README.md b/bindings/react-native/README.md index b333d88b3..cd084b305 100644 --- a/bindings/react-native/README.md +++ b/bindings/react-native/README.md @@ -33,3 +33,16 @@ const result = await Cactus.complete(handle, messagesJson, null, null, null, fal await Cactus.destroy(handle); ``` + +## Streaming transcription + +Push 16 kHz mono PCM16 (base64) as it arrives; each call returns `{"success":true,"confirmed":...,"pending":...}` (`confirmed` is final, `pending` is the volatile tail). + +```ts +const stream = await Cactus.streamTranscribeStart(handle, '{"language":"en"}'); +for (const chunkBase64 of pcmChunks) { + const json = await Cactus.streamTranscribeProcess(stream, chunkBase64); + // parse json -> append "confirmed", show "pending" as a live preview +} +const tail = await Cactus.streamTranscribeStop(stream); +``` diff --git a/bindings/react-native/android/CactusModule.kt b/bindings/react-native/android/CactusModule.kt index d54834c50..534022141 100644 --- a/bindings/react-native/android/CactusModule.kt +++ b/bindings/react-native/android/CactusModule.kt @@ -226,6 +226,41 @@ class CactusModule(reactContext: ReactApplicationContext) : ReactContextBaseJava promise.resolve(decodeNullTerminatedUtf8(buffer)) } + @ReactMethod + fun streamTranscribeStart(handle: String, optionsJson: String?, promise: Promise) { + val nativeHandle = parseHandle(handle, promise) ?: return + val stream = CactusJNI.nativeStreamTranscribeStart(nativeHandle, optionsJson) + if (stream == 0L) { + fail(promise, "Failed to start streaming transcription") + return + } + promise.resolve(stream.toString()) + } + + @ReactMethod + fun streamTranscribeProcess(streamHandle: String, pcmDataBase64: String?, promise: Promise) { + val buffer = ByteArray(DEFAULT_BUFFER_SIZE) + val stream = parseHandle(streamHandle, promise) ?: return + val rc = CactusJNI.nativeStreamTranscribeProcess(stream, decodeBase64OrNull(pcmDataBase64), buffer) + if (rc < 0) { + fail(promise, "Stream transcription failed") + return + } + promise.resolve(decodeNullTerminatedUtf8(buffer)) + } + + @ReactMethod + fun streamTranscribeStop(streamHandle: String, promise: Promise) { + val buffer = ByteArray(DEFAULT_BUFFER_SIZE) + val stream = parseHandle(streamHandle, promise) ?: return + val rc = CactusJNI.nativeStreamTranscribeStop(stream, buffer) + if (rc < 0) { + fail(promise, "Stream transcription stop failed") + return + } + promise.resolve(decodeNullTerminatedUtf8(buffer)) + } + @ReactMethod fun embed(handle: String, text: String, normalize: Boolean, promise: Promise) { val nativeHandle = parseHandle(handle, promise) ?: return diff --git a/bindings/react-native/apple/Cactus.m b/bindings/react-native/apple/Cactus.m index 738c7e2af..cbc31f1a2 100644 --- a/bindings/react-native/apple/Cactus.m +++ b/bindings/react-native/apple/Cactus.m @@ -12,6 +12,9 @@ @interface RCT_EXTERN_MODULE(Cactus, RCTEventEmitter) RCT_EXTERN_METHOD(tokenize:(NSString *)handle text:(NSString *)text resolver:(RCTPromiseResolveBlock)resolve rejecter:(RCTPromiseRejectBlock)reject) RCT_EXTERN_METHOD(scoreWindow:(NSString *)handle tokens:(NSArray *)tokens start:(nonnull NSNumber *)start end:(nonnull NSNumber *)end context:(nonnull NSNumber *)context resolver:(RCTPromiseResolveBlock)resolve rejecter:(RCTPromiseRejectBlock)reject) RCT_EXTERN_METHOD(transcribe:(NSString *)handle audioPath:(NSString * _Nullable)audioPath prompt:(NSString * _Nullable)prompt optionsJson:(NSString * _Nullable)optionsJson pcmDataBase64:(NSString * _Nullable)pcmDataBase64 resolver:(RCTPromiseResolveBlock)resolve rejecter:(RCTPromiseRejectBlock)reject) +RCT_EXTERN_METHOD(streamTranscribeStart:(NSString *)handle optionsJson:(NSString * _Nullable)optionsJson resolver:(RCTPromiseResolveBlock)resolve rejecter:(RCTPromiseRejectBlock)reject) +RCT_EXTERN_METHOD(streamTranscribeProcess:(NSString *)streamHandle pcmDataBase64:(NSString * _Nullable)pcmDataBase64 resolver:(RCTPromiseResolveBlock)resolve rejecter:(RCTPromiseRejectBlock)reject) +RCT_EXTERN_METHOD(streamTranscribeStop:(NSString *)streamHandle resolver:(RCTPromiseResolveBlock)resolve rejecter:(RCTPromiseRejectBlock)reject) RCT_EXTERN_METHOD(embed:(NSString *)handle text:(NSString *)text normalize:(BOOL)normalize resolver:(RCTPromiseResolveBlock)resolve rejecter:(RCTPromiseRejectBlock)reject) RCT_EXTERN_METHOD(imageEmbed:(NSString *)handle imagePath:(NSString *)imagePath resolver:(RCTPromiseResolveBlock)resolve rejecter:(RCTPromiseRejectBlock)reject) RCT_EXTERN_METHOD(audioEmbed:(NSString *)handle audioPath:(NSString *)audioPath resolver:(RCTPromiseResolveBlock)resolve rejecter:(RCTPromiseRejectBlock)reject) diff --git a/bindings/react-native/apple/Cactus.swift b/bindings/react-native/apple/Cactus.swift index 5a43119e2..7a49191c3 100644 --- a/bindings/react-native/apple/Cactus.swift +++ b/bindings/react-native/apple/Cactus.swift @@ -224,6 +224,65 @@ final class Cactus: RCTEventEmitter { resolve(stringFromBuffer(buffer)) } + @objc(streamTranscribeStart:optionsJson:resolver:rejecter:) + func streamTranscribeStart( + _ handle: String, + optionsJson: String?, + resolver resolve: RCTPromiseResolveBlock, + rejecter reject: RCTPromiseRejectBlock + ) { + guard let nativeHandle = decodeHandle(handle) else { + self.reject(reject, "Invalid native handle") + return + } + guard let stream = cactus_stream_transcribe_start(nativeHandle, optionsJson) else { + self.reject(reject, "Failed to start streaming transcription") + return + } + resolve(encodeHandle(stream)) + } + + @objc(streamTranscribeProcess:pcmDataBase64:resolver:rejecter:) + func streamTranscribeProcess( + _ streamHandle: String, + pcmDataBase64: String?, + resolver resolve: RCTPromiseResolveBlock, + rejecter reject: RCTPromiseRejectBlock + ) { + guard let stream = decodeHandle(streamHandle) else { + self.reject(reject, "Invalid stream handle") + return + } + var buffer = [CChar](repeating: 0, count: defaultBufferSize) + let rc = decodeBase64(pcmDataBase64)?.withUnsafeBytes { rawBuffer in + cactus_stream_transcribe_process(stream, rawBuffer.bindMemory(to: UInt8.self).baseAddress, rawBuffer.count, &buffer, buffer.count) + } ?? cactus_stream_transcribe_process(stream, nil, 0, &buffer, buffer.count) + guard rc >= 0 else { + self.reject(reject, "Stream transcription failed") + return + } + resolve(stringFromBuffer(buffer)) + } + + @objc(streamTranscribeStop:resolver:rejecter:) + func streamTranscribeStop( + _ streamHandle: String, + resolver resolve: RCTPromiseResolveBlock, + rejecter reject: RCTPromiseRejectBlock + ) { + guard let stream = decodeHandle(streamHandle) else { + self.reject(reject, "Invalid stream handle") + return + } + var buffer = [CChar](repeating: 0, count: defaultBufferSize) + let rc = cactus_stream_transcribe_stop(stream, &buffer, buffer.count) + guard rc >= 0 else { + self.reject(reject, "Stream transcription stop failed") + return + } + resolve(stringFromBuffer(buffer)) + } + @objc(embed:text:normalize:resolver:rejecter:) func embed(_ handle: String, text: String, normalize: Bool, resolver resolve: RCTPromiseResolveBlock, rejecter reject: RCTPromiseRejectBlock) { guard let nativeHandle = decodeHandle(handle) else { diff --git a/bindings/react-native/index.ts b/bindings/react-native/index.ts index ad93cdf97..5c5c64d20 100644 --- a/bindings/react-native/index.ts +++ b/bindings/react-native/index.ts @@ -35,6 +35,12 @@ export interface CactusNativeModule { optionsJson: string | null, pcmDataBase64: string | null, ): Promise; + streamTranscribeStart(handle: string, optionsJson: string | null): Promise; + streamTranscribeProcess( + streamHandle: string, + pcmDataBase64: string | null, + ): Promise; + streamTranscribeStop(streamHandle: string): Promise; embed(handle: string, text: string, normalize: boolean): Promise; imageEmbed(handle: string, imagePath: string): Promise; audioEmbed(handle: string, audioPath: string): Promise; diff --git a/bindings/rust/README.md b/bindings/rust/README.md index 96d44952f..f264dd51f 100644 --- a/bindings/rust/README.md +++ b/bindings/rust/README.md @@ -47,3 +47,21 @@ fn main() { } } ``` + +## Streaming transcription + +Push 16 kHz mono PCM16 as it arrives; each call returns `{"success":true,"confirmed":...,"pending":...}` (`confirmed` is final, `pending` is the volatile tail). All `unsafe`, like the rest: + +```rust +let opts = CString::new(r#"{"language":"en"}"#).unwrap(); +let stream = cactus::cactus_stream_transcribe_start(model, opts.as_ptr()); +let mut buf = vec![0u8; 65536]; +for chunk in pcm_chunks { // each: 16 kHz mono i16 bytes + cactus::cactus_stream_transcribe_process( + stream, chunk.as_ptr(), chunk.len(), + buf.as_mut_ptr() as *mut c_char, buf.len(), + ); + // append "confirmed"; show "pending" as a live preview +} +cactus::cactus_stream_transcribe_stop(stream, buf.as_mut_ptr() as *mut c_char, buf.len()); +``` diff --git a/bindings/rust/cactus.rs b/bindings/rust/cactus.rs index d30adfa13..e126d6c00 100644 --- a/bindings/rust/cactus.rs +++ b/bindings/rust/cactus.rs @@ -2,6 +2,7 @@ use std::os::raw::{c_char, c_float, c_int, c_void}; pub type CactusModelT = *mut c_void; pub type CactusIndexT = *mut c_void; +pub type CactusStreamTranscribeT = *mut c_void; pub type CactusTokenCallback = Option; pub type CactusLogCallback = Option; @@ -24,6 +25,9 @@ unsafe extern "C" { pub fn cactus_tokenize(model: CactusModelT, text: *const c_char, token_buffer: *mut u32, token_buffer_len: usize, out_token_len: *mut usize) -> c_int; pub fn cactus_score_window(model: CactusModelT, tokens: *const u32, token_len: usize, start: usize, end: usize, context: usize, response_buffer: *mut c_char, buffer_size: usize) -> c_int; pub fn cactus_transcribe(model: CactusModelT, audio_file_path: *const c_char, prompt: *const c_char, response_buffer: *mut c_char, buffer_size: usize, options_json: *const c_char, callback: CactusTokenCallback, user_data: *mut c_void, pcm_buffer: *const u8, pcm_buffer_size: usize) -> c_int; + pub fn cactus_stream_transcribe_start(model: CactusModelT, options_json: *const c_char) -> CactusStreamTranscribeT; + pub fn cactus_stream_transcribe_process(stream: CactusStreamTranscribeT, pcm_buffer: *const u8, pcm_buffer_size: usize, response_buffer: *mut c_char, buffer_size: usize) -> c_int; + pub fn cactus_stream_transcribe_stop(stream: CactusStreamTranscribeT, response_buffer: *mut c_char, buffer_size: usize) -> c_int; pub fn cactus_embed(model: CactusModelT, text: *const c_char, embeddings_buffer: *mut c_float, buffer_size: usize, embedding_dim: *mut usize, normalize: bool) -> c_int; pub fn cactus_image_embed(model: CactusModelT, image_path: *const c_char, embeddings_buffer: *mut c_float, buffer_size: usize, embedding_dim: *mut usize) -> c_int; diff --git a/bindings/swift/README.md b/bindings/swift/README.md index 94d826240..0cd20bbf9 100644 --- a/bindings/swift/README.md +++ b/bindings/swift/README.md @@ -29,3 +29,19 @@ let response = String(cString: buf) cactus_destroy(model) ``` + +### Streaming transcription + +The streaming functions are exposed the same way. Push 16 kHz mono PCM16 as it arrives and read back `{"success":true,"confirmed":...,"pending":...}` — `confirmed` words are final, `pending` is the volatile tail. + +```swift +let stream = cactus_stream_transcribe_start(model, "{\"language\":\"en\"}") +var buf = [CChar](repeating: 0, count: 65536) +for chunk in pcmChunks { // each chunk: 16 kHz mono 16-bit PCM + chunk.withUnsafeBytes { raw in + _ = cactus_stream_transcribe_process(stream, raw.bindMemory(to: UInt8.self).baseAddress, raw.count, &buf, buf.count) + } + // parse String(cString: buf) -> append "confirmed", show "pending" live +} +cactus_stream_transcribe_stop(stream, &buf, buf.count) +``` diff --git a/blog/hybrid_transcription.md b/blog/hybrid_transcription.md index 87dc15806..973bf22ac 100644 --- a/blog/hybrid_transcription.md +++ b/blog/hybrid_transcription.md @@ -73,7 +73,7 @@ Check out the open-source engine and documentation on GitHub. ## See Also -- [Cactus Engine API Reference](/docs/cactus_engine.md) — Full transcription API docs (`cactus_transcribe`) +- [Cactus Engine API Reference](/docs/cactus_engine.md) — Transcription API: one-shot `cactus_transcribe` and streaming `cactus_stream_transcribe_*` - [Python Binding](/python/) — Python bindings - [Swift Binding](/bindings/swift/) — Swift API - [Kotlin Binding](/bindings/kotlin/) — Kotlin API diff --git a/blog/lfm2_24b_a2b.md b/blog/lfm2_24b_a2b.md index 2fde03658..098889abf 100644 --- a/blog/lfm2_24b_a2b.md +++ b/blog/lfm2_24b_a2b.md @@ -105,6 +105,7 @@ Ready to run LFM2-24B locally on your Mac? Here's how to get up and running. ```bash git clone https://github.com/cactus-compute/cactus.git cd cactus +source ./setup # Build the Cactus engine (shared library for Python FFI) cactus build --python @@ -134,14 +135,13 @@ For building your own applications and agents, use the Python FFI bindings direc ```python import json -from cactus import cactus_init, cactus_complete, cactus_reset, cactus_destroy +from cactus import get_bundle_dir, cactus_init, cactus_complete, cactus_reset, cactus_destroy -# Load the model -model = cactus_init("weights/lfm2-24b-a2b", None, False) +model = cactus_init(str(get_bundle_dir("LiquidAI/LFM2-24B-A2B")), None, False) # Simple chat completion messages = json.dumps([{"role": "user", "content": "Write a Python function to sort a list"}]) -response = json.loads(cactus_complete(model, messages, None, None, None)) +response = cactus_complete(model, messages, None, None, None) print(response["response"]) # Generated text print(f"{response['decode_tps']:.1f} tokens/sec") @@ -180,7 +180,7 @@ tools = [ ] messages = json.dumps([{"role": "user", "content": "Calculate the factorial of 10"}]) -response = json.loads(cactus_complete(model, messages, None, json.dumps(tools), None)) +response = cactus_complete(model, messages, None, json.dumps(tools), None) if response["function_calls"]: print(response["function_calls"]) # Model's tool invocation @@ -192,7 +192,7 @@ Cactus measures model confidence during generation. When the model isn't confide ```python options = json.dumps({"confidence_threshold": 0.7}) -response = json.loads(cactus_complete(model, messages, options, None, None)) +response = cactus_complete(model, messages, options, None, None) if response["cloud_handoff"]: # Route to cloud API for this query diff --git a/blog/parakeet.md b/blog/parakeet.md index e62b52a37..508c453fa 100644 --- a/blog/parakeet.md +++ b/blog/parakeet.md @@ -158,9 +158,9 @@ cactus download nvidia/parakeet-ctc-1.1b For integrating Parakeet into your own applications, use the Python FFI bindings directly: ```python -from cactus import cactus_init, cactus_transcribe, cactus_destroy +from cactus import get_bundle_dir, cactus_init, cactus_transcribe, cactus_destroy -model = cactus_init("weights/parakeet-ctc-1.1b", None, False) +model = cactus_init(str(get_bundle_dir("nvidia/parakeet-ctc-1.1b")), None, False) result = cactus_transcribe(model, "/path/to/audio.wav") @@ -206,17 +206,19 @@ use std::ffi::CString; use std::os::raw::c_char; use std::ptr; +mod cactus; + fn main() { let model_path = CString::new("weights/parakeet-ctc-1.1b").unwrap(); let audio_path = CString::new("audio.wav").unwrap(); let model = unsafe { - cactus_sys::cactus_init(model_path.as_ptr(), ptr::null(), false) + cactus::cactus_init(model_path.as_ptr(), ptr::null(), false) }; let mut buf = vec![0u8; 16384]; let rc = unsafe { - cactus_sys::cactus_transcribe( + cactus::cactus_transcribe( model, audio_path.as_ptr(), ptr::null(), @@ -232,7 +234,7 @@ fn main() { println!("Transcript: {}", response); } - unsafe { cactus_sys::cactus_destroy(model) }; + unsafe { cactus::cactus_destroy(model) }; } ``` @@ -241,13 +243,15 @@ fn main() { The Swift binding exposes top-level functions that map directly to the C FFI: ```swift -import Foundation +import cactus -let model = try cactusInit("weights/parakeet-ctc-1.1b", nil, false) -let resultJson = try cactusTranscribe(model, "/path/to/audio.wav", nil, nil, nil, nil) +let model = cactus_init("weights/parakeet-ctc-1.1b", nil, false) +var buf = [CChar](repeating: 0, count: 65536) +cactus_transcribe(model, "/path/to/audio.wav", nil, &buf, buf.count, nil, nil, nil, nil, 0) +let resultJson = String(cString: buf) print(resultJson) -cactusDestroy(model) +cactus_destroy(model) ``` ### 8. Use the [Kotlin Binding](/bindings/kotlin/) @@ -258,7 +262,7 @@ The Kotlin binding exposes top-level functions that map directly to the C FFI: import com.cactus.* val model = cactusInit("weights/parakeet-ctc-1.1b", null, false) -val resultJson = cactusTranscribe(model, "/path/to/audio.wav", null, null, null, null) +val resultJson = cactusTranscribe(model, "/path/to/audio.wav", "", null, null, null) println(resultJson) cactusDestroy(model) diff --git a/cactus-engine/CMakeLists.txt b/cactus-engine/CMakeLists.txt index 24c7c8ca4..7e7cc348e 100644 --- a/cactus-engine/CMakeLists.txt +++ b/cactus-engine/CMakeLists.txt @@ -29,6 +29,7 @@ set(ENGINE_SOURCES src/embed.cpp src/complete.cpp src/transcribe.cpp + src/stream.cpp src/tokenizer.cpp ) if(APPLE) diff --git a/cactus-engine/cactus_engine.h b/cactus-engine/cactus_engine.h index 395757c4d..376628563 100644 --- a/cactus-engine/cactus_engine.h +++ b/cactus-engine/cactus_engine.h @@ -21,6 +21,7 @@ extern "C" { typedef void* cactus_model_t; typedef void* cactus_index_t; +typedef void* cactus_stream_transcribe_t; typedef void (*cactus_token_callback)(const char* token, uint32_t token_id, void* user_data); CACTUS_FFI_EXPORT cactus_model_t cactus_init( @@ -107,6 +108,25 @@ CACTUS_FFI_EXPORT int cactus_transcribe( size_t pcm_buffer_size ); +CACTUS_FFI_EXPORT cactus_stream_transcribe_t cactus_stream_transcribe_start( + cactus_model_t model, + const char* options_json +); + +CACTUS_FFI_EXPORT int cactus_stream_transcribe_process( + cactus_stream_transcribe_t stream, + const uint8_t* pcm_buffer, + size_t pcm_buffer_size, + char* response_buffer, + size_t buffer_size +); + +CACTUS_FFI_EXPORT int cactus_stream_transcribe_stop( + cactus_stream_transcribe_t stream, + char* response_buffer, + size_t buffer_size +); + CACTUS_FFI_EXPORT int cactus_embed( cactus_model_t model, const char* text, diff --git a/cactus-engine/src/engine.h b/cactus-engine/src/engine.h index 4302c96f3..604f730f5 100644 --- a/cactus-engine/src/engine.h +++ b/cactus-engine/src/engine.h @@ -661,12 +661,27 @@ class Model { float min_p = 0.15f, float repetition_penalty = 1.1f, float* out_token_time_start = nullptr, float* out_token_time_end = nullptr); - std::vector transcribe_parakeet_tdt(const std::vector& audio_features); + struct ParakeetTdtStreamState { + bool initialized = false; + uint32_t last_token = 0; + size_t time_index = 0; + std::vector> dec_state; + std::vector pending; + float confirmed_sec = 0.0f; + size_t decoded_tokens = 0; + double raw_decode_ms = 0.0; + }; + + std::vector transcribe_parakeet_tdt(const std::vector& audio_features, + ParakeetTdtStreamState* stream = nullptr, + bool is_final = true, + size_t end_frame = 0); std::vector transcribe_whisper_seq2seq(const std::vector& audio_features, const std::vector& decoder_prompt_tokens, size_t max_tokens, const std::vector>& stop_token_sequences, - const std::atomic* should_stop = nullptr); + const std::atomic* should_stop = nullptr, + int64_t suppress_token_id = -1); std::vector get_embeddings(const std::vector& tokens, bool pooled = true, bool normalize = false, const std::string& profile_file = ""); @@ -912,6 +927,7 @@ class Model { ToolCallConstrainer tool_constrainer_; std::unordered_map vocab_bias_; + int64_t suppressed_token_id_ = -1; bool handoff_probe_loaded_ = false; uint32_t handoff_probe_feat_dim_ = 0; diff --git a/cactus-engine/src/model.cpp b/cactus-engine/src/model.cpp index a2f8ed89f..a955b871f 100644 --- a/cactus-engine/src/model.cpp +++ b/cactus-engine/src/model.cpp @@ -1754,6 +1754,7 @@ uint32_t Model::argmax_component_logits(Component& comp, size_t logit_row, float }; float second_v = -std::numeric_limits::infinity(); auto observe_logit = [&](size_t i, float v) { + if (static_cast(i) == suppressed_token_id_) return; v = score_with_bias(i, v); if (v > best_v) { second_v = best_v; @@ -2859,13 +2860,16 @@ std::vector Model::transcribe_whisper_seq2seq( const std::vector& decoder_prompt_tokens, size_t max_tokens, const std::vector>& stop_token_sequences, - const std::atomic* should_stop) { + const std::atomic* should_stop, + int64_t suppress_token_id) { if (decoder_prompt_tokens.empty() || max_tokens == 0) return {}; if (decode_route_ != DecodeRoute::ENCODER_CROSS_KV_STEP || encoder_cross_kv_source_kind_ != "audio_features") { CACTUS_LOG_ERROR("model", "Whisper bundle missing encoder_cross_kv_decoder_step route metadata"); return {}; } if (!prepare_encoder_cross_kv_from_audio(audio_features)) return {}; + struct SuppressGuard { int64_t& id; ~SuppressGuard() { id = -1; } } guard{suppressed_token_id_}; + suppressed_token_id_ = suppress_token_id; return run_encoder_cross_kv_decode_loop( decoder_prompt_tokens, max_tokens, @@ -2873,8 +2877,11 @@ std::vector Model::transcribe_whisper_seq2seq( should_stop); } -std::vector Model::transcribe_parakeet_tdt(const std::vector& audio_features) { +std::vector Model::transcribe_parakeet_tdt(const std::vector& audio_features, + ParakeetTdtStreamState* stream, bool is_final, + size_t end_frame) { std::vector emitted; + double raw_decode_ms = 0.0; Component* audio_enc = components_.count("audio_encoder") ? &components_.at("audio_encoder") : nullptr; Component* dec = components_.count("decoder") ? &components_.at("decoder") : nullptr; @@ -2991,10 +2998,21 @@ std::vector Model::transcribe_parakeet_tdt(const std::vector& a auto& buf = dec->input_buffers[idx]; std::memset(buf.data(), 0, buf.size()); }; - zero_state("state_h_0"); - zero_state("state_c_0"); - zero_state("state_h_1"); - zero_state("state_c_1"); + if (stream && stream->initialized && stream->dec_state.size() == 4) { + const char* sn[4] = {"state_h_0", "state_c_0", "state_h_1", "state_c_1"}; + for (size_t i = 0; i < 4; ++i) { + int idx = input_index(*dec, sn[i]); + if (idx < 0) continue; + auto& buf = dec->input_buffers[idx]; + std::memcpy(buf.data(), stream->dec_state[i].data(), + std::min(buf.size(), stream->dec_state[i].size())); + } + } else { + zero_state("state_h_0"); + zero_state("state_c_0"); + zero_state("state_h_1"); + zero_state("state_c_1"); + } std::vector durations = config_.tdt_durations; if (durations.empty()) { @@ -3003,8 +3021,8 @@ std::vector Model::transcribe_parakeet_tdt(const std::vector& a if (durations.empty()) durations.push_back(1); const uint32_t configured_blank = config_.tdt_blank_id; - uint32_t last_token = configured_blank; - size_t time_index = 0; + uint32_t last_token = (stream && stream->initialized) ? stream->last_token : configured_blank; + size_t time_index = (stream && stream->initialized) ? stream->time_index : 0; const int ef_idx = input_index(*dec, "encoder_frame"); const int tok_in_idx = input_index(*dec, "token_ids"); @@ -3046,7 +3064,35 @@ std::vector Model::transcribe_parakeet_tdt(const std::vector& a }; } - while (time_index < T) { + size_t commit_to = T; + if (stream) { + size_t valid_hidden = T; + if (!used_npu && expected_frames > 0) + valid_hidden = std::min(T, (copy_frames * T) / expected_frames); + commit_to = (end_frame > 0) ? std::min(end_frame, valid_hidden) : valid_hidden; + } + Tokenizer* stream_tok = stream ? get_tokenizer() : nullptr; + const auto& tdt_vocab_bias = get_vocab_bias(); + const float frame_sec = (160.0f / 16000.0f) * + static_cast(std::max(1, config_.subsampling_factor)); + constexpr uint32_t kMaxStreamDurationSkipFrames = 2; + + auto snapshot_state = [&]() { + std::vector> snap(state_copy_count); + for (size_t s = 0; s < state_copy_count; ++s) { + const uint8_t* p = static_cast(state_copies[s].in_data); + snap[s].assign(p, p + state_copies[s].bytes); + } + return snap; + }; + + std::vector> snap_state; + uint32_t snap_last_token = last_token; + size_t snap_time = time_index; + size_t confirmed_count = 0; + if (stream) snap_state = snapshot_state(); + + while (time_index < commit_to) { const uint8_t* frame_ptr = hidden_ptr + time_index * frame_bytes; write_typed_buffer(ef_buf, ef_desc.precision, frame_ptr, frame_bytes, hidden_precision); @@ -3059,7 +3105,10 @@ std::vector Model::transcribe_parakeet_tdt(const std::vector& a case Precision::INT8: *reinterpret_cast(tok_data) = static_cast(last_token); break; default: *reinterpret_cast(tok_data) = static_cast(last_token); break; } + const auto exec_t0 = std::chrono::steady_clock::now(); dec->graph->execute(); + raw_decode_ms += std::chrono::duration( + std::chrono::steady_clock::now() - exec_t0).count(); const void* logits_ptr = dec->graph->get_output(logits_node); auto get_logit = [&](size_t i) -> float { @@ -3072,6 +3121,10 @@ std::vector Model::transcribe_parakeet_tdt(const std::vector& a float best_token_score = -std::numeric_limits::infinity(); for (size_t i = 0; i < token_class_count; ++i) { float v = get_logit(i); + if (stream && !tdt_vocab_bias.empty()) { + auto it = tdt_vocab_bias.find(static_cast(i)); + if (it != tdt_vocab_bias.end()) v += it->second; + } if (v > best_token_score) { best_token_score = v; next_token = static_cast(i); } } uint32_t best_duration_idx = 0; @@ -3081,9 +3134,19 @@ std::vector Model::transcribe_parakeet_tdt(const std::vector& a if (v > best_duration_score) { best_duration_score = v; best_duration_idx = static_cast(i); } } - const uint32_t skip = durations[std::min(best_duration_idx, static_cast(durations.size() - 1))]; + uint32_t skip = durations[std::min(best_duration_idx, static_cast(durations.size() - 1))]; + if (stream && skip > kMaxStreamDurationSkipFrames) skip = kMaxStreamDurationSkipFrames; if (next_token != effective_blank) { + if (stream && !is_final && stream_tok && !emitted.empty()) { + std::string piece = stream_tok->decode({next_token}); + if (!piece.empty() && piece[0] == ' ') { + snap_state = snapshot_state(); + snap_last_token = last_token; + snap_time = time_index; + confirmed_count = emitted.size(); + } + } emitted.push_back(next_token); last_token = next_token; for (size_t s = 0; s < state_copy_count; ++s) { @@ -3108,6 +3171,26 @@ std::vector Model::transcribe_parakeet_tdt(const std::vector& a if (!advanced) time_index += 1; } + if (stream) { + stream->initialized = true; + stream->decoded_tokens = emitted.size(); + stream->raw_decode_ms = raw_decode_ms; + stream->pending.clear(); + if (is_final) { + stream->dec_state = snapshot_state(); + stream->last_token = last_token; + stream->time_index = time_index; + stream->confirmed_sec = static_cast(time_index) * frame_sec; + } else { + stream->dec_state = std::move(snap_state); + stream->last_token = snap_last_token; + stream->time_index = snap_time; + stream->confirmed_sec = static_cast(snap_time) * frame_sec; + stream->pending.assign(emitted.begin() + confirmed_count, emitted.end()); + emitted.resize(confirmed_count); + } + } + return emitted; } diff --git a/cactus-engine/src/stream.cpp b/cactus-engine/src/stream.cpp new file mode 100644 index 000000000..182d81e4e --- /dev/null +++ b/cactus-engine/src/stream.cpp @@ -0,0 +1,407 @@ +#include "../cactus_engine.h" +#include "utils.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace cactus::ffi; + +namespace { + +constexpr size_t kScratchSize = 1u << 16; +constexpr size_t kSampleRate = 16000; +constexpr float kSampleRateF = 16000.0f; +constexpr size_t kLeftContextSamples = 8 * kSampleRate; // left context re-fed each window +constexpr size_t kRightContextSamples = 1 * kSampleRate; // look-ahead before confirming +constexpr size_t kChunkSamples = 1 * kSampleRate; // warm decode step +constexpr size_t kColdStartSamples = 6 * kSampleRate; // first decode from empty state +constexpr size_t kResumeColdSamples = 2 * kSampleRate; // first decode after silence +constexpr size_t kSilenceResetSamples = 3 * kSampleRate; // silence run that triggers a cold restart +constexpr size_t kMaxDecodeSamples = 10 * kSampleRate; + +struct StreamStats { + size_t decode_tokens = 0; + double total_time_ms = 0.0; + double decode_tps = 0.0; + double time_to_first_token_ms = 0.0; + + void finalize() { + if (total_time_ms > 0.0) decode_tps = decode_tokens * 1000.0 / total_time_ms; + } +}; + +struct StreamTranscribe { + CactusModelHandle* model = nullptr; + std::string options_json; + bool is_parakeet = false; + + std::vector samples; + size_t samples_decoded_up_to = 0; + size_t silence_run = 0; + bool cold_restart = false; + + cactus::engine::Model::ParakeetTdtStreamState pstate; + std::vector committed_tokens; + std::string emitted_text; + std::string previous_pending; + + std::vector whisper_prev_segments; + size_t whisper_last_len = 0; +}; + +double json_num(const std::string& json, const std::string& key) { + float v = 0.0f; + return try_parse_json_float(json, key, v) ? static_cast(v) : 0.0; +} + +std::vector to_pcm16(const std::vector& samples) { + std::vector pcm(samples.size()); + for (size_t i = 0; i < samples.size(); ++i) { + float x = std::max(-32768.0f, std::min(32767.0f, samples[i] * 32768.0f)); + pcm[i] = static_cast(x); + } + return pcm; +} + +std::string with_timestamps_option(std::string opts) { + if (opts.find("\"timestamps\"") != std::string::npos) return opts; + if (opts.empty() || opts == "{}") return "{\"timestamps\":true}"; + return opts.substr(0, opts.find_last_of('}')) + ",\"timestamps\":true}"; +} + +std::string trim(const std::string& s) { + size_t b = s.find_first_not_of(' '), e = s.find_last_not_of(' '); + return b == std::string::npos ? std::string() : s.substr(b, e - b + 1); +} + +std::string strip_annotations(const std::string& text) { + static const std::regex pattern(R"(\([^)]*\)|\[[^\]]*\]|\.\.\.)"); + std::string out = std::regex_replace(text, pattern, ""); + out = std::regex_replace(out, std::regex(R"(\s+)"), " "); + return trim(out); +} + +std::vector parse_segments(const std::string& json) { + std::vector segs; + for (const std::string& obj : split_json_array(json_array_field(json, "segments"))) { + float start = 0.0f, end = 0.0f; + try_parse_json_float(obj, "start", start); + try_parse_json_float(obj, "end", end); + segs.push_back({start, end, strip_annotations(json_string_field(obj, "text"))}); + } + return segs; +} + +std::vector whisper_transcribe(StreamTranscribe* s, const std::vector& samples, + StreamStats& stats) { + std::vector pcm = to_pcm16(samples); + std::string scratch(kScratchSize, '\0'); + const std::string opts = with_timestamps_option(s->options_json); + const bool prev_should_stop = s->model->should_stop.load(); + const int rc = cactus_transcribe( + static_cast(s->model), nullptr, nullptr, + scratch.data(), scratch.size(), opts.c_str(), + nullptr, nullptr, + reinterpret_cast(pcm.data()), pcm.size() * sizeof(int16_t)); + if (prev_should_stop) s->model->should_stop.store(true); + if (rc <= 0) return {}; + const std::string json(scratch.c_str()); + stats.decode_tps = json_num(json, "decode_tps"); + stats.total_time_ms = json_num(json, "total_time_ms"); + stats.time_to_first_token_ms = json_num(json, "time_to_first_token_ms"); + stats.decode_tokens = static_cast(json_num(json, "decode_tokens")); + return parse_segments(json); +} + +std::vector window_features(std::vector window, size_t mel_bins) { + if (window.empty()) return {}; + auto cfg = cactus::audio::get_parakeet_spectrogram_config(); + const size_t waveform_samples = window.size(); + cactus::audio::apply_preemphasis(window, 0.97f); + std::vector features = cactus::audio::compute_spectrogram_graph( + window, cfg, mel_bins, 0.0f, 8000.0f, cactus::audio::WHISPER_SAMPLE_RATE, 0, 0); + cactus::audio::normalize_parakeet_log_mel(features, mel_bins); + size_t valid_frames = waveform_samples / cfg.hop_length; + if (valid_frames == 0) valid_frames = 1; + cactus::audio::trim_mel_frames(features, mel_bins, valid_frames); + return features; +} + +std::string parakeet_decode_window(StreamTranscribe* s, size_t window_start, size_t window_end, + size_t decode_start_frame, size_t decode_end_frame, + bool is_final, std::string* pending_text, StreamStats& stats) { + if (pending_text) pending_text->clear(); + auto* model = s->model->model.get(); + const size_t mel_bins = std::max(1, static_cast(model->get_config().num_mel_bins)); + std::vector features = window_features( + std::vector(s->samples.begin() + window_start, s->samples.begin() + window_end), mel_bins); + if (features.empty()) return ""; + + s->pstate.time_index = decode_start_frame; + const auto t0 = std::chrono::steady_clock::now(); + std::vector tokens = model->transcribe_parakeet_tdt( + features, &s->pstate, is_final, is_final ? 0 : decode_end_frame); + stats.total_time_ms += std::chrono::duration(std::chrono::steady_clock::now() - t0).count(); + stats.decode_tokens += s->pstate.decoded_tokens; + + auto* tokenizer = model->get_tokenizer(); + if (!tokenizer) return ""; + + s->committed_tokens.insert(s->committed_tokens.end(), tokens.begin(), tokens.end()); + std::string full = tokenizer->decode(s->committed_tokens); + std::string delta = full.size() > s->emitted_text.size() ? full.substr(s->emitted_text.size()) : std::string(); + s->emitted_text = full; + + if (pending_text && !s->pstate.pending.empty()) { + std::vector combined = s->committed_tokens; + combined.insert(combined.end(), s->pstate.pending.begin(), s->pstate.pending.end()); + std::string with_pending = tokenizer->decode(combined); + if (with_pending.size() > full.size()) *pending_text = with_pending.substr(full.size()); + } + + if (!is_final && !tokens.empty() && s->pstate.confirmed_sec > 0.0f) { + s->samples_decoded_up_to = window_start + static_cast(s->pstate.confirmed_sec * kSampleRateF); + } + return delta; +} + +size_t parakeet_spf(StreamTranscribe* s) { + const uint32_t subsampling = std::max(1, s->model->model->get_config().subsampling_factor); + return cactus::audio::get_parakeet_spectrogram_config().hop_length * subsampling; +} + +std::string parakeet_process(StreamTranscribe* s, std::string& pending, StreamStats& stats) { + std::lock_guard lock(s->model->model_mutex); + const size_t spf = parakeet_spf(s); + + std::string confirmed; + for (;;) { + const size_t total = s->samples.size(); + const size_t decodable = total > kRightContextSamples ? total - kRightContextSamples : 0; + const bool cold = s->samples_decoded_up_to == 0 || s->cold_restart; + const size_t cold_min = s->cold_restart ? kResumeColdSamples : kColdStartSamples; + const size_t min_chunk = cold ? cold_min : kChunkSamples; + if (decodable <= s->samples_decoded_up_to || + decodable - s->samples_decoded_up_to < min_chunk) { + break; + } + const size_t decode_to = std::min(decodable, s->samples_decoded_up_to + kMaxDecodeSamples); + const size_t window_start = cold ? s->samples_decoded_up_to + : (s->samples_decoded_up_to > kLeftContextSamples ? s->samples_decoded_up_to - kLeftContextSamples : 0); + const size_t window_end = decode_to + kRightContextSamples; + const size_t decode_start_frame = (s->samples_decoded_up_to - window_start) / spf; + const size_t decode_end_frame = decode_start_frame + (decode_to - s->samples_decoded_up_to) / spf; + if (cold) s->pstate = {}; + + std::string pend; + const size_t prev_cursor = s->samples_decoded_up_to; + const size_t tokens_before = stats.decode_tokens; + confirmed += parakeet_decode_window(s, window_start, window_end, + decode_start_frame, decode_end_frame, false, &pend, stats); + pending = pend; + if (s->pstate.decoded_tokens > 0) { s->cold_restart = false; s->silence_run = 0; } + if (s->samples_decoded_up_to > prev_cursor) continue; + if (s->pstate.decoded_tokens == 0) { + s->silence_run += decode_to - s->samples_decoded_up_to; + s->samples_decoded_up_to = decode_to; + if (s->silence_run >= kSilenceResetSamples) s->cold_restart = true; + continue; + } + if (decode_to - s->samples_decoded_up_to < cold_min) break; + stats.decode_tokens = tokens_before; + confirmed += parakeet_decode_window(s, window_start, window_end, + decode_start_frame, decode_end_frame, true, &pend, stats); + pending = pend; + s->samples_decoded_up_to = decode_to; + } + + const size_t keep_from = s->samples_decoded_up_to > kLeftContextSamples + ? s->samples_decoded_up_to - kLeftContextSamples : 0; + if (keep_from > kLeftContextSamples) { + s->samples.erase(s->samples.begin(), s->samples.begin() + keep_from); + s->samples_decoded_up_to -= keep_from; + } + + stats.finalize(); + if (confirmed.empty() && pending.empty()) pending = s->previous_pending; + else s->previous_pending = pending; + return confirmed; +} + +std::string parakeet_flush(StreamTranscribe* s, StreamStats& stats) { + std::lock_guard lock(s->model->model_mutex); + const size_t total = s->samples.size(); + if (total <= s->samples_decoded_up_to) return ""; + const size_t spf = parakeet_spf(s); + const bool cold = s->samples_decoded_up_to == 0 || s->cold_restart; + if (cold) s->pstate = {}; + const size_t window_start = cold + ? s->samples_decoded_up_to + : (s->samples_decoded_up_to > kLeftContextSamples ? s->samples_decoded_up_to - kLeftContextSamples : 0); + const size_t decode_start_frame = (s->samples_decoded_up_to - window_start) / spf; + std::string confirmed = parakeet_decode_window(s, window_start, total, decode_start_frame, 0, true, nullptr, stats); + stats.finalize(); + return confirmed; +} + +std::string segments_text(const std::vector& segs, size_t from, size_t to) { + std::string out; + for (size_t i = from; i < to && i < segs.size(); ++i) { + if (segs[i].text.empty()) continue; + if (!out.empty()) out += ' '; + out += segs[i].text; + } + return out; +} + +std::string whisper_process(StreamTranscribe* s, std::string& pending, StreamStats& stats) { + if (s->samples.size() < s->whisper_last_len + kChunkSamples) { + pending = s->previous_pending; + return ""; + } + s->whisper_last_len = s->samples.size(); + std::vector segs = whisper_transcribe(s, s->samples, stats); + + size_t agree = 0; + float confirmed_end_sec = 0.0f; + while (agree < segs.size() && agree < s->whisper_prev_segments.size() && + segs[agree].text == s->whisper_prev_segments[agree].text) { + confirmed_end_sec = std::min(segs[agree].end, s->whisper_prev_segments[agree].end); + ++agree; + } + + std::string confirmed; + pending = segments_text(segs, agree, segs.size()); + if (agree > 0 && confirmed_end_sec > 0.0f) { + confirmed = segments_text(segs, 0, agree); + const size_t cut = std::min(static_cast(confirmed_end_sec * kSampleRateF), s->samples.size()); + s->samples.erase(s->samples.begin(), s->samples.begin() + cut); + s->whisper_last_len -= cut; + std::vector tail; + for (size_t i = agree; i < segs.size(); ++i) + tail.push_back({segs[i].start - confirmed_end_sec, segs[i].end - confirmed_end_sec, segs[i].text}); + s->whisper_prev_segments = std::move(tail); + } else { + s->whisper_prev_segments = segs; + } + + s->previous_pending = pending; + return confirmed; +} + +std::string whisper_flush(StreamTranscribe* s, StreamStats& stats) { + std::vector segs = whisper_transcribe(s, s->samples, stats); + return segments_text(segs, 0, segs.size()); +} + +int write_result(char* buffer, size_t size, const std::string& confirmed, + const std::string& pending, const StreamStats& stats) { + std::ostringstream os; + os << "{\"success\":true,\"confirmed\":\"" << escape_json_string(confirmed) + << "\",\"pending\":\"" << escape_json_string(pending) + << "\",\"decode_tps\":" << stats.decode_tps + << ",\"total_time_ms\":" << stats.total_time_ms + << ",\"time_to_first_token_ms\":" << stats.time_to_first_token_ms + << ",\"decode_tokens\":" << stats.decode_tokens << "}"; + const std::string json = os.str(); + if (!buffer || size == 0) return 0; + if (json.size() >= size) { + handle_error_response("Stream response buffer too small", buffer, size); + return -1; + } + std::memcpy(buffer, json.c_str(), json.size() + 1); + return static_cast(json.size()); +} + +} // namespace + +extern "C" { + +cactus_stream_transcribe_t cactus_stream_transcribe_start(cactus_model_t model, const char* options_json) { + if (!model) { + last_error_message = "stream_transcribe_start: model is null"; + CACTUS_LOG_ERROR("stream_transcribe_start", last_error_message); + return nullptr; + } + try { + auto* handle = static_cast(model); + const auto type = handle->model->get_config().model_type; + const bool is_whisper = type == cactus::engine::Config::ModelType::WHISPER; + const bool is_parakeet = type == cactus::engine::Config::ModelType::PARAKEET_TDT; + if (!is_whisper && !is_parakeet) { + last_error_message = "stream_transcribe_start: only Whisper and Parakeet models support streaming"; + CACTUS_LOG_ERROR("stream_transcribe_start", last_error_message); + return nullptr; + } + auto s = std::make_unique(); + s->model = handle; + s->is_parakeet = is_parakeet; + if (options_json && options_json[0] != '\0') s->options_json = options_json; + CACTUS_LOG_INFO("stream_transcribe_start", "streaming session opened"); + return static_cast(s.release()); + } catch (const std::exception& e) { + last_error_message = std::string("stream_transcribe_start: ") + e.what(); + CACTUS_LOG_ERROR("stream_transcribe_start", last_error_message); + return nullptr; + } +} + +int cactus_stream_transcribe_process(cactus_stream_transcribe_t stream, + const uint8_t* pcm_buffer, size_t pcm_buffer_size, + char* response_buffer, size_t buffer_size) { + if (!stream) { + last_error_message = "stream_transcribe_process: stream is null"; + CACTUS_LOG_ERROR("stream_transcribe_process", last_error_message); + handle_error_response(last_error_message, response_buffer, buffer_size); + return -1; + } + auto* s = static_cast(stream); + try { + if (pcm_buffer && pcm_buffer_size >= sizeof(int16_t)) { + std::vector news = cactus::audio::pcm_buffer_to_float_samples(pcm_buffer, pcm_buffer_size); + s->samples.insert(s->samples.end(), news.begin(), news.end()); + } + StreamStats stats; + std::string pending; + std::string confirmed = s->is_parakeet ? parakeet_process(s, pending, stats) + : whisper_process(s, pending, stats); + return write_result(response_buffer, buffer_size, confirmed, pending, stats); + } catch (const std::exception& e) { + last_error_message = std::string("stream_transcribe_process: ") + e.what(); + CACTUS_LOG_ERROR("stream_transcribe_process", last_error_message); + handle_error_response(last_error_message, response_buffer, buffer_size); + return -1; + } +} + +int cactus_stream_transcribe_stop(cactus_stream_transcribe_t stream, + char* response_buffer, size_t buffer_size) { + if (!stream) { + last_error_message = "stream_transcribe_stop: stream is null"; + CACTUS_LOG_ERROR("stream_transcribe_stop", last_error_message); + handle_error_response(last_error_message, response_buffer, buffer_size); + return -1; + } + auto* s = static_cast(stream); + int result = 0; + try { + StreamStats stats; + std::string confirmed = s->is_parakeet ? parakeet_flush(s, stats) : whisper_flush(s, stats); + result = write_result(response_buffer, buffer_size, confirmed, "", stats); + } catch (const std::exception& e) { + last_error_message = std::string("stream_transcribe_stop: ") + e.what(); + CACTUS_LOG_ERROR("stream_transcribe_stop", last_error_message); + handle_error_response(last_error_message, response_buffer, buffer_size); + result = -1; + } + delete s; + return result; +} + +} // extern "C" diff --git a/cactus-engine/src/transcribe.cpp b/cactus-engine/src/transcribe.cpp index 9a650b91c..9a3c7ec84 100644 --- a/cactus-engine/src/transcribe.cpp +++ b/cactus-engine/src/transcribe.cpp @@ -10,6 +10,46 @@ using namespace cactus::ffi; +namespace { + +constexpr float kWhisperTimestampStepSec = 0.02f; + +uint32_t whisper_timestamp_begin(cactus::engine::Tokenizer* tokenizer) { + return tokenizer->encode("<|notimestamps|>")[0] + 1; +} + +std::vector parse_whisper_timestamp_segments( + cactus::engine::Tokenizer* tokenizer, const std::vector& tokens, + uint32_t timestamp_begin, std::string& clean_text) { + std::vector segments; + std::vector text_tokens; + std::vector flat; + float seg_start = -1.0f; + auto flush = [&](float end_sec) { + if (!text_tokens.empty()) { + std::string text = tokenizer->decode(text_tokens); + if (!text.empty() && text[0] == ' ') text.erase(0, 1); + segments.push_back({seg_start, end_sec, text}); + } + text_tokens.clear(); + }; + for (uint32_t tok : tokens) { + if (tok >= timestamp_begin) { + const float t = static_cast(tok - timestamp_begin) * kWhisperTimestampStepSec; + if (seg_start < 0.0f) seg_start = t; + else { flush(t); seg_start = t; } + } else { + text_tokens.push_back(tok); + flat.push_back(tok); + } + } + flush(seg_start); + clean_text = tokenizer->decode(flat); + return segments; +} + +} // namespace + extern "C" { int cactus_preprocess_audio_features( @@ -231,11 +271,31 @@ int cactus_transcribe( return -1; } + const bool want_timestamps = is_whisper && options.timestamps; + const uint32_t ts_begin = want_timestamps ? whisper_timestamp_begin(tokenizer) : 0; + std::vector tokens; if (prompt && prompt[0] != '\0') { tokens = tokenizer->encode(prompt); } else if (is_whisper) { tokens = handle->model->get_config().decoder_prompt_token_ids; + std::string language = json_string_field(options_json ? options_json : "", "language"); + if (!language.empty()) { + std::vector language_token = tokenizer->encode("<|" + language + "|>"); + if (language_token.size() == 1) { + for (uint32_t& token : tokens) { + std::string piece = tokenizer->decode({token}); + if (piece.size() == 6 && piece[0] == '<' && piece[1] == '|' && + piece[4] == '|' && piece[5] == '>' && + std::islower((unsigned char)piece[2]) && std::islower((unsigned char)piece[3])) { + token = language_token[0]; + break; + } + } + } + } + if (want_timestamps) + tokens.erase(std::remove(tokens.begin(), tokens.end(), ts_begin - 1), tokens.end()); } if (tokens.empty() && is_whisper) { @@ -267,6 +327,7 @@ int cactus_transcribe( } std::string final_text; + std::vector segments; std::vector generated_tokens; generated_tokens.reserve(options.max_tokens); const size_t prompt_tokens = tokens.size(); @@ -291,7 +352,8 @@ int cactus_transcribe( tokens, options.max_tokens, stop_token_sequences, - &handle->should_stop); + &handle->should_stop, + want_timestamps ? static_cast(ts_begin) - 1 : -1); auto t_first = std::chrono::high_resolution_clock::now(); time_to_first_token = std::chrono::duration_cast(t_first - start_time).count() / 1000.0; @@ -302,9 +364,14 @@ int cactus_transcribe( break; } } - final_text = tokenizer->decode(generated_tokens); + if (want_timestamps) { + segments = parse_whisper_timestamp_segments(tokenizer, generated_tokens, ts_begin, final_text); + } else { + final_text = tokenizer->decode(generated_tokens); + } if (callback) { for (uint32_t tok : generated_tokens) { + if (want_timestamps && tok >= ts_begin) continue; std::string piece = tokenizer->decode({tok}); callback(piece.c_str(), tok, user_data); } @@ -369,7 +436,7 @@ int cactus_transcribe( std::string json = construct_response_json( final_text, {}, time_to_first_token, total_time_ms, prefill_tps, decode_tps, prompt_tokens, completion_tokens, - confidence); + confidence, false, "", segments); if (json.size() >= buffer_size) { handle_error_response("Response buffer too small", response_buffer, buffer_size); return -1; diff --git a/cactus-engine/src/utils.h b/cactus-engine/src/utils.h index f9c0b5085..95e2368da 100644 --- a/cactus-engine/src/utils.h +++ b/cactus-engine/src/utils.h @@ -517,6 +517,7 @@ struct InferenceOptions { bool force_tools = false; bool include_stop_sequences = false; bool use_vad = true; + bool timestamps = false; bool telemetry_enabled = true; bool auto_handoff = true; bool handoff_with_images = true; @@ -619,9 +620,20 @@ inline std::string json_array_field(const std::string& json, const std::string& int depth = 1; size_t end = start + 1; + bool in_string = false, escape = false; while (end < json.size() && depth > 0) { - if (json[end] == '[') depth++; - else if (json[end] == ']') depth--; + const char c = json[end]; + if (in_string) { + if (escape) escape = false; + else if (c == '\\') escape = true; + else if (c == '"') in_string = false; + } else if (c == '"') { + in_string = true; + } else if (c == '[') { + depth++; + } else if (c == ']') { + depth--; + } end++; } return json.substr(start, end - start); @@ -1460,6 +1472,13 @@ inline InferenceOptions parse_inference_options_json(const std::string& json) { options.use_vad = (json.substr(pos, 4) == "true"); } + pos = json.find("\"timestamps\""); + if (pos != std::string::npos) { + pos = json.find(':', pos) + 1; + while (pos < json.length() && std::isspace(static_cast(json[pos]))) pos++; + options.timestamps = (json.substr(pos, 4) == "true"); + } + pos = json.find("\"telemetry_enabled\""); if (pos != std::string::npos) { pos = json.find(':', pos) + 1; @@ -1643,8 +1662,8 @@ inline void partition_thinking_response(const std::string& input, std::string& t } struct TranscriptSegment { - float start; - float end; + float start = 0.0f; + float end = 0.0f; std::string text; }; diff --git a/cactus-engine/tests/CMakeLists.txt b/cactus-engine/tests/CMakeLists.txt index 804fa319c..e3dd2800d 100644 --- a/cactus-engine/tests/CMakeLists.txt +++ b/cactus-engine/tests/CMakeLists.txt @@ -46,3 +46,12 @@ endif() add_executable(transcribe transcribe.cpp) target_link_libraries(transcribe PRIVATE cactus_engine) +if(SDL2_FOUND) + target_compile_definitions(transcribe PRIVATE HAVE_SDL2) + if(TARGET SDL2::SDL2) + target_link_libraries(transcribe PRIVATE SDL2::SDL2) + else() + target_link_libraries(transcribe PRIVATE ${SDL2_LIBRARY}) + target_include_directories(transcribe PRIVATE ${SDL2_INCLUDE_DIR}) + endif() +endif() diff --git a/cactus-engine/tests/test_stream.cpp b/cactus-engine/tests/test_stream.cpp new file mode 100644 index 000000000..3b83b3711 --- /dev/null +++ b/cactus-engine/tests/test_stream.cpp @@ -0,0 +1,150 @@ +#include "test_utils.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace EngineTestUtils; + +static const char* g_model = std::getenv("CACTUS_TEST_TRANSCRIPTION_MODEL"); +static const char* g_assets = std::getenv("CACTUS_TEST_ASSETS"); + +static std::vector words_of(const std::string& text) { + std::vector words; + std::istringstream iss(text); + std::string w; + while (iss >> w) { + std::string n; + for (char c : w) if (std::isalnum((unsigned char)c)) n += (char)std::tolower((unsigned char)c); + if (!n.empty()) words.push_back(n); + } + return words; +} + +static double recall(const std::vector& ref, const std::vector& hyp) { + if (ref.empty()) return 1.0; + size_t found = 0; + for (const auto& w : ref) if (std::find(hyp.begin(), hyp.end(), w) != hyp.end()) ++found; + return (double)found / (double)ref.size(); +} + +static std::vector load_wav(const std::string& path) { + FILE* f = fopen(path.c_str(), "rb"); + if (!f) return {}; + char tag[4]; + uint32_t sz; + if (fread(tag, 1, 4, f) != 4 || std::strncmp(tag, "RIFF", 4)) { fclose(f); return {}; } + fread(&sz, 4, 1, f); + if (fread(tag, 1, 4, f) != 4 || std::strncmp(tag, "WAVE", 4)) { fclose(f); return {}; } + uint16_t ch = 1, bits = 16, fmt = 0; + uint32_t rate = 16000; + bool have_fmt = false; + std::vector mono; + while (fread(tag, 1, 4, f) == 4 && fread(&sz, 4, 1, f) == 1) { + if (!std::strncmp(tag, "fmt ", 4)) { + uint16_t af, c, al, bps; + uint32_t sr, br; + fread(&af, 2, 1, f); fread(&c, 2, 1, f); fread(&sr, 4, 1, f); + fread(&br, 4, 1, f); fread(&al, 2, 1, f); fread(&bps, 2, 1, f); + fmt = af; ch = c ? c : 1; bits = bps; rate = sr; have_fmt = true; + if (sz > 16) fseek(f, (long)sz - 16, SEEK_CUR); + } else if (!std::strncmp(tag, "data", 4)) { + if (!have_fmt || fmt != 1 || bits != 16) { fclose(f); return {}; } + size_t frames = sz / (2u * ch); + std::vector raw(sz / 2); + fread(raw.data(), 2, raw.size(), f); + mono.resize(frames); + for (size_t i = 0; i < frames; ++i) { + int acc = 0; + for (uint16_t k = 0; k < ch; ++k) acc += raw[i * ch + k]; + mono[i] = (int16_t)(acc / ch); + } + break; + } else { + fseek(f, (long)(sz + (sz & 1)), SEEK_CUR); + } + } + fclose(f); + if (mono.empty() || rate == 16000) return mono; + double ratio = 16000.0 / (double)rate; + size_t on = (size_t)(mono.size() * ratio); + std::vector out(on); + for (size_t i = 0; i < on; ++i) { + double pos = (double)i / ratio; + size_t i0 = (size_t)pos; + double fr = pos - (double)i0; + int16_t a = mono[std::min(i0, mono.size() - 1)]; + int16_t b = mono[std::min(i0 + 1, mono.size() - 1)]; + out[i] = (int16_t)((double)a + ((double)b - (double)a) * fr); + } + return out; +} + +static std::string transcribe_oneshot(cactus_model_t model, const std::vector& pcm) { + char resp[1 << 16] = {0}; + int rc = cactus_transcribe(model, nullptr, nullptr, resp, sizeof(resp), + R"({"telemetry_enabled": false, "auto_handoff": false, "timestamps": true})", + nullptr, nullptr, + reinterpret_cast(pcm.data()), pcm.size() * sizeof(int16_t)); + return rc <= 0 ? "" : json_string(std::string(resp), "response"); +} + +static bool test_stream_matches_oneshot() { + std::cout << "\n=== STREAM vs ONE-SHOT ===\n"; + if (!g_model || !g_assets) { std::cout << "SKIP (model/assets not set)\n"; return true; } + cactus_model_t model = cactus_init(g_model, nullptr, false); + if (!model) { std::cerr << "[x] model init failed\n"; return false; } + + std::vector pcm = load_wav(std::string(g_assets) + "/test.wav"); + if (pcm.size() > 20 * 16000) pcm.resize(20 * 16000); + auto golden = words_of(transcribe_oneshot(model, pcm)); + if (golden.size() < 4) { cactus_destroy(model); std::cout << " SKIP (test.wav has no usable speech)\n"; return true; } + + cactus_stream_transcribe_t stream = cactus_stream_transcribe_start(model, nullptr); + if (!stream) { cactus_destroy(model); std::cerr << "[x] stream start failed\n"; return false; } + std::string streamed; + std::vector resp(1 << 16); + auto append = [&]() { + std::string c = json_string(std::string(resp.data()), "confirmed"); + if (c.empty()) return; + if (!streamed.empty() && streamed.back() != ' ' && c.front() != ' ') streamed += ' '; + streamed += c; + }; + for (size_t off = 0; off < pcm.size(); off += 16000) { + size_t n = std::min(16000, pcm.size() - off); + resp[0] = '\0'; + if (cactus_stream_transcribe_process(stream, reinterpret_cast(pcm.data() + off), + n * sizeof(int16_t), resp.data(), resp.size()) < 0) { + cactus_stream_transcribe_stop(stream, nullptr, 0); + cactus_destroy(model); + std::cerr << "[x] process failed\n"; + return false; + } + append(); + } + resp[0] = '\0'; + cactus_stream_transcribe_stop(stream, resp.data(), resp.size()); + append(); + cactus_destroy(model); + + auto sw = words_of(streamed); + double rc = recall(golden, sw), pr = recall(sw, golden); + bool ok = rc >= 0.85 && pr >= 0.85; + std::cout << " recall=" << std::fixed << std::setprecision(3) << rc << " precision=" << pr + << " " << (ok ? "OK" : "FAIL") << "\n Status: " << (ok ? "PASSED" : "FAILED") << "\n"; + return ok; +} + +int main() { + TestUtils::TestRunner runner("Stream Tests"); + runner.run_test("stream_matches_oneshot", test_stream_matches_oneshot()); + runner.print_summary(); + return runner.all_passed() ? 0 : 1; +} diff --git a/cactus-engine/tests/transcribe.cpp b/cactus-engine/tests/transcribe.cpp index 3772be1d3..84d22b0e9 100644 --- a/cactus-engine/tests/transcribe.cpp +++ b/cactus-engine/tests/transcribe.cpp @@ -1,82 +1,406 @@ #include "../cactus_engine.h" -#include -#include -#include #include -#include #include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#ifdef HAVE_SDL2 +#include +#include +#include +#endif + +constexpr size_t RESPONSE_BUFFER_SIZE = 65536; + +namespace ansi { +constexpr const char* reset = "\033[0m"; +constexpr const char* bold = "\033[1m"; +constexpr const char* dim = "\033[2m"; +constexpr const char* green = "\033[32m"; +constexpr const char* yellow = "\033[33m"; +constexpr const char* cyan = "\033[36m"; +} + +static bool stdout_is_terminal() { + return isatty(STDOUT_FILENO) != 0 && std::getenv("NO_COLOR") == nullptr; +} +static const bool g_color = stdout_is_terminal(); + +static std::string colored(const std::string& text, const char* color) { + return g_color ? color + text + ansi::reset : text; +} + +static int terminal_width() { + struct winsize w; + if (ioctl(STDOUT_FILENO, TIOCGWINSZ, &w) == -1 || w.ws_col == 0) return 80; + return w.ws_col; +} + +static void print_rule() { + std::string rule(terminal_width(), '-'); + std::cout << colored(rule, ansi::dim) << "\n"; +} + +static void print_banner() { + if (!g_color) { + std::cout << "Cactus Transcription\n\n"; + return; + } + std::cout << ansi::bold << ansi::green + << " ██████╗ █████╗ ██████╗████████╗██╗ ██╗███████╗\n" + << "██╔════╝██╔══██╗██╔════╝╚══██╔══╝██║ ██║██╔════╝\n" + << "██║ ███████║██║ ██║ ██║ ██║███████╗\n" + << "██║ ██╔══██║██║ ██║ ██║ ██║╚════██║\n" + << "╚██████╗██║ ██║╚██████╗ ██║ ╚██████╔╝███████║\n" + << " ╚═════╝╚═╝ ╚═╝ ╚═════╝ ╚═╝ ╚═════╝ ╚══════╝\n" + << ansi::reset << ansi::dim << " Transcription" << ansi::reset << "\n\n"; +} + +static std::string json_string_value(const std::string& json, const std::string& key) { + std::string pattern = "\"" + key + "\":\""; + size_t start = json.find(pattern); + if (start == std::string::npos) return ""; + start += pattern.length(); + size_t end = start; + while (end < json.length() && json[end] != '"') { + if (json[end] == '\\' && end + 1 < json.length()) end++; + end++; + } + return json.substr(start, end - start); +} + +static double json_number_value(const std::string& json, const std::string& key) { + std::string pattern = "\"" + key + "\":"; + size_t start = json.find(pattern); + if (start == std::string::npos) return 0.0; + start += pattern.length(); + while (start < json.length() && std::isspace((unsigned char)json[start])) start++; + return std::strtod(json.c_str() + start, nullptr); +} + +static std::string whisper_prompt(const std::string& model_path, const std::string& language) { + std::string p = model_path; + std::transform(p.begin(), p.end(), p.begin(), [](unsigned char c) { return (char)std::tolower(c); }); + if (p.find("whisper") != std::string::npos) + return "<|startoftranscript|><|" + language + "|><|transcribe|>"; + return ""; +} + +static void print_token(const char* token, uint32_t, void*) { + std::cout << token << std::flush; +} + +static int transcribe_file(cactus_model_t model, const std::string& audio_path, + const std::string& model_path, const std::string& language) { + std::string prompt = whisper_prompt(model_path, language); + std::vector response_buffer(RESPONSE_BUFFER_SIZE, 0); + std::string options = "{\"max_tokens\":500,\"auto_handoff\":false,\"language\":\"" + language + "\"}"; + + auto start = std::chrono::steady_clock::now(); + int rc = cactus_transcribe( + model, audio_path.c_str(), prompt.empty() ? nullptr : prompt.c_str(), + response_buffer.data(), response_buffer.size(), + options.c_str(), print_token, nullptr, nullptr, 0); + double total_s = std::chrono::duration(std::chrono::steady_clock::now() - start).count(); + if (rc < 0) { + std::cerr << "\nTranscription failed: " << cactus_get_last_error() << "\n"; + return -1; + } + + std::string json(response_buffer.data()); + double model_ms = json_number_value(json, "total_time_ms"); + std::ostringstream stats; + stats << std::fixed << std::setprecision(2) << "[processed: " << total_s << "s"; + if (model_ms > 0.0) stats << " | model: " << model_ms / 1000.0 << "s"; + stats << "]"; + std::cout << "\n\n" << colored(stats.str(), ansi::dim) << "\n"; + return 0; +} + +#ifdef HAVE_SDL2 + +constexpr int TARGET_SAMPLE_RATE = 16000; +constexpr int AUDIO_BUFFER_MS = 100; +constexpr int PROCESS_INTERVAL_MS = 250; + +struct AudioState { + std::mutex mutex; + std::vector buffer; + std::atomic recording{false}; + int actual_sample_rate{TARGET_SAMPLE_RATE}; +}; +static AudioState g_audio; + +static std::vector resample_to_16k(const std::vector& input, int source_rate) { + if (source_rate == TARGET_SAMPLE_RATE) return input; + size_t num_input = input.size() / 2; + const int16_t* in = reinterpret_cast(input.data()); + double ratio = static_cast(TARGET_SAMPLE_RATE) / source_rate; + size_t num_output = static_cast(num_input * ratio); + std::vector out(num_output); + for (size_t i = 0; i < num_output; i++) { + double src = i / ratio; + size_t i0 = static_cast(src); + size_t i1 = std::min(i0 + 1, num_input - 1); + double frac = src - i0; + out[i] = static_cast(in[i0] * (1.0 - frac) + in[i1] * frac); + } + std::vector result(num_output * 2); + std::memcpy(result.data(), out.data(), result.size()); + return result; +} + +static void audio_callback(void*, Uint8* stream, int len) { + if (!g_audio.recording) return; + std::lock_guard lock(g_audio.mutex); + g_audio.buffer.insert(g_audio.buffer.end(), stream, stream + len); +} + +static size_t wrap_index(const std::string& s, size_t limit) { + size_t len = 0; + bool in_esc = false; + for (size_t i = 0; i < s.length(); ++i) { + char c = s[i]; + if (c == '\033') { in_esc = true; continue; } + if (in_esc) { if (c == 'm') in_esc = false; continue; } + if ((c & 0xC0) != 0x80) len++; + if (len >= limit && c == ' ') return i; + } + return std::string::npos; +} + +static int run_live_transcription(cactus_model_t model, const std::string& language) { + if (SDL_Init(SDL_INIT_AUDIO) < 0) { + std::cerr << "Failed to init SDL audio: " << SDL_GetError() << "\n"; + return 1; + } + + SDL_AudioSpec want, have; + SDL_zero(want); + want.freq = TARGET_SAMPLE_RATE; + want.format = AUDIO_S16LSB; + want.channels = 1; + want.samples = (TARGET_SAMPLE_RATE * AUDIO_BUFFER_MS) / 1000; + want.callback = audio_callback; + SDL_AudioDeviceID device = SDL_OpenAudioDevice(nullptr, 1, &want, &have, SDL_AUDIO_ALLOW_FREQUENCY_CHANGE); + if (device == 0) { + std::cerr << "Failed to open microphone: " << SDL_GetError() << "\n"; + SDL_Quit(); + return 1; + } + g_audio.actual_sample_rate = have.freq; + + std::string options = "{\"language\":\"" + language + "\"}"; + cactus_stream_transcribe_t stream = cactus_stream_transcribe_start(model, options.c_str()); + if (!stream) { + std::cerr << "Failed to start streaming (live mode needs a Whisper or Parakeet TDT model): " + << cactus_get_last_error() << "\n"; + SDL_CloseAudioDevice(device); + SDL_Quit(); + return 1; + } + + std::cout << colored("Listening…", ansi::yellow) << " press " << colored("Enter", ansi::cyan) << " to stop\n"; + print_rule(); + std::cout << "\n"; + + g_audio.buffer.clear(); + g_audio.recording = true; + SDL_PauseAudioDevice(device, 0); + + std::atomic should_stop{false}; + std::thread input_thread([&should_stop]() { + std::string line; + std::getline(std::cin, line); + should_stop = true; + }); + + std::string confirmed_text; + std::string current_line; + int pending_lines = 0; + std::string stats; + std::vector response_buffer(RESPONSE_BUFFER_SIZE, 0); + auto last_process = std::chrono::steady_clock::now(); + + auto step = [&](std::vector chunk, bool render) { + if (chunk.empty()) return; + std::vector pcm = resample_to_16k(chunk, g_audio.actual_sample_rate); + response_buffer[0] = '\0'; + auto t0 = std::chrono::steady_clock::now(); + if (cactus_stream_transcribe_process(stream, pcm.data(), pcm.size(), + response_buffer.data(), response_buffer.size()) < 0) return; + + std::string json(response_buffer.data()); + std::string confirmed = json_string_value(json, "confirmed"); + std::string pending = json_string_value(json, "pending"); + confirmed_text += confirmed; + if (!render) return; + + double latency_ms = std::chrono::duration(std::chrono::steady_clock::now() - t0).count(); + if (!confirmed.empty() || !pending.empty()) { + std::ostringstream st; + st << "[latency: " << (int)latency_ms << "ms]"; + stats = colored(st.str(), ansi::dim); + } + + int limit = (int)(terminal_width() * 0.7); + + if (pending_lines > 0) { + std::cout << "\r\033[2K"; + for (int i = 0; i < pending_lines; ++i) std::cout << "\033[1A\033[2K"; + } else { + std::cout << "\r"; + } + + if (!confirmed.empty()) current_line += colored(confirmed, ansi::green); + + while (true) { + size_t idx = wrap_index(current_line, (size_t)limit); + if (idx == std::string::npos) break; + std::cout << "\r\033[K" << current_line.substr(0, idx) << ansi::reset << "\n"; + current_line = std::string(ansi::green) + current_line.substr(idx + 1); + } + std::cout << "\r\033[K" << current_line; + + std::string ghost = stats; + if (!pending.empty()) { + if (!ghost.empty()) ghost += "\n"; + ghost += colored("[pending] " + pending, ansi::yellow); + } + pending_lines = 0; + if (!ghost.empty()) { + std::cout << "\n"; + std::stringstream ss(ghost); + std::string line; + bool first = true; + while (std::getline(ss, line)) { + while (true) { + size_t idx = wrap_index(line, (size_t)limit); + if (idx == std::string::npos) break; + if (!first) std::cout << "\n"; + std::cout << line.substr(0, idx); + line = line.substr(idx + 1); + pending_lines++; + first = false; + } + if (!first) std::cout << "\n"; + std::cout << line; + pending_lines++; + first = false; + } + } + std::cout << std::flush; + }; + + while (!should_stop) { + auto now = std::chrono::steady_clock::now(); + if (now - last_process >= std::chrono::milliseconds(PROCESS_INTERVAL_MS)) { + last_process = now; + std::vector chunk; + { std::lock_guard lock(g_audio.mutex); chunk.swap(g_audio.buffer); } + step(std::move(chunk), true); + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + + g_audio.recording = false; + SDL_PauseAudioDevice(device, 1); + std::vector tail; + { std::lock_guard lock(g_audio.mutex); tail.swap(g_audio.buffer); } + step(std::move(tail), false); + + response_buffer[0] = '\0'; + cactus_stream_transcribe_stop(stream, response_buffer.data(), response_buffer.size()); + confirmed_text += json_string_value(std::string(response_buffer.data()), "confirmed"); -static void print_usage(const char* prog) { - std::cerr << "Usage: " << prog - << " [audio_file.wav] [--language ]\n"; + std::cout << "\n\n"; + print_rule(); + std::cout << colored("Final transcript:", ansi::bold) << "\n" << confirmed_text << "\n"; + print_rule(); + + if (input_thread.joinable()) input_thread.detach(); + SDL_CloseAudioDevice(device); + SDL_Quit(); + return 0; +} + +#else + +static int run_live_transcription(cactus_model_t, const std::string&) { + std::cerr << "Live microphone transcription requires SDL2. Install SDL2 and rebuild, " + "or pass an audio file (brew install sdl2 / apt-get install libsdl2-dev).\n"; + return 1; +} + +#endif + +static void print_usage(const char* argv0) { + std::cerr << "Usage: " << argv0 << " [audio_file] [--language ]\n" + << " With an audio file: one-shot transcription. Without: live from the microphone.\n"; } int main(int argc, char** argv) { - if (argc < 2) { + std::cout << std::unitbuf; + + if (argc < 2 || std::string(argv[1]) == "-h" || std::string(argv[1]) == "--help") { print_usage(argv[0]); - return 2; + return argc < 2 ? 1 : 0; } std::string model_path = argv[1]; - std::string audio_path; + std::string audio_file; std::string language = "en"; - for (int i = 2; i < argc; ++i) { std::string arg = argv[i]; - if (arg == "--language" && i + 1 < argc) { - language = argv[++i]; - } else if (arg == "-h" || arg == "--help") { + if (arg == "-h" || arg == "--help") { print_usage(argv[0]); return 0; - } else if (audio_path.empty() && arg.rfind("-", 0) != 0) { - audio_path = arg; + } else if (arg == "--language") { + if (i + 1 >= argc) { + std::cerr << "Error: --language requires a value\n"; + return 1; + } + language = argv[++i]; + } else if (!arg.empty() && arg[0] == '-') { + std::cerr << "Error: unknown option '" << arg << "'\n"; + print_usage(argv[0]); + return 1; + } else { + audio_file = arg; } } - if (audio_path.empty()) { - std::cerr << "Error: no audio file path provided.\n"; - print_usage(argv[0]); - return 2; - } - std::cout << "Loading model from " << model_path << "...\n"; cactus_model_t model = cactus_init(model_path.c_str(), nullptr, false); if (!model) { - std::cerr << "Failed to initialize model from " << model_path << "\n"; - const char* err = cactus_get_last_error(); - if (err && *err) std::cerr << " " << err << "\n"; + std::cerr << "Failed to initialize model: " << cactus_get_last_error() << "\n"; return 1; } - std::cout << "Model loaded.\n"; - - std::ostringstream opts; - opts << "{\"language\":\"" << language << "\",\"telemetry_enabled\":false,\"auto_handoff\":false,\"max_tokens\":512}"; - constexpr size_t kBufSize = 1 << 16; - std::string response(kBufSize, '\0'); + if (g_color) std::cout << "\033[2J\033[3J\033[H"; + print_banner(); - int rc = cactus_transcribe( - model, - audio_path.c_str(), - nullptr, - response.data(), - response.size(), - opts.str().c_str(), - nullptr, - nullptr, - nullptr, - 0); - - if (rc <= 0) { - std::cerr << "Transcription failed: " << response.c_str() << "\n"; - cactus_destroy(model); - return 1; + int result; + if (!audio_file.empty()) { + std::cout << colored("Transcribing ", ansi::bold) << audio_file << "\n"; + print_rule(); + std::cout << "\n"; + result = transcribe_file(model, audio_file, model_path, language); + } else { + result = run_live_transcription(model, language); } - response.resize(std::strlen(response.c_str())); - std::cout << response << std::endl; - cactus_destroy(model); - return 0; + std::cout << "\nGoodbye.\n"; + return result >= 0 ? 0 : 1; } diff --git a/docs/cactus_engine.md b/docs/cactus_engine.md index b00bea62a..0b8e851e8 100644 --- a/docs/cactus_engine.md +++ b/docs/cactus_engine.md @@ -25,9 +25,9 @@ cactus auth `cactus download` fetches a **pre-built runtime bundle** (CQ weights + serialized graph + manifest) from [huggingface.co/Cactus-Compute](https://huggingface.co/Cactus-Compute) into -`transpiled/-cq[-]/`. Defaults to the generic CPU -bundle; pass `--platform apple` for the Apple Silicon variant. The result -can be loaded directly via `cactus_init()`. +`weights/-cq[-]/`. Defaults to `--platform auto` (the +best build for your host, e.g. Apple Silicon on macOS); pass `--platform cpu` +for the generic build. The result can be loaded directly via `cactus_init()`. For models not on Cactus-Compute, build a bundle from source with `cactus convert ` followed by `cactus transpile `. @@ -191,6 +191,7 @@ int cactus_complete( "function_calls": [], "segments": [], "confidence": 0.85, + "confidence_threshold": 0.7, "time_to_first_token_ms": 150.5, "total_time_ms": 1250.3, "prefill_tps": 166.1, @@ -202,6 +203,8 @@ int cactus_complete( } ``` +`confidence_threshold` is the resolved value `confidence` is compared against — model-dependent (see the options table above), or whatever you pass; the `0.7` here is just the fallback default. `cloud_handoff` becomes `true` when `confidence` drops below it. + The `thinking` field is only present in the JSON when the model produced a chain-of-thought block: ```json { @@ -213,6 +216,7 @@ The `thinking` field is only present in the JSON when the model produced a chain "function_calls": [], "segments": [], "confidence": 0.91, + "confidence_threshold": 0.7, "time_to_first_token_ms": 150.5, "total_time_ms": 1250.3, "prefill_tps": 166.1, @@ -234,6 +238,7 @@ The `thinking` field is only present in the JSON when the model produced a chain "function_calls": [], "segments": [], "confidence": 0.18, + "confidence_threshold": 0.7, "time_to_first_token_ms": 45.2, "total_time_ms": 45.2, "prefill_tps": 619.5, @@ -245,7 +250,7 @@ The `thinking` field is only present in the JSON when the model produced a chain } ``` -When `cloud_handoff` is true, the model's confidence dropped below `confidence_threshold` (default: 0.7) and the response was fulfilled by a cloud-based model. The `response` field contains the cloud-provided answer. +When `cloud_handoff` is true, the model's confidence dropped below the resolved `confidence_threshold` (see the request options above) and the response was fulfilled by a cloud-based model. The `response` field contains the cloud-provided answer. **Error Response:** ```json @@ -255,7 +260,7 @@ When `cloud_handoff` is true, the model's confidence dropped below `confidence_t "cloud_handoff": false, "response": null, "function_calls": [], - "confidence": 0.0, + "confidence": null, "time_to_first_token_ms": 0.0, "total_time_ms": 0.0, "prefill_tps": 0.0, @@ -284,6 +289,7 @@ Note: `ram_usage_mb` reflects actual current RAM usage even in error responses. ], "segments": [], "confidence": 0.92, + "confidence_threshold": 0.7, "time_to_first_token_ms": 120.0, "total_time_ms": 450.5, "prefill_tps": 375.0, @@ -426,6 +432,34 @@ if (result == 0) { } ``` +### `cactus_render_prompt` +Renders the chat-templated prompt string for the given messages without running inference. Useful for debugging prompt formatting, estimating token budgets, or feeding the rendered text to another tool. + +```c +int cactus_render_prompt( + cactus_model_t model, // Model handle + const char* messages_json, // JSON array of messages (same format as cactus_complete) + const char* options_json, // Optional generation options (can be NULL) + const char* tools_json, // Optional tools definition (can be NULL) + char* prompt_buffer, // Buffer for the rendered prompt text + size_t buffer_size // Size of prompt_buffer +); +``` + +**Returns:** The rendered prompt length in bytes (excluding the null terminator) on success; -1 on invalid parameters or rendering error; -2 if the buffer is too small to hold the rendered prompt and its null terminator. + +**Note:** The output is plain prompt text, not JSON. + +**Example:** +```c +const char* messages = "[{\"role\": \"user\", \"content\": \"Hello!\"}]"; +char prompt[4096]; +int len = cactus_render_prompt(model, messages, NULL, NULL, prompt, sizeof(prompt)); +if (len >= 0) { + printf("Rendered prompt:\n%s\n", prompt); +} +``` + ### `cactus_score_window` Scores a window of tokens for perplexity calculation or token probability analysis. @@ -513,13 +547,17 @@ int cactus_transcribe( **Options Format:** ```json { - "max_tokens": 448 + "max_tokens": 448, + "language": "en", + "timestamps": true } ``` | Option | Type | Default | Description | |--------|------|---------|-------------| | `max_tokens` | int | auto | Maximum tokens to generate. When unset, it defaults to the larger of 100 and an audio-length estimate (`audio_sec × 20` for Whisper, `audio_sec × 30` for Parakeet). For Whisper the result is then capped so the prompt tokens plus generated tokens fit the decoder's 448-position limit. | +| `language` | string | model default | Whisper only. Two-letter language code (e.g. `en`, `es`, `de`) substituted into the decoder prompt's language token. Ignored by Parakeet and when an explicit `prompt` is supplied. | +| `timestamps` | bool | false | Whisper only. Decodes timestamp tokens and populates `segments` with `{start, end, text}` entries (seconds). Empty otherwise, including all Parakeet transcription. | **Response Format:** ```json @@ -531,6 +569,7 @@ int cactus_transcribe( "function_calls": [], "segments": [], "confidence": 1.0, + "confidence_threshold": -1.0, "time_to_first_token_ms": 120.0, "total_time_ms": 450.0, "prefill_tps": 50.0, @@ -543,8 +582,9 @@ int cactus_transcribe( ``` - `response`: Full transcription text -- `segments`: Always empty for the current native transcription paths +- `segments`: `{start, end, text}` entries (seconds), populated only for Whisper when the `timestamps` option is set; empty otherwise, including all Parakeet transcription - `cloud_handoff`: Always false for transcription +- `confidence_threshold`: `-1.0` (unset) — transcription does not resolve a cloud-handoff threshold **Example (file-based):** ```c @@ -569,6 +609,83 @@ int result = cactus_transcribe(whisper, NULL, NULL, pcm_data, pcm_size); ``` +### Streaming transcription +Transcribe continuously while audio is still being captured, instead of waiting for a complete recording. You push PCM as it arrives and read back text as soon as it stabilizes. Supported for the dedicated speech models (Whisper and Parakeet TDT). + +The session emits text in two parts on every call: + +- **`confirmed`** — newly finalized words. Append them to your running transcript; they never change. +- **`pending`** — the current best guess for the still-changing tail. Replace it on every call (do not append it); it is for live display only. + +Text is confirmed once two successive re-transcriptions of the audio agree (LocalAgreement, compared per **segment** for Whisper) or once a following token starts a new word (Parakeet TDT). Confirmed audio is dropped from the front of the buffer as the window advances, so memory stays bounded and the active window never approaches the model's fixed input length. + +#### `cactus_stream_transcribe_start` +Opens a streaming session bound to an already-initialized speech model. +```c +cactus_stream_transcribe_t cactus_stream_transcribe_start( + cactus_model_t model, // Whisper or Parakeet TDT model handle + const char* options_json // Optional (can be NULL); forwarded to cactus_transcribe +); +``` +**Returns:** an opaque session handle, or `NULL` on error (see `cactus_get_last_error`). Free it with `cactus_stream_transcribe_stop`. + +`options_json` (optional) is forwarded to the underlying `cactus_transcribe` call **for Whisper only** (e.g. `language`); the Parakeet TDT path ignores it. Chunking and segmentation are handled internally with no user-facing tunables. + +#### `cactus_stream_transcribe_process` +Feeds the next slice of audio. Input is 16-bit signed PCM, **16 kHz, mono** — the same format as `cactus_transcribe`'s `pcm_buffer`. Feed reasonably small chunks (≈ 0.1–2 s) for low latency. +```c +int cactus_stream_transcribe_process( + cactus_stream_transcribe_t stream, + const uint8_t* pcm_buffer, // 16-bit PCM, 16 kHz, mono + size_t pcm_buffer_size, // size in bytes + char* response_buffer, + size_t buffer_size +); +``` +**Returns:** bytes written to `response_buffer`, or -1 on error. + +**Response Format:** (also includes timing stats: `decode_tps`, `total_time_ms`, `time_to_first_token_ms`, `decode_tokens`) +```json +{ "success": true, "confirmed": "the quick brown", "pending": "fox jumps", "decode_tps": 0, "total_time_ms": 0, "time_to_first_token_ms": 0, "decode_tokens": 0 } +``` + +#### `cactus_stream_transcribe_stop` +Flushes any buffered audio, returns the final `confirmed` text, and destroys the session. +```c +int cactus_stream_transcribe_stop( + cactus_stream_transcribe_t stream, + char* response_buffer, // may be NULL to discard the final text + size_t buffer_size +); +``` +**Returns:** bytes written (0 when discarded), or -1 on error. + +**Example:** +```c +cactus_model_t whisper = cactus_init("../../weights/whisper-base", NULL, false); +cactus_stream_transcribe_t stream = cactus_stream_transcribe_start(whisper, NULL); + +char response[16384]; + +// Push audio as it is captured (here, 0.5s chunks of 16kHz mono PCM16). +for (each chunk) { + int rc = cactus_stream_transcribe_process( + stream, chunk_pcm, chunk_bytes, response, sizeof(response)); + if (rc < 0) break; + // Parse "confirmed" from response and append it to your transcript; + // show "confirmed-so-far + pending" live. +} + +// Flush the tail; its "confirmed" holds the final words. +cactus_stream_transcribe_stop(stream, response, sizeof(response)); +``` + +The `transcribe` CLI uses this streaming path for **live microphone** transcription (no file; a colored UI with running captions, press Enter to stop). With a **file** it does a one-shot transcription (long files are windowed internally, so there is no 30s limit): +```bash +cactus transcribe openai/whisper-base # live microphone (press Enter to stop) +cactus transcribe openai/whisper-base --file audio.wav # one-shot file transcription (any length) +``` + ### `cactus_embed` Generates text embeddings for semantic search, similarity, and RAG applications. @@ -1036,7 +1153,7 @@ int main() { ### Tool Calling ```c const char* tools = - "[{\"function\": {" + "[{\"type\": \"function\", \"function\": {" " \"name\": \"get_weather\"," " \"description\": \"Get weather for a location\"," " \"parameters\": {" diff --git a/docs/cactus_graph.md b/docs/cactus_graph.md index 89c42fcf9..2c56e0f50 100644 --- a/docs/cactus_graph.md +++ b/docs/cactus_graph.md @@ -371,6 +371,7 @@ GraphFile::save_node(graph, node_id, "output.bin"); #### GraphFile Namespace ```cpp +GraphFile::save_graph(graph, "graph.cactus"); GraphFile::SerializedGraph serialized = GraphFile::load_graph("graph.cactus"); CactusGraph loaded = CactusGraph::from_serialized(serialized); ``` @@ -506,11 +507,11 @@ size_t similarity = fixture.graph().divide(dot_product, fixture.graph().multiply 4. **Check precision**: Test operations with different precision types ### Contributing Graph Operations -1. ** Define the op in core graph types ** +1. **Define the op in core graph types** Add the new OpType in `cactus-graph/cactus_graph.h` If the op needs additional parameters, add the fields to OpParams in the same file -2. ** Add a graph builder API ** +2. **Add a graph builder API** Add a builder method in `cactus-graph/src/builder.cpp` and its declaration in `cactus-graph/cactus_graph.h` Follow the pattern of existing builder methods, e.g. for a new "relu" op: @@ -519,22 +520,22 @@ size_t CactusGraph::relu(size_t input) { return add_node(OpType::RELU, {input}, {}); } ``` -3. ** Implement the op in the execution engine ** +3. **Implement the op in the execution engine** Implement the kernel or graph op code in the relevant file, usually in `cactus-kernels/src/` Register the new op in the dispatch table in `cactus-graph/src/execute.cpp` for the supported backends (CPU, NPU) -4. ** Export op in FFI bindings ** +4. **Export op in FFI bindings** - header: `cactus-graph/cactus_graph.h` (in the `extern "C"` block) - implementation: `cactus-graph/src/graph_ffi.cpp` -5. ** Add python ctypes declaration ** +5. **Add python ctypes declaration** Add `_lib.cactus_graph_my_new_op.argtypes/restype` in `python/cactus/bindings/cactus.py` -6. ** Add python graph wrapper ** +6. **Add python graph wrapper** Add `Graph.my_new_op(...)` in `python/cactus/bindings/cactus.py`, and optionally a Tensor convenience method. -7. ** Add serialization schema entry if needed ** +7. **Add serialization schema entry if needed** If your op has extra parameters that need to be saved/loaded that are not in the default node, add new ParamField enum values. @@ -554,7 +555,7 @@ The syntax pattern there is: {ParamField::Mode, FieldPersistence::Persistent}, }}, ``` -8. ** Add test coverage ** +8. **Add test coverage** Add unit tests to `cactus-graph/tests/test_graph.cpp` covering the native graph function, and add python tests in `python/tests/test_graph.py` covering the Python API and end-to-end execution. diff --git a/docs/cactus_hybrid.md b/docs/cactus_hybrid.md index 1bc7f12b9..9e4f5fbfd 100644 --- a/docs/cactus_hybrid.md +++ b/docs/cactus_hybrid.md @@ -163,7 +163,8 @@ routed to the cloud. The response JSON indicates whether handoff occurred: "success": true, "cloud_handoff": true, "response": "Cloud-provided answer.", - "confidence": 0.18 + "confidence": 0.18, + "confidence_threshold": 0.7 } ``` @@ -171,9 +172,9 @@ In Python: ```python import json -from cactus import cactus_init, cactus_complete +from cactus import get_bundle_dir, cactus_init, cactus_complete -model = cactus_init("weights/gemma-4-E2B-it", None, False) +model = cactus_init(str(get_bundle_dir("google/gemma-4-E2B-it")), None, False) options = json.dumps({ "confidence_threshold": 0.7, "auto_handoff": True diff --git a/docs/cactus_kernels.md b/docs/cactus_kernels.md index 24ed0f52e..33cbb6dbe 100644 --- a/docs/cactus_kernels.md +++ b/docs/cactus_kernels.md @@ -27,12 +27,12 @@ enum class Precision { ## Element-wise Arithmetic ```cpp -void cactus_add_f16(const __fp16* a, const __fp16* b, __fp16* output, size_t n); -void cactus_subtract_f16(const __fp16* a, const __fp16* b, __fp16* output, size_t n); -void cactus_multiply_f16(const __fp16* a, const __fp16* b, __fp16* output, size_t n); -void cactus_divide_f16(const __fp16* a, const __fp16* b, __fp16* output, size_t n); -void cactus_add_f16_clipped(const __fp16* a, const __fp16* b, __fp16* output, size_t n); -void cactus_add_scaled_f16(const __fp16* base, const __fp16* src, __fp16* output, size_t n, float scale); +void cactus_add_f16(const __fp16* a, const __fp16* b, __fp16* output, size_t num_elements); +void cactus_subtract_f16(const __fp16* a, const __fp16* b, __fp16* output, size_t num_elements); +void cactus_multiply_f16(const __fp16* a, const __fp16* b, __fp16* output, size_t num_elements); +void cactus_divide_f16(const __fp16* a, const __fp16* b, __fp16* output, size_t num_elements); +void cactus_add_f16_clipped(const __fp16* a, const __fp16* b, __fp16* output, size_t num_elements); +void cactus_add_scaled_f16(const __fp16* base, const __fp16* src, __fp16* output, size_t num_elements, float scale); ``` ### Broadcast Variants @@ -53,7 +53,7 @@ void cactus_add_broadcast_f16( enum class ScalarOpType { ADD, SUBTRACT, MULTIPLY, DIVIDE, ABS, EXP, POW, SQRT, COS, SIN, LOG }; void cactus_scalar_op_f16( - const __fp16* input, __fp16* output, size_t n, + const __fp16* input, __fp16* output, size_t num_elements, float scalar_value, ScalarOpType op_type); ``` @@ -61,11 +61,11 @@ void cactus_scalar_op_f16( ```cpp // Full reductions (all elements) -double cactus_sum_all_f16(const __fp16* data, size_t n); -double cactus_mean_all_f16(const __fp16* data, size_t n); -double cactus_variance_all_f16(const __fp16* data, size_t n); -__fp16 cactus_min_all_f16(const __fp16* data, size_t n); -__fp16 cactus_max_all_f16(const __fp16* data, size_t n); +double cactus_sum_all_f16(const __fp16* data, size_t num_elements); +double cactus_mean_all_f16(const __fp16* data, size_t num_elements); +double cactus_variance_all_f16(const __fp16* data, size_t num_elements); +__fp16 cactus_min_all_f16(const __fp16* data, size_t num_elements); +__fp16 cactus_max_all_f16(const __fp16* data, size_t num_elements); // Axis reductions void cactus_sum_axis_f16(const __fp16* input, __fp16* output, @@ -91,20 +91,22 @@ CQ quantization uses Hadamard rotation + per-group codebooks for 1-4 bit weight ```cpp struct CactusQuantMatrix { - uint32_t bits; // 1, 2, 3, or 4 - uint32_t K, N; // matrix dimensions - uint32_t group_size; // elements per codebook group + uint32_t bits; // 1, 2, 3, or 4 + uint32_t K, N; // matrix dimensions + uint32_t group_size; // elements per codebook group uint32_t num_groups; - uint32_t flags; // ORTHOGONAL, INTERLEAVED_4ROW + uint32_t flags; // ORTHOGONAL, INTERLEAVED_4ROW const __fp16* codebook; const __fp16* input_scale; + const __fp16* input_scale_recip; const __fp16* norms; const uint8_t* packed_indices; const int8_t* left_signs; const int8_t* right_signs; const uint32_t* permutation; const __fp16* rotation; - // ... + const int8_t* expanded; + const float* norm_f32; }; // Unified dispatch (picks gemv or gemm based on M) @@ -231,14 +233,14 @@ void cactus_gpt_j_rope_f16( ## Activation Functions ```cpp -void cactus_relu_f16(const __fp16* input, __fp16* output, size_t n); -void cactus_leaky_relu_f16(const __fp16* input, __fp16* output, size_t n, float negative_slope); -void cactus_clamp_f16(const __fp16* input, __fp16* output, size_t n, float lo, float hi); -void cactus_silu_f16(const __fp16* input, __fp16* output, size_t n); -void cactus_gelu_f16(const __fp16* input, __fp16* output, size_t n); -void cactus_gelu_f16_erf(const __fp16* input, __fp16* output, size_t n); -void cactus_sigmoid_f16(const __fp16* input, __fp16* output, size_t n); -void cactus_tanh_f16(const __fp16* input, __fp16* output, size_t n); +void cactus_relu_f16(const __fp16* input, __fp16* output, size_t num_elements); +void cactus_leaky_relu_f16(const __fp16* input, __fp16* output, size_t num_elements, float negative_slope); +void cactus_clamp_f16(const __fp16* input, __fp16* output, size_t num_elements, float lo, float hi); +void cactus_silu_f16(const __fp16* input, __fp16* output, size_t num_elements); +void cactus_gelu_f16(const __fp16* input, __fp16* output, size_t num_elements); +void cactus_gelu_f16_erf(const __fp16* input, __fp16* output, size_t num_elements); +void cactus_sigmoid_f16(const __fp16* input, __fp16* output, size_t num_elements); +void cactus_tanh_f16(const __fp16* input, __fp16* output, size_t num_elements); void cactus_glu_f16(const __fp16* input, __fp16* output, size_t outer_size, size_t split_size, size_t inner_size); @@ -278,7 +280,7 @@ void cactus_conv2d_f16_k3s1p1_nchw(...); // 3x3 stride-1 pad-1 ```cpp void cactus_lstm_cell_f16( - const __fp16* x, const __fp16* h_prev, const __fp16* c_prev, + const __fp16* x_input, const __fp16* h_prev, const __fp16* c_prev, const __fp16* weight_ih, const __fp16* weight_hh, const __fp16* bias_ih, const __fp16* bias_hh, __fp16* h_new, __fp16* c_new, @@ -286,20 +288,20 @@ void cactus_lstm_cell_f16( void cactus_bilstm_sequence_f16( const __fp16* input, - const __fp16* w_ih_fwd, const __fp16* w_hh_fwd, const __fp16* b_ih_fwd, const __fp16* b_hh_fwd, - const __fp16* w_ih_bwd, const __fp16* w_hh_bwd, const __fp16* b_ih_bwd, const __fp16* b_hh_bwd, + const __fp16* weight_ih_fwd, const __fp16* weight_hh_fwd, const __fp16* bias_ih_fwd, const __fp16* bias_hh_fwd, + const __fp16* weight_ih_bwd, const __fp16* weight_hh_bwd, const __fp16* bias_ih_bwd, const __fp16* bias_hh_bwd, __fp16* output, size_t batch_size, size_t seq_len, size_t input_size, size_t hidden_size); void cactus_gated_deltanet_decode_f16( - const __fp16* q, const __fp16* k, const __fp16* v, - const __fp16* gate, const __fp16* beta, const __fp16* state, + const __fp16* q_data, const __fp16* k_data, const __fp16* v_data, + const __fp16* g_data, const __fp16* b_data, const __fp16* s_data, __fp16* out, size_t B, size_t Hq, size_t Hv, size_t K, size_t V, float scale); void cactus_gated_deltanet_prefill_f16( - const __fp16* q, const __fp16* k, const __fp16* v, - const __fp16* gate, const __fp16* beta, const __fp16* state, + const __fp16* q_data, const __fp16* k_data, const __fp16* v_data, + const __fp16* g_data, const __fp16* b_data, const __fp16* s_data, __fp16* out, size_t B, size_t T, size_t Hq, size_t Hv, - size_t K, size_t V, size_t chunk_size, float scale); + size_t K, size_t V, size_t requested_chunk_size, float scale); ``` ## Sampling diff --git a/docs/cactus_transpiler.md b/docs/cactus_transpiler.md index a1fe8a375..a0dbd7773 100644 --- a/docs/cactus_transpiler.md +++ b/docs/cactus_transpiler.md @@ -52,7 +52,8 @@ cactus convert ./classifier.pt ./classifier-cactus --bits 4 ```bash # Transpile: captures the graph, optimizes, and saves a bundle -cactus transpile ./classifier.pt --weights-dir ./classifier-cactus +cactus transpile ./classifier.pt \ + --weights-dir ./classifier-cactus --artifact-dir ./classifier-cactus # Run the transpiled bundle cactus run ./classifier-cactus @@ -121,6 +122,7 @@ the model in a thin adapter that exposes a stable, task-specific interface: | `multimodal_causal_lm_logits` | `input_ids`, image/audio features | next-token logits | | `encoder_hidden_states` | `input_features` | hidden states | | `ctc_logits` | `input_features` | CTC logits | +| `seq2seq_transcription` | `input_features` | transcription token IDs | | `tdt_transcription` | `input_features` | TDT token IDs | This is where model-family-specific knowledge lives (Gemma4, Qwen, LFM2, Parakeet, etc.). @@ -195,7 +197,8 @@ The most common use case is transpiling a model from HuggingFace: cactus convert google/gemma-4-E2B-it ./gemma4-weights --bits 4 # 2. Transpile -cactus transpile google/gemma-4-E2B-it --weights-dir ./gemma4-weights +cactus transpile google/gemma-4-E2B-it \ + --weights-dir ./gemma4-weights --artifact-dir ./gemma4-weights # 3. Run cactus run ./gemma4-weights --prompt "Hello!" @@ -423,7 +426,7 @@ cactus transpile [options] | Option | Description | |--------|-------------| | `--weights-dir ` | Path to converted CQ weights (default: `weights/`) | -| `--task ` | Force task type (default: `auto` — inferred from weights). Choices: `causal_lm_logits`, `multimodal_causal_lm_logits`, `ctc_logits`, `encoder_hidden_states`, `seq2seq_transcription`, `tdt_transcription` | +| `--task ` | Force task type (default: `auto` — inferred from model config). Choices: `causal_lm_logits`, `multimodal_causal_lm_logits`, `ctc_logits`, `encoder_hidden_states`, `seq2seq_transcription`, `tdt_transcription` | | `--prompt ` | Representative prompt for shape capture | | `--image-file ` | Representative image (repeatable) | | `--audio-file ` | Representative audio file | @@ -456,7 +459,7 @@ huggingface.co/Cactus-Compute) or a local path to a bundle directory. | Option | Description | |--------|-------------| | `--bits 1\|2\|3\|4` | CQ quantization bits when downloading (default: 4) | -| `--platform cpu\|apple` | Target accelerator (default: cpu — generic ARM) | +| `--platform auto\|cpu\|apple` | Target accelerator (default: auto — best for the host) | | `--token ` | HuggingFace token (gated models) | | `--prompt ` | Input prompt | | `--input-ids ` | Comma-separated token IDs | diff --git a/docs/finetuning.md b/docs/finetuning.md index 25ad62ee4..2e846a0b4 100644 --- a/docs/finetuning.md +++ b/docs/finetuning.md @@ -6,7 +6,7 @@ keywords: ["fine-tuning", "LoRA", "Unsloth", "mobile deployment", "on-device AI" # Deploying Unsloth Fine-Tunes to Cactus for Phones -- Cactus is an inference engine for mobile devices, macs and ARM chips like Raspberry Pi. +- Cactus is an inference engine for mobile devices, Macs, and ARM chips like Raspberry Pi. - At INT8, Cactus runs `Qwen3-0.6B` and `LFM2-1.2B` at `60-70 toks/sec` on iPhone 17 Pro, `13-18 toks/sec` on budget Pixel 6a. - INT4 quantization provides ~50% memory reduction with minimal quality loss. - Task-Specific INT8 tunes of `Gemma3-270m` hit `150 toks/sec` on iPhone 17 Pro and `23 toks/sec` on Raspberry Pi 5. @@ -87,7 +87,7 @@ Build the native library: ```bash cactus build --apple ``` -```bash +``` Build complete! Total time: 58 seconds Static libraries: @@ -127,7 +127,7 @@ cactus_destroy(model) You can now build iOS apps using the following code, but to see performance on any device while testing, -run cactus tests by plugging any iphone to your Mac then running: +run cactus tests by plugging any iPhone to your Mac then running: ```bash cactus test --model --transcription-model --ios @@ -135,7 +135,7 @@ cactus test --model --transcription-model /android/libcactus_engine.so Static library location: /android/libcactus_engine.a @@ -175,7 +175,7 @@ cactusDestroy(model) You can now build Android apps using the following code, but to see performance on any device while testing, -run cactus tests by plugging any android phone to your Mac then running: +run cactus tests by plugging any Android phone to your Mac then running: ```bash cactus test --model --transcription-model --android @@ -183,7 +183,7 @@ cactus test --model --transcription-model ```bash -# Download pre-built bundles (defaults to the generic CPU variant) +# Download pre-built bundles (defaults to --platform auto; use --platform cpu for the generic build) cactus download LiquidAI/LFM2-VL-450M cactus download openai/whisper-small --platform apple # CoreML/NPU variant @@ -32,8 +32,7 @@ cactus auth ```python -from cactus import ensure_model -from cactus import cactus_init, cactus_complete, cactus_destroy +from cactus import ensure_model, cactus_init, cactus_complete, cactus_destroy import json # Downloads the pre-built bundle from HuggingFace if not already present @@ -63,7 +62,7 @@ bundle = ensure_model("openai/whisper-tiny") # Or resolve the expected on-disk location explicitly bundle_dir = get_bundle_dir("openai/whisper-tiny", bits=4, platform=None) -# -> Path("transpiled/whisper-tiny-cq4") (or `-cq4-apple` with platform="apple") +# -> Path("weights/whisper-tiny-cq4") (or `-cq4-apple` with platform="apple") ``` ### Init / Lifecycle @@ -78,14 +77,14 @@ cactus_get_last_error() -> str | None ### Completion -Returns a `dict` with `success`, `error`, `cloud_handoff`, `response`, optional `thinking` (only present when the model emits chain-of-thought content, placed before `function_calls`), `function_calls`, `segments` (always `[]` for completion — populated only in transcription responses), `confidence`, timing stats (`time_to_first_token_ms`, `total_time_ms`, `prefill_tps`, `decode_tps`, `ram_usage_mb`), and token counts (`prefill_tokens`, `decode_tokens`, `total_tokens`). +Returns a `dict` with `success`, `error`, `cloud_handoff`, `response`, optional `thinking` (only present when the model emits chain-of-thought content, placed before `function_calls`), `function_calls`, `segments` (always `[]` for completion — populated only for Whisper transcription with the `timestamps` option), `confidence`, timing stats (`time_to_first_token_ms`, `total_time_ms`, `prefill_tps`, `decode_tps`, `ram_usage_mb`), and token counts (`prefill_tokens`, `decode_tokens`, `total_tokens`). ```python result = cactus_complete( model: int, - messages_json: str, # JSON array of {role, content} - options_json: str | None, # optional inference options - tools_json: str | None, # optional tool definitions + messages: list | str, # list of {role, content} dicts or JSON string + options: dict | str | None, # optional inference options + tools: list | str | None, # optional tool definitions callback: Callable[[str, int], None] | None, # streaming token callback pcm_data: list[int] | None = None # optional raw audio bytes ) -> dict @@ -96,7 +95,7 @@ result = cactus_complete( options = json.dumps({"max_tokens": 256, "temperature": 0.7}) def on_token(token, token_id): print(token, end="", flush=True) -result = cactus_complete(model, messages_json, options, None, on_token) +result = cactus_complete(model, messages, options, None, on_token) if result["cloud_handoff"]: # response already contains cloud result pass @@ -112,6 +111,7 @@ if result["cloud_handoff"]: "function_calls": [], "segments": [], "confidence": 0.92, + "confidence_threshold": 0.7, "time_to_first_token_ms": 45.2, "total_time_ms": 163.7, "prefill_tps": 619.5, @@ -130,11 +130,11 @@ Pre-processes input text and populates the KV cache without generating output to ```python cactus_prefill( model: int, - messages_json: str, # JSON array of {role, content} - options_json: str | None, # optional inference options - tools_json: str | None, # optional tool definitions + messages: list | str, # list of {role, content} dicts or JSON string + options: dict | str | None, # optional inference options + tools: list | str | None, # optional tool definitions pcm_data: list[int] | None = None # optional raw audio bytes -) -> None +) -> dict ``` ```python @@ -187,20 +187,20 @@ result = cactus_complete(model, completion_messages, None, tools, None) ### Transcription -Returns a `dict` with the `response` field (transcribed text), the `segments` array (timestamped segments as `{"start": , "end": , "text": ""}` — Whisper: phrase-level from timestamp tokens; Parakeet TDT: word-level from frame timing; Parakeet CTC and Moonshine: one segment per transcription window (consecutive VAD speech regions up to 30s)), and other metadata. +Returns a `dict` with the `response` field (transcribed text) and a `segments` array of `{start, end, text}` objects. `segments` is populated only for Whisper models when the `timestamps` option is set (`{"timestamps": True}`); it is empty otherwise, including for all Parakeet transcription. ```python result = cactus_transcribe( model: int, audio_path: str | None, prompt: str | None, - options_json: str | None, + options: dict | str | None, callback: Callable[[str, int], None] | None, pcm_data: list[int] | bytes | None ) -> dict ``` -**Custom vocabulary** biases the decoder toward domain-specific words (supported for Whisper and Moonshine models). Pass `custom_vocabulary` and `vocabulary_boost` in `options_json`: +**Custom vocabulary** biases the decoder toward domain-specific words (supported for Whisper and Moonshine models). Pass `custom_vocabulary` and `vocabulary_boost` in `options`: ```python options = json.dumps({ @@ -211,12 +211,33 @@ result = cactus_transcribe(model, "medical_notes.wav", None, options, None, None ``` ```python -result = cactus_transcribe(model, "/path/to/audio.wav", None, None, None, None) +result = cactus_transcribe(model, "/path/to/audio.wav", None, {"timestamps": True}, None, None) print(result["response"]) for seg in result["segments"]: print(f"[{seg['start']:.3f}s - {seg['end']:.3f}s] {seg['text']}") ``` +### Streaming transcription + +Transcribe continuously while audio is still being captured (Whisper and Parakeet TDT). Open a session, push 16 kHz mono 16-bit PCM chunks, and read text back as it stabilizes: `confirmed` words are final (append them to your transcript), `pending` is the volatile tail (replace it each call, for live display only). + +```python +stream = cactus_stream_transcribe_start(model: int, options: dict | str | None) -> int +result = cactus_stream_transcribe_process(stream: int, pcm_data: bytes) -> dict # {"confirmed": str, "pending": str, plus per-call timing stats} +result = cactus_stream_transcribe_stop(stream: int) -> dict # {"confirmed": str, "pending": ""}; destroys the session +``` + +`options` is forwarded to `cactus_transcribe` for **Whisper only** (e.g. `language`, `max_tokens`); the Parakeet TDT path ignores it. Chunking is handled internally. + +```python +stream = cactus_stream_transcribe_start(model, {"language": "en"}) +transcript = "" +for chunk in pcm_chunks: # each chunk: 16 kHz mono 16-bit PCM bytes + out = cactus_stream_transcribe_process(stream, chunk) + transcript += out["confirmed"] # show out["pending"] separately as a live preview +transcript += cactus_stream_transcribe_stop(stream)["confirmed"] +``` + ### Embeddings ```python @@ -256,7 +277,7 @@ cactus_index_add(index: int, ids: list[int], documents: list[str], metadatas: list[str] | None, embeddings: list[list[float]]) cactus_index_delete(index: int, ids: list[int]) result = cactus_index_get(index: int, ids: list[int]) -> dict -result = cactus_index_query(index: int, embedding: list[float], options_json: str | None) -> dict +result = cactus_index_query(index: int, embedding: list[float], options: dict | str | None) -> dict cactus_index_compact(index: int) cactus_index_destroy(index: int) ``` @@ -279,7 +300,7 @@ cactus_telemetry_flush() cactus_telemetry_shutdown() ``` -Functions that return a value raise `RuntimeError` on failure. `cactus_prefill`, `cactus_index_add`, `cactus_index_delete`, and `cactus_index_compact` also raise `RuntimeError` on failure despite not returning a value. Truly void functions that never raise: `cactus_destroy`, `cactus_reset`, `cactus_stop`, `cactus_index_destroy`, logging and telemetry functions. +Functions that return a value raise `RuntimeError` on failure. `cactus_index_add`, `cactus_index_delete`, and `cactus_index_compact` also raise `RuntimeError` on failure despite not returning a value. Truly void functions that never raise: `cactus_destroy`, `cactus_reset`, `cactus_stop`, `cactus_index_destroy`, logging and telemetry functions. ## Vision (VLM) @@ -335,7 +356,7 @@ g.set_input(a, np.array([[2, 4], [6, 8]], dtype=np.float16)) g.set_input(b, np.array([[1, 2], [3, 4]], dtype=np.float16)) g.execute() -print(y.numpy()) # [9. 36. 81. 144.] +print(y.numpy()) # [9. 144. 729. 2304.] ``` Supported ops: `+`, `-`, `*`, `/`, `abs`, `pow`, `view`, `flatten`, `concat`, `cat`, `relu`, `sigmoid`, `tanh`, `gelu`, `softmax`. diff --git a/python/cactus/__init__.py b/python/cactus/__init__.py index ac35d47ed..3f791dd07 100644 --- a/python/cactus/__init__.py +++ b/python/cactus/__init__.py @@ -24,6 +24,9 @@ "cactus_image_embed", "cactus_audio_embed", "cactus_transcribe", + "cactus_stream_transcribe_start", + "cactus_stream_transcribe_process", + "cactus_stream_transcribe_stop", "cactus_detect_language", "cactus_tokenize", "cactus_render_prompt", diff --git a/python/cactus/bindings/cactus.py b/python/cactus/bindings/cactus.py index 9a0e53974..aa40fc578 100644 --- a/python/cactus/bindings/cactus.py +++ b/python/cactus/bindings/cactus.py @@ -666,6 +666,18 @@ class cactus_tensor_info_t(ctypes.Structure): ] _lib.cactus_transcribe.restype = ctypes.c_int +_lib.cactus_stream_transcribe_start.argtypes = [ctypes.c_void_p, ctypes.c_char_p] +_lib.cactus_stream_transcribe_start.restype = ctypes.c_void_p + +_lib.cactus_stream_transcribe_process.argtypes = [ + ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint8), ctypes.c_size_t, + ctypes.c_char_p, ctypes.c_size_t +] +_lib.cactus_stream_transcribe_process.restype = ctypes.c_int + +_lib.cactus_stream_transcribe_stop.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_size_t] +_lib.cactus_stream_transcribe_stop.restype = ctypes.c_int + _bind_optional( "cactus_detect_language", [ @@ -1028,12 +1040,15 @@ def cactus_transcribe(model, audio_path, prompt=None, options=None, callback=Non model: Model handle. audio_path: Path to a WAV audio file. prompt: Optional prompt to guide transcription. - options: Optional dict of transcription options. + options: Optional dict of transcription options. Set ``timestamps: True`` (Whisper + only) to populate the ``segments`` array. callback: Optional function(text, token_id) for streaming tokens. pcm_data: Optional raw PCM audio bytes (alternative to audio_path). Returns: - A dict with transcription text and segments. + A dict with the transcribed text under ``response`` and a ``segments`` array of + ``{start, end, text}`` objects, populated only for Whisper when ``timestamps`` is set + (empty otherwise, including all Parakeet transcription). """ buf = ctypes.create_string_buffer(1 << 20) cb = _make_token_callback(callback) @@ -1047,6 +1062,62 @@ def cactus_transcribe(model, audio_path, prompt=None, options=None, callback=Non return _from_json(buf) +def cactus_stream_transcribe_start(model, options=None): + """Open a streaming transcription session on a Whisper or Parakeet TDT model. + + Args: + model: Model handle. + options: Optional dict forwarded to the underlying transcribe call + (e.g. {"language": "en", "max_tokens": 256}); chunking is + handled internally. + + Returns: + An opaque stream handle. Feed audio with cactus_stream_transcribe_process + and finish with cactus_stream_transcribe_stop. + """ + stream = _lib.cactus_stream_transcribe_start(model, _to_json(options)) + if not stream: + raise RuntimeError(_err("Failed to start streaming transcription")) + return stream + + +def cactus_stream_transcribe_process(stream, pcm_data): + """Feed the next slice of audio to a streaming session. + + Args: + stream: Stream handle from cactus_stream_transcribe_start. + pcm_data: 16 kHz mono 16-bit PCM audio bytes for this chunk. + + Returns: + A dict with "confirmed" (newly finalized text, append it), "pending" (volatile tail, + replace it each call), and per-call stats ("decode_tps", + "total_time_ms", "time_to_first_token_ms", "decode_tokens"). + """ + buf = ctypes.create_string_buffer(1 << 16) + pcm_ptr, pcm_size = _prepare_pcm(pcm_data) + rc = _lib.cactus_stream_transcribe_process(stream, pcm_ptr, pcm_size, buf, len(buf)) + if rc < 0: + raise RuntimeError(_err("Stream transcription failed")) + return _from_json(buf) + + +def cactus_stream_transcribe_stop(stream): + """Flush buffered audio, return the final text, and destroy the session. + + Args: + stream: Stream handle from cactus_stream_transcribe_start. + + Returns: + A dict {"success": True, "confirmed": , "pending": ""}. The handle is invalid + after this call. + """ + buf = ctypes.create_string_buffer(1 << 16) + rc = _lib.cactus_stream_transcribe_stop(stream, buf, len(buf)) + if rc < 0: + raise RuntimeError(_err("Stream transcription stop failed")) + return _from_json(buf) + + def cactus_detect_language(model, audio_path, options=None, pcm_data=None): """Detect the spoken language in audio. diff --git a/python/cactus/cli/__init__.py b/python/cactus/cli/__init__.py index 40847fc84..c0bc5ba34 100644 --- a/python/cactus/cli/__init__.py +++ b/python/cactus/cli/__init__.py @@ -105,8 +105,8 @@ def create_parser(): --token HuggingFace token (gated models) --reconvert force local convert+transpile fallback - cactus transcribe [model] transcribe audio with a model - --file audio file to transcribe (required) + cactus transcribe [model] live microphone transcription with a model + --file audio file to transcribe (WAV) --language language code (default: en) --bits 1|2|3|4 CQ quantization (default: 4) --platform {_PLATFORM_PIPE:<22} target accelerator (default: auto) @@ -267,7 +267,7 @@ def create_parser(): transcribe_parser.add_argument("model_id", nargs="?", default=DEFAULT_TRANSCRIPTION_MODEL_ID, type=_hf_id_or_path, help=f"HuggingFace model id (default: {DEFAULT_TRANSCRIPTION_MODEL_ID})") - transcribe_parser.add_argument("--file", dest="audio_file", required=True, + transcribe_parser.add_argument("--file", dest="audio_file", default=None, help="Audio file to transcribe (WAV)") transcribe_parser.add_argument("--language", default="en", help="Language code (default: en)") @@ -304,7 +304,7 @@ def create_parser(): type=_hf_id_or_path, help=f"HF transcription model ID under test (default: {DEFAULT_TEST_TRANSCRIPTION_MODEL_ID})") test_parser.add_argument("--suite", default=None, - help="Run a single test suite by name; resolved across all components (e.g. llm → engine, performance → kernels + graph)") + help="Run a single test suite by name; resolved across all components (e.g. llm → engine)") test_parser.add_argument("--list", action="store_true", help="List available components and engine tests, then exit") test_parser.add_argument("--android", action="store_true", help="Run tests on Android") diff --git a/python/cactus/cli/transcribe.py b/python/cactus/cli/transcribe.py index d13f4a519..33af24ae0 100644 --- a/python/cactus/cli/transcribe.py +++ b/python/cactus/cli/transcribe.py @@ -9,11 +9,12 @@ def cmd_transcribe(args): from .model import ensure_runnable_bundle, TranspileOptions from .download import resolve_platform - audio_path = Path(args.audio_file).expanduser() - if not audio_path.is_file(): - print_color(RED, f"Audio file not found: {audio_path}") - return 1 - args.audio_file = str(audio_path) + if args.audio_file: + audio_path = Path(args.audio_file).expanduser() + if not audio_path.is_file(): + print_color(RED, f"Audio file not found: {audio_path}") + return 1 + args.audio_file = str(audio_path) if args.no_cloud_tele: os.environ["CACTUS_NO_CLOUD_TELE"] = "1" @@ -42,7 +43,9 @@ def cmd_transcribe(args): if binary is None: return 1 - cmd = [str(binary), str(bundle_dir), args.audio_file] + cmd = [str(binary), str(bundle_dir)] + if args.audio_file: + cmd.append(args.audio_file) if args.language: cmd.extend(["--language", args.language]) diff --git a/python/cactus/server.py b/python/cactus/server.py index 0697b66ac..c8f9206a6 100644 --- a/python/cactus/server.py +++ b/python/cactus/server.py @@ -590,6 +590,8 @@ async def create_transcription( options["temperature"] = temperature if language: options["language"] = language + if response_format == "verbose_json": + options["timestamps"] = True mgr: ModelManager = request.app.state.manager async with mgr.acquire(model_id) as slot: async with slot.lock: diff --git a/python/cactus/transpile/README.md b/python/cactus/transpile/README.md index 52053dda9..a0a059658 100644 --- a/python/cactus/transpile/README.md +++ b/python/cactus/transpile/README.md @@ -63,7 +63,7 @@ Two important details: order (`python/cactus/cli/model.py:ensure_runnable_bundle`): 1. `` is an existing path with `components/manifest.json` → run as-is. -2. Cached download at `transpiled/-cq[-]/` → run. +2. Cached download at `weights/-cq[-]/` → run. 3. Fresh download from `huggingface.co/Cactus-Compute` via `download_bundle`. 4. Fallback: local `ensure_bundle` (convert + transpile for custom models not on HF). @@ -856,7 +856,7 @@ By default this: `--weights-dir` or found in the default `weights/` location - resolves the runtime library - launches `python/cactus/transpile/hf_model.py` -- saves artifacts under `./transpiled//` +- saves artifacts under `weights//` (alongside the converted CQ weights; override with `--artifact-dir`) - adds `--skip-execute`, so it does not run the graph unless you ask To also execute right after transpiling: