mudler--localai/core/backend/transcript.go

package backend

import (
	"context"
	"fmt"
	"maps"
	"time"

	"github.com/mudler/LocalAI/core/config"
	"github.com/mudler/LocalAI/core/schema"
	"github.com/mudler/LocalAI/core/trace"

	grpcPkg "github.com/mudler/LocalAI/pkg/grpc"
	"github.com/mudler/LocalAI/pkg/grpc/proto"
	"github.com/mudler/LocalAI/pkg/model"
)

// TranscriptionRequest groups the parameters accepted by ModelTranscription.
// Use this so callers don't have to pass long positional arg lists when they
// only care about a subset of fields.
type TranscriptionRequest struct {
	Audio                  string
	Language               string
	Translate              bool
	Diarize                bool
	Prompt                 string
	Temperature            float32
	TimestampGranularities []string
}

// modelIdentity is ModelConfig.Model, the value the backend received as
// ModelOptions.Model at LoadModel, so it can reject a request that reached it
// through a stale distributed route (#10952). It is a parameter rather than a
// TranscriptionRequest field because the request is built by HTTP handlers that
// have no ModelConfig, while every caller of this method does.
func (r *TranscriptionRequest) toProto(threads uint32, modelIdentity string) *proto.TranscriptRequest {
	return &proto.TranscriptRequest{
		ModelIdentity:          modelIdentity,
		Dst:                    r.Audio,
		Language:               r.Language,
		Translate:              r.Translate,
		Diarize:                r.Diarize,
		Threads:                threads,
		Prompt:                 r.Prompt,
		Temperature:            r.Temperature,
		TimestampGranularities: r.TimestampGranularities,
	}
}

func loadTranscriptionModel(ctx context.Context, ml *model.ModelLoader, modelConfig config.ModelConfig, appConfig *config.ApplicationConfig) (grpcPkg.Backend, error) {
	if modelConfig.Backend == "" {
		modelConfig.Backend = model.WhisperBackend
	}
	// model.WithContext(ctx) overrides the app-context default set in
	// ModelOptions so distributed routing decisions reach the request's
	// X-LocalAI-Node holder via distributedhdr.Stamp.
	opts := ModelOptions(modelConfig, appConfig, model.WithContext(ctx))
	transcriptionModel, err := ml.Load(opts...)
	if err != nil {
		recordModelLoadFailure(appConfig, modelConfig.Name, modelConfig.Backend, err, nil)
		return nil, err
	}
	if transcriptionModel == nil {
		return nil, fmt.Errorf("could not load transcription model")
	}
	return transcriptionModel, nil
}

func ModelTranscription(ctx context.Context, audio, language string, translate, diarize bool, prompt string, ml *model.ModelLoader, modelConfig config.ModelConfig, appConfig *config.ApplicationConfig) (*schema.TranscriptionResult, error) {
	return ModelTranscriptionWithOptions(ctx, TranscriptionRequest{
		Audio:     audio,
		Language:  language,
		Translate: translate,
		Diarize:   diarize,
		Prompt:    prompt,
	}, ml, modelConfig, appConfig)
}

