mirror of
https://github.com/mudler/LocalAI.git
synced 2026-05-17 13:10:23 -04:00
* refactor(transcription): propagate request ctx through ModelTranscription* Replaces context.Background() with the HTTP request ctx so client disconnects start cancelling the gRPC call. No backend-side abort wiring yet — that comes in a later commit. Pure plumbing. Assisted-by: Claude:claude-haiku-4-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * fix(cli): pass ctx to backend.ModelTranscription Follow-up toe65d3e1fwhich threaded ctx through ModelTranscription but missed the CLI caller. CLI commands have no request-scoped ctx, so context.Background() is correct here. Assisted-by: Claude:claude-haiku-4-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor(audio): propagate request ctx into TTS, sound-gen, audio-transform Same ctx-plumbing pattern applied to the rest of the audio path. CLI callers use context.Background() since there is no request scope; HTTP callers use c.Request().Context(). Assisted-by: Claude:claude-haiku-4-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor(backend): propagate request ctx into biometric, detection, rerank, diarization paths Replaces remaining context.Background() sites in core/backend with the caller's ctx. After this commit, every core/backend/*.go entry point threads the request ctx end-to-end to the gRPC client. Assisted-by: Claude:claude-haiku-4-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * refactor(grpc): plumb ctx through AIModel.AudioTranscription{,Stream} Adds context.Context as first parameter to the AIModel interface methods that wrap whisper-style transcription. Server-side gRPC handler now forwards the per-RPC ctx (server-streaming uses stream.Context()). Whisper, Voxtral, vibevoice-cpp, and sherpa-onnx accept the parameter; none uses it yet — the actual cancellation primitive lands in the next commit so this is pure plumbing. Assisted-by: Claude:claude-sonnet-4-6 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * feat(whisper): add abort_callback hook in the C++ bridge Installs a std::atomic<int> flag, wires it into whisper_full_params.abort_callback, and exposes a set_abort(int) C symbol so Go can flip the flag from a goroutine watching the request context. transcribe() now distinguishes abort (return 2) from real whisper_full failure (return 1). Assisted-by: Claude:claude-haiku-4-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * feat(whisper): register set_abort symbol in the purego loader Adds the Go-side binding for the new C export so the next commit can call CppSetAbort(1) from a watcher goroutine on ctx.Done(). Assisted-by: Claude:claude-haiku-4-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * feat(whisper): honor ctx cancellation and return codes.Canceled A watcher goroutine watches ctx.Done() during AudioTranscription and calls CppSetAbort(1) on cancel. whisper_full sees abort_callback return true at the next compute graph step, returns non-zero, and the bridge returns 2 -> AudioTranscription maps that to codes.Canceled. Adds an opt-in test (gated on WHISPER_MODEL_PATH / WHISPER_AUDIO_PATH) that asserts cancellation latency under 5s and proves the abort flag resets cleanly so the next transcription succeeds. Assisted-by: Claude:claude-sonnet-4-6 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * fix(whisper): join the cancel watcher goroutine before returning Follow-up to85edf9d2. The previous commit used `defer close(done)` and called the watcher "joined synchronously" — but close() only signals, it does not block until the goroutine exits. That left a window where a late CppSetAbort(1) from a cancelled call could land on the next call, after its C-side g_abort reset but before whisper_full() began polling the abort callback, corrupting the second transcription. Switch to a sync.WaitGroup join so wg.Wait() blocks until the watcher has actually returned from its select. Assisted-by: Claude:claude-sonnet-4-6 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * fix(whisper): short-circuit pre-cancelled ctx in AudioTranscription If ctx is already Done() at entry, return codes.Canceled immediately instead of running the full transcription. The C-side g_abort reset happens at the start of transcribe() and would otherwise overwrite a watcher-set abort flag from an already-cancelled ctx, producing a spurious successful transcription on a request the client has already abandoned. Assisted-by: Claude:claude-haiku-4-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * fix(tests/distributed): update testLLM mock for new AudioTranscription signature Phase B (93c48e19) added context.Context to AIModel.AudioTranscription but missed the testLLM mock in tests/e2e/distributed. CI golangci-lint caught it: *testLLM did not implement grpc.AIModel because the method signature lacked the ctx parameter, which broke the distributed test suite compilation and cascaded through every backend-build job that runs `go build ./...`. Assisted-by: Claude:claude-opus-4-7 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * test(whisper): port cancellation test to Ginkgo/Gomega Project policy (.agents/coding-style.md, enforced by golangci-lint forbidigo) is that all Go tests must use Ginkgo v2 + Gomega — no stdlib testing patterns (t.Skip, t.Fatalf, etc.). Convert the cancellation test to a Describe/It block with Skip(...) for env gating and Expect/HaveOccurred for assertions. Same coverage: cancel mid-flight returns codes.Canceled within 5s and a follow-up transcription succeeds, proving the C-side g_abort flag resets cleanly. Assisted-by: Claude:claude-opus-4-7 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> --------- Signed-off-by: Ettore Di Giacinto <mudler@localai.io> Co-authored-by: Ettore Di Giacinto <mudler@localai.io>
284 lines
7.6 KiB
Go
284 lines
7.6 KiB
Go
package backend
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/binary"
|
|
"encoding/json"
|
|
"fmt"
|
|
"maps"
|
|
"os"
|
|
"path/filepath"
|
|
"time"
|
|
|
|
"github.com/mudler/LocalAI/core/config"
|
|
"github.com/mudler/LocalAI/core/trace"
|
|
laudio "github.com/mudler/LocalAI/pkg/audio"
|
|
|
|
"github.com/mudler/LocalAI/pkg/grpc/proto"
|
|
"github.com/mudler/LocalAI/pkg/model"
|
|
"github.com/mudler/LocalAI/pkg/utils"
|
|
)
|
|
|
|
func ModelTTS(
|
|
ctx context.Context,
|
|
text,
|
|
voice,
|
|
language string,
|
|
loader *model.ModelLoader,
|
|
appConfig *config.ApplicationConfig,
|
|
modelConfig config.ModelConfig,
|
|
) (string, *proto.Result, error) {
|
|
opts := ModelOptions(modelConfig, appConfig)
|
|
ttsModel, err := loader.Load(opts...)
|
|
if err != nil {
|
|
recordModelLoadFailure(appConfig, modelConfig.Name, modelConfig.Backend, err, nil)
|
|
return "", nil, err
|
|
}
|
|
|
|
if ttsModel == nil {
|
|
return "", nil, fmt.Errorf("could not load tts model %q", modelConfig.Model)
|
|
}
|
|
|
|
audioDir := filepath.Join(appConfig.GeneratedContentDir, "audio")
|
|
if err := os.MkdirAll(audioDir, 0750); err != nil {
|
|
return "", nil, fmt.Errorf("failed creating audio directory: %s", err)
|
|
}
|
|
|
|
fileName := utils.GenerateUniqueFileName(audioDir, "tts", ".wav")
|
|
filePath := filepath.Join(audioDir, fileName)
|
|
|
|
// We join the model name to the model path here. This seems to only be done for TTS and is HIGHLY suspect.
|
|
// This should be addressed in a follow up PR soon.
|
|
// Copying it over nearly verbatim, as TTS backends are not functional without this.
|
|
modelPath := ""
|
|
// Checking first that it exists and is not outside ModelPath
|
|
// TODO: we should actually first check if the modelFile is looking like
|
|
// a FS path
|
|
mp := filepath.Join(loader.ModelPath, modelConfig.Model)
|
|
if _, err := os.Stat(mp); err == nil {
|
|
if err := utils.VerifyPath(mp, appConfig.SystemState.Model.ModelsPath); err != nil {
|
|
return "", nil, err
|
|
}
|
|
modelPath = mp
|
|
} else {
|
|
modelPath = modelConfig.Model // skip this step if it fails?????
|
|
}
|
|
|
|
var startTime time.Time
|
|
if appConfig.EnableTracing {
|
|
trace.InitBackendTracingIfEnabled(appConfig.TracingMaxItems)
|
|
startTime = time.Now()
|
|
}
|
|
|
|
res, err := ttsModel.TTS(ctx, &proto.TTSRequest{
|
|
Text: text,
|
|
Model: modelPath,
|
|
Voice: voice,
|
|
Dst: filePath,
|
|
Language: &language,
|
|
})
|
|
|
|
if appConfig.EnableTracing {
|
|
errStr := ""
|
|
if err != nil {
|
|
errStr = err.Error()
|
|
} else if !res.Success {
|
|
errStr = fmt.Sprintf("TTS error: %s", res.Message)
|
|
}
|
|
|
|
data := map[string]any{
|
|
"text": text,
|
|
"voice": voice,
|
|
"language": language,
|
|
}
|
|
if err == nil && res.Success {
|
|
if snippet := trace.AudioSnippet(filePath); snippet != nil {
|
|
maps.Copy(data, snippet)
|
|
}
|
|
}
|
|
trace.RecordBackendTrace(trace.BackendTrace{
|
|
Timestamp: startTime,
|
|
Duration: time.Since(startTime),
|
|
Type: trace.BackendTraceTTS,
|
|
ModelName: modelConfig.Name,
|
|
Backend: modelConfig.Backend,
|
|
Summary: trace.TruncateString(text, 200),
|
|
Error: errStr,
|
|
Data: data,
|
|
})
|
|
}
|
|
|
|
if err != nil {
|
|
return "", nil, err
|
|
}
|
|
|
|
// return RPC error if any
|
|
if !res.Success {
|
|
return "", nil, fmt.Errorf("error during TTS: %s", res.Message)
|
|
}
|
|
|
|
return filePath, res, err
|
|
}
|
|
|
|
func ModelTTSStream(
|
|
ctx context.Context,
|
|
text,
|
|
voice,
|
|
language string,
|
|
loader *model.ModelLoader,
|
|
appConfig *config.ApplicationConfig,
|
|
modelConfig config.ModelConfig,
|
|
audioCallback func([]byte) error,
|
|
) error {
|
|
opts := ModelOptions(modelConfig, appConfig)
|
|
ttsModel, err := loader.Load(opts...)
|
|
if err != nil {
|
|
recordModelLoadFailure(appConfig, modelConfig.Name, modelConfig.Backend, err, nil)
|
|
return err
|
|
}
|
|
|
|
if ttsModel == nil {
|
|
return fmt.Errorf("could not load tts model %q", modelConfig.Model)
|
|
}
|
|
|
|
// We join the model name to the model path here. This seems to only be done for TTS and is HIGHLY suspect.
|
|
// This should be addressed in a follow up PR soon.
|
|
// Copying it over nearly verbatim, as TTS backends are not functional without this.
|
|
modelPath := ""
|
|
// Checking first that it exists and is not outside ModelPath
|
|
// TODO: we should actually first check if the modelFile is looking like
|
|
// a FS path
|
|
mp := filepath.Join(loader.ModelPath, modelConfig.Model)
|
|
if _, err := os.Stat(mp); err == nil {
|
|
if err := utils.VerifyPath(mp, appConfig.SystemState.Model.ModelsPath); err != nil {
|
|
return err
|
|
}
|
|
modelPath = mp
|
|
} else {
|
|
modelPath = modelConfig.Model // skip this step if it fails?????
|
|
}
|
|
|
|
var startTime time.Time
|
|
if appConfig.EnableTracing {
|
|
trace.InitBackendTracingIfEnabled(appConfig.TracingMaxItems)
|
|
startTime = time.Now()
|
|
}
|
|
|
|
var sampleRate uint32 = 16000 // default
|
|
headerSent := false
|
|
var callbackErr error
|
|
|
|
// Collect up to 30s of audio for tracing
|
|
var snippetPCM []byte
|
|
var totalPCMBytes int
|
|
snippetCapped := false
|
|
|
|
err = ttsModel.TTSStream(ctx, &proto.TTSRequest{
|
|
Text: text,
|
|
Model: modelPath,
|
|
Voice: voice,
|
|
Language: &language,
|
|
}, func(reply *proto.Reply) {
|
|
// First message contains sample rate info
|
|
if !headerSent && len(reply.Message) > 0 {
|
|
var info map[string]any
|
|
if json.Unmarshal(reply.Message, &info) == nil {
|
|
if sr, ok := info["sample_rate"].(float64); ok {
|
|
sampleRate = uint32(sr)
|
|
}
|
|
}
|
|
// Send WAV header with placeholder size (0xFFFFFFFF for streaming)
|
|
header := laudio.WAVHeader{
|
|
ChunkID: [4]byte{'R', 'I', 'F', 'F'},
|
|
ChunkSize: 0xFFFFFFFF, // Unknown size for streaming
|
|
Format: [4]byte{'W', 'A', 'V', 'E'},
|
|
Subchunk1ID: [4]byte{'f', 'm', 't', ' '},
|
|
Subchunk1Size: 16,
|
|
AudioFormat: 1, // PCM
|
|
NumChannels: 1, // Mono
|
|
SampleRate: sampleRate,
|
|
ByteRate: sampleRate * 2, // SampleRate * BlockAlign
|
|
BlockAlign: 2, // 16-bit = 2 bytes
|
|
BitsPerSample: 16,
|
|
Subchunk2ID: [4]byte{'d', 'a', 't', 'a'},
|
|
Subchunk2Size: 0xFFFFFFFF, // Unknown size for streaming
|
|
}
|
|
|
|
var buf bytes.Buffer
|
|
if writeErr := binary.Write(&buf, binary.LittleEndian, header); writeErr != nil {
|
|
callbackErr = writeErr
|
|
return
|
|
}
|
|
|
|
if writeErr := audioCallback(buf.Bytes()); writeErr != nil {
|
|
callbackErr = writeErr
|
|
return
|
|
}
|
|
headerSent = true
|
|
}
|
|
|
|
// Stream audio chunks
|
|
if len(reply.Audio) > 0 {
|
|
if writeErr := audioCallback(reply.Audio); writeErr != nil {
|
|
callbackErr = writeErr
|
|
}
|
|
// Accumulate PCM for tracing snippet
|
|
totalPCMBytes += len(reply.Audio)
|
|
if appConfig.EnableTracing && !snippetCapped {
|
|
maxBytes := int(sampleRate) * 2 * trace.MaxSnippetSeconds // 16-bit mono
|
|
if len(snippetPCM)+len(reply.Audio) <= maxBytes {
|
|
snippetPCM = append(snippetPCM, reply.Audio...)
|
|
} else {
|
|
remaining := maxBytes - len(snippetPCM)
|
|
if remaining > 0 {
|
|
// Align to sample boundary (2 bytes per sample)
|
|
remaining = remaining &^ 1
|
|
snippetPCM = append(snippetPCM, reply.Audio[:remaining]...)
|
|
}
|
|
snippetCapped = true
|
|
}
|
|
}
|
|
}
|
|
})
|
|
|
|
resultErr := err
|
|
if callbackErr != nil {
|
|
resultErr = callbackErr
|
|
}
|
|
|
|
if appConfig.EnableTracing {
|
|
errStr := ""
|
|
if resultErr != nil {
|
|
errStr = resultErr.Error()
|
|
}
|
|
|
|
data := map[string]any{
|
|
"text": text,
|
|
"voice": voice,
|
|
"language": language,
|
|
"streaming": true,
|
|
}
|
|
if resultErr == nil && len(snippetPCM) > 0 {
|
|
if snippet := trace.AudioSnippetFromPCM(snippetPCM, int(sampleRate), totalPCMBytes); snippet != nil {
|
|
maps.Copy(data, snippet)
|
|
}
|
|
}
|
|
trace.RecordBackendTrace(trace.BackendTrace{
|
|
Timestamp: startTime,
|
|
Duration: time.Since(startTime),
|
|
Type: trace.BackendTraceTTS,
|
|
ModelName: modelConfig.Name,
|
|
Backend: modelConfig.Backend,
|
|
Summary: trace.TruncateString(text, 200),
|
|
Error: errStr,
|
|
Data: data,
|
|
})
|
|
}
|
|
|
|
if callbackErr != nil {
|
|
return callbackErr
|
|
}
|
|
return err
|
|
}
|