From fb8b7a359a56af8d24eed1b44264f2187e314a9c Mon Sep 17 00:00:00 2001 From: localai-org-maint-bot Date: Mon, 7 Sep 2026 17:12:37 +0200 Subject: [PATCH] fix(distributed): stage sound detection audio (#11907) * fix(distributed): stage sound detection audio Sound detection passes frontend temporary paths directly to remote workers, unlike transcription. Stage the WAV before classification so CED can read it without a shared temporary directory. Preserve the original request for retries and propagate staging errors without calling the backend. Cover staging, request preservation, and error handling with regression tests. Assisted-by: Codex:GPT-6 golangci-lint * test(distributed): verify routed sound staging Call sound detection through the client returned by SmartRouter.Route. This checks interface dispatch through both routing wrappers, rather than constructing FileStagingClient directly. The test fails without the sound-staging override and passes with it. Assisted-by: Codex:GPT-6 golangci-lint --------- Co-authored-by: Ettore Di Giacinto --- core/services/nodes/file_staging_client.go | 14 +++ .../file_staging_sound_detection_test.go | 94 +++++++++++++++++++ docs/content/features/audio-classification.md | 4 + 3 files changed, 112 insertions(+) create mode 100644 core/services/nodes/file_staging_sound_detection_test.go diff --git a/core/services/nodes/file_staging_client.go b/core/services/nodes/file_staging_client.go index bfc202c82..d6c14cf2e 100644 --- a/core/services/nodes/file_staging_client.go +++ b/core/services/nodes/file_staging_client.go @@ -16,6 +16,7 @@ import ( pb "github.com/mudler/LocalAI/pkg/grpc/proto" "github.com/mudler/xlog" ggrpc "google.golang.org/grpc" + "google.golang.org/protobuf/proto" ) // FileStagingClient wraps a grpc.Backend to transparently handle file transfer @@ -342,6 +343,19 @@ func (f *FileStagingClient) SoundGeneration(ctx context.Context, in *pb.SoundGen return result, nil } +func (f *FileStagingClient) SoundDetection(ctx context.Context, in *pb.SoundDetectionRequest, opts ...ggrpc.CallOption) (*pb.SoundDetectionResponse, error) { + if in.Src != "" && isFilePath(in.Src) { + backendPath, _, err := f.stageInputFile(ctx, requestID(), in.Src, "inputs") + if err != nil { + return nil, fmt.Errorf("staging audio for sound detection: %w", err) + } + // Keep the frontend path available if the caller retries on another node. + in = proto.Clone(in).(*pb.SoundDetectionRequest) + in.Src = backendPath + } + return f.Backend.SoundDetection(ctx, in, opts...) +} + func (f *FileStagingClient) AudioTranscription(ctx context.Context, in *pb.TranscriptRequest, opts ...ggrpc.CallOption) (*pb.TranscriptResult, error) { reqID := requestID() diff --git a/core/services/nodes/file_staging_sound_detection_test.go b/core/services/nodes/file_staging_sound_detection_test.go new file mode 100644 index 000000000..e66b18f5f --- /dev/null +++ b/core/services/nodes/file_staging_sound_detection_test.go @@ -0,0 +1,94 @@ +package nodes + +import ( + "context" + "errors" + + grpc "github.com/mudler/LocalAI/pkg/grpc" + pb "github.com/mudler/LocalAI/pkg/grpc/proto" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + ggrpc "google.golang.org/grpc" +) + +type soundStagingBackend struct { + grpc.Backend + request *pb.SoundDetectionRequest + err error +} + +func (b *soundStagingBackend) SoundDetection(_ context.Context, in *pb.SoundDetectionRequest, _ ...ggrpc.CallOption) (*pb.SoundDetectionResponse, error) { + b.request = in + return &pb.SoundDetectionResponse{}, b.err +} + +type soundStagingFailure struct{ FileStager } + +func (s *soundStagingFailure) EnsureRemote(context.Context, string, string, string) (string, error) { + return "", errors.New("upload failed") +} + +type soundRouteFactory struct{ client grpc.Backend } + +func (f *soundRouteFactory) NewClient(string, bool) grpc.Backend { return f.client } + +var _ = Describe("FileStagingClient sound detection", func() { + It("stages sound audio through the client returned by SmartRouter.Route", func(ctx SpecContext) { + node := &BackendNode{ID: "worker-1", Name: "worker", Address: "10.0.0.1:50051"} + reg := &fakeModelRouter{ + findAndLockNode: node, + findAndLockNM: &NodeModel{NodeID: node.ID, ModelName: "ced", Address: "10.0.0.1:9001"}, + } + backend := &soundStagingBackend{Backend: &stubBackend{healthResult: true}} + stager := &fakeFileStager{} + router := NewSmartRouter(reg, SmartRouterOptions{ + ClientFactory: &soundRouteFactory{client: backend}, + FileStager: stager, + Unloader: &fakeUnloader{}, + }) + result, err := router.Route(ctx, "ced", "ced.gguf", "ced", "", nil, false) + Expect(err).NotTo(HaveOccurred()) + Expect(result).NotTo(BeNil()) + defer result.Release() + request := &pb.SoundDetectionRequest{Src: "/tmp/realtime-sound-window-test.wav", ModelIdentity: "ced.gguf"} + _, err = result.Client.SoundDetection(ctx, request) + Expect(err).NotTo(HaveOccurred()) + Expect(stager.ensureCalls).To(HaveLen(1)) + Expect(stager.ensureCalls[0].localPath).To(Equal(request.Src)) + Expect(backend.request.Src).To(Equal("/remote/" + stager.ensureCalls[0].key)) + Expect(request.Src).To(Equal("/tmp/realtime-sound-window-test.wav")) + }) + + It("stages audio on the worker without changing the caller's request", func(ctx SpecContext) { + backend := &soundStagingBackend{} + stager := &fakeFileStager{} + client := NewFileStagingClient(backend, stager, "worker-1") + request := &pb.SoundDetectionRequest{Src: "/tmp/realtime-sound-window-test.wav", ModelIdentity: "ced.gguf", TopK: 5, Threshold: 0.25} + _, err := client.SoundDetection(ctx, request) + Expect(err).NotTo(HaveOccurred()) + Expect(stager.ensureCalls).To(HaveLen(1)) + Expect(stager.ensureCalls[0].nodeID).To(Equal("worker-1")) + Expect(stager.ensureCalls[0].localPath).To(Equal(request.Src)) + Expect(backend.request.Src).To(Equal("/remote/" + stager.ensureCalls[0].key)) + Expect(backend.request.ModelIdentity).To(Equal(request.ModelIdentity)) + Expect(backend.request.TopK).To(Equal(request.TopK)) + Expect(backend.request.Threshold).To(Equal(request.Threshold)) + Expect(request.Src).To(Equal("/tmp/realtime-sound-window-test.wav")) + }) + It("does not call the backend when staging fails", func(ctx SpecContext) { + backend := &soundStagingBackend{} + client := NewFileStagingClient(backend, &soundStagingFailure{}, "worker-1") + _, err := client.SoundDetection(ctx, &pb.SoundDetectionRequest{Src: "/tmp/clip.wav"}) + Expect(err).To(MatchError(ContainSubstring("upload failed"))) + Expect(backend.request).To(BeNil()) + }) + It("passes through requests without a file and preserves backend errors", func(ctx SpecContext) { + failure := errors.New("classifier failed") + backend := &soundStagingBackend{err: failure} + stager := &fakeFileStager{} + client := NewFileStagingClient(backend, stager, "worker-1") + _, err := client.SoundDetection(ctx, &pb.SoundDetectionRequest{}) + Expect(err).To(MatchError(failure)) + Expect(stager.ensureCalls).To(BeEmpty()) + }) +}) diff --git a/docs/content/features/audio-classification.md b/docs/content/features/audio-classification.md index 045074340..09759ab72 100644 --- a/docs/content/features/audio-classification.md +++ b/docs/content/features/audio-classification.md @@ -11,6 +11,10 @@ LocalAI exposes this through the `/v1/audio/classification` endpoint, modelled a Because classification is exposed as a regular OpenAI-style endpoint, any HTTP client works - there is no Python dependency on the consumer side. +In distributed mode, LocalAI stages uploaded audio and realtime sound-detection +windows on the selected worker before classification. The API server and worker +do not need a shared temporary directory. + ## Endpoint ```