cmd/tapir/uploads.go

@@ -7,10 +7,13 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
+	"time"
 
 	"git.d-ma.be/mathias/tapir/internal/adapters/audio"
+	"git.d-ma.be/mathias/tapir/internal/adapters/llm"
 	"git.d-ma.be/mathias/tapir/internal/adapters/transcribe"
 	"git.d-ma.be/mathias/tapir/internal/config"
+	"git.d-ma.be/mathias/tapir/internal/reconcile"
 	"git.d-ma.be/mathias/tapir/internal/uploads"
 	"git.d-ma.be/mathias/tapir/internal/web"
 )
@@ -52,16 +55,7 @@ func buildUploads(ctx context.Context, cfg config.Config, st uploadsStore, log *
 		log.Warn("uploads: marked interrupted by a restart (audio is not retained; they must be re-uploaded)", "count", n)
 	}
 
-	// No overall client timeout: a 90 minute recording takes minutes. The deadline
-	// is the worker's per-job context.
-	client := &http.Client{}
-	primary := transcribe.NewGateway(cfg.GatewayURL, cfg.GatewayKey, cfg.TranscribeModel, client)
-	var fallback transcribe.Backend
-	if cfg.TranscribeFallbackModel != "" && cfg.TranscribeFallbackModel != cfg.TranscribeModel {
-		fallback = transcribe.NewGateway(cfg.GatewayURL, cfg.GatewayKey, cfg.TranscribeFallbackModel, client)
-	}
-
-	worker := uploads.NewWorker(st, &audio.FFmpeg{Bin: cfg.FFmpegBin}, transcribe.NewChain(primary, fallback),
+	worker := uploads.NewWorker(st, &audio.FFmpeg{Bin: cfg.FFmpegBin}, buildTranscriber(cfg),
 		cfg.UploadTmpDir, cfg.TranscribeTimeout)
 	intake := uploads.NewIntake(st, &audio.FFprobe{Bin: cfg.FFprobeBin}, worker,
 		uploadLimits(cfg), cfg.UploadTmpDir)
@@ -98,3 +92,39 @@ func wireUploads(app *web.App, cfg config.Config, in *uploads.Intake) {
 	app.UploadChatMaxChars = cfg.MaxUploadedTranscriptChars
 	app.UploadChatModel = cfg.UploadChatModel
 }
+
+// adjudicatorTimeout bounds one adjudication call. A call takes ~2 s on gemma4-26b and ~10 s
+// on 31b; this is the wait before an unresponsive model is counted as an error.
+const adjudicatorTimeout = 2 * time.Minute
+
+// buildTranscriber builds what the worker transcribes with. By default that is the
+// single-engine chain (the primary, then the optional cloud fallback). When
+// TAPIR_RECONCILE_BASE_MODEL is set it is the two-engine ensemble (ADR-034): the base
+// engine and the primary run on the same WAV and a LOCAL model reconciles them. Config
+// has already refused any non-local model, so nothing here can send private audio or
+// text to a cloud model for reconciliation; the cloud fallback is unchanged.
+func buildTranscriber(cfg config.Config) uploads.Transcriber {
+	// No overall client timeout: a 90 minute recording takes minutes. The deadline is
+	// the worker's per-job context.
+	client := &http.Client{}
+	primary := transcribe.NewGateway(cfg.GatewayURL, cfg.GatewayKey, cfg.TranscribeModel, client)
+	var fallback transcribe.Backend
+	if cfg.TranscribeFallbackModel != "" && cfg.TranscribeFallbackModel != cfg.TranscribeModel {
+		fallback = transcribe.NewGateway(cfg.GatewayURL, cfg.GatewayKey, cfg.TranscribeFallbackModel, client)
+	}
+	if cfg.ReconcileBaseModel == "" {
+		return transcribe.NewChain(primary, fallback)
+	}
+
+	base := transcribe.NewGateway(cfg.GatewayURL, cfg.GatewayKey, cfg.ReconcileBaseModel, client)
+	// Temperature 0 and a short reply: the adjudicator's accuracy was measured that way.
+	chat := llm.New(cfg.GatewayURL, cfg.GatewayKey, cfg.ReconcileModel, adjudicatorTimeout,
+		llm.WithTemperature(0), llm.WithMaxTokens(200))
+	var adj reconcile.Adjudicator = reconcile.NewLLMAdjudicator(chat)
+	if cfg.ReconcileCheckOrder {
+		adj = reconcile.NewOrderChecked(adj)
+	}
+	return transcribe.NewEnsemble(base, primary, fallback, adj, reconcile.Options{
+		BaseEngine: cfg.ReconcileBaseModel, AltEngine: cfg.TranscribeModel, Reconciler: cfg.ReconcileModel,
+	})
+}