func ModelTranscriptionWithOptions(ctx context.Context, req TranscriptionRequest, ml *model.ModelLoader, modelConfig config.ModelConfig, appConfig *config.ApplicationConfig) (*schema.TranscriptionResult, error) {
	transcriptionModel, err := loadTranscriptionModel(ctx, ml, modelConfig, appConfig)
	if err != nil {
		return nil, err
	}
	release, err := AcquireGlobalBackendSlot()
	if err != nil {
		return nil, err
	}
	defer release()

	var startTime time.Time
	var traceID string
	var audioSnippet map[string]any
	if appConfig.EnableTracing {
		trace.InitBackendTracingIfEnabled(appConfig.TracingMaxItems, appConfig.TracingMaxBodyBytes)
		startTime = time.Now()
		traceID = trace.BeginBackendTrace(trace.BackendTrace{Timestamp: startTime, Type: trace.BackendTraceTranscription, ModelName: modelConfig.Name, Backend: modelConfig.Backend, Summary: trace.TruncateString(req.Audio, 200)})
		// Capture audio before the backend call — the backend may delete the file.
		audioSnippet = trace.AudioSnippet(req.Audio, appConfig.TracingMaxBodyBytes)
	}
	defer trace.CancelBackendTrace(traceID)

	r, err := transcriptionModel.AudioTranscription(ctx, req.toProto(uint32(*modelConfig.Threads), modelConfig.Model))
	if err != nil {
		if appConfig.EnableTracing {
			errData := map[string]any{
				"audio_file": req.Audio,
				"language":   req.Language,
				"translate":  req.Translate,
				"diarize":    req.Diarize,
				"prompt":     req.Prompt,
			}
			if audioSnippet != nil {
				maps.Copy(errData, audioSnippet)
			}
			trace.RecordBackendTrace(trace.BackendTrace{
				ID:        traceID,
				Timestamp: startTime,
				Duration:  time.Since(startTime),
				Type:      trace.BackendTraceTranscription,
				ModelName: modelConfig.Name,
				Backend:   modelConfig.Backend,
				Summary:   trace.TruncateString(req.Audio, 200),
				Error:     err.Error(),
				Data:      errData,
			})
		}
		return nil, err
	}
	tr := transcriptResultFromProto(r)

	if appConfig.EnableTracing {
		data := map[string]any{
			"audio_file":     req.Audio,
			"language":       req.Language,
			"translate":      req.Translate,
			"diarize":        req.Diarize,
			"prompt":         req.Prompt,
			"result_text":    tr.Text,
			"segments_count": len(tr.Segments),
		}
		if audioSnippet != nil {
			maps.Copy(data, audioSnippet)
		}
		trace.RecordBackendTrace(trace.BackendTrace{
			ID:        traceID,
			Timestamp: startTime,
			Duration:  time.Since(startTime),
			Type:      trace.BackendTraceTranscription,
			ModelName: modelConfig.Name,
			Backend:   modelConfig.Backend,
			Summary:   trace.TruncateString(req.Audio+" -> "+tr.Text, 200),
			Data:      data,
		})
	}

	return tr, err
}

// TranscriptionStreamChunk is a streaming event emitted by
// ModelTranscriptionStream. Either Delta carries an incremental text fragment,
// or Final carries the completed transcription as the very last event.
type TranscriptionStreamChunk struct {
	Delta string
	Final *schema.TranscriptionResult
}

// ModelTranscriptionStream runs the gRPC streaming transcription RPC and
// invokes onChunk for each event the backend produces. Backends that don't
// support real streaming should still emit one terminal event with Final set,
// which the HTTP layer turns into a single delta + done SSE pair.
func ModelTranscriptionStream(ctx context.Context, req TranscriptionRequest, ml *model.ModelLoader, modelConfig config.ModelConfig, appConfig *config.ApplicationConfig, onChunk func(TranscriptionStreamChunk)) error {
	transcriptionModel, err := loadTranscriptionModel(ctx, ml, modelConfig, appConfig)
	if err != nil {
		return err
	}

	pbReq := req.toProto(uint32(*modelConfig.Threads), modelConfig.Model)
	pbReq.Stream = true
	release, err := AcquireGlobalBackendSlot()
	if err != nil {
		return err
	}
	defer release()

	return transcriptionModel.AudioTranscriptionStream(ctx, pbReq, func(chunk *proto.TranscriptStreamResponse) {
		if chunk == nil {
			return
		}
		out := TranscriptionStreamChunk{Delta: chunk.Delta}
		if chunk.FinalResult != nil {
			out.Final = transcriptResultFromProto(chunk.FinalResult)
		}
		onChunk(out)
	})
}

func transcriptResultFromProto(r *proto.TranscriptResult) *schema.TranscriptionResult {
	if r == nil {
		return &schema.TranscriptionResult{}
	}
	tr := &schema.TranscriptionResult{
		Text:     r.Text,
		Language: r.Language,
		Duration: float64(r.Duration),
		Eou:      r.Eou,
	}

	for _, s := range r.Segments {
		var tks []int
		for _, t := range s.Tokens {
			tks = append(tks, int(t))
		}
		var words []schema.TranscriptionWord
		for _, w := range s.Words {
			var word = schema.TranscriptionWord{
				Start: time.Duration(w.Start),
				End:   time.Duration(w.End),
				Text:  w.Text,
			}
			words = append(words, word)
			tr.Words = append(tr.Words, word)
		}
		tr.Segments = append(tr.Segments,
			schema.TranscriptionSegment{
				Text:    s.Text,
				Id:      int(s.Id),
				Start:   time.Duration(s.Start),
				End:     time.Duration(s.End),
				Tokens:  tks,
				Speaker: s.Speaker,
				Words:   words,
			})
	}
	return tr
}