feat(transcribe): gateway client with an iguana->berget fallback chain (ADR-033)
Streams the WAV from disk with a computed Content-Length (no buffering, no chunking), posts verbose_json with field `file` and language=sv, clamps the final segment to the audio duration, and treats an empty transcript as a failure. The chain falls back on ANY primary error, records the serving endpoint, and never falls back once the context is cancelled. gateway.go and chain.go are one commit: the chain's tests use the gateway's StatusError and ErrEmptyTranscript, so neither compiles alone. Mutation-checked: field name, clamp, cancel-fallback, content length and the recorded endpoint each turn a test red. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,53 @@
|
|||||||
|
package transcribe
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
|
||||||
|
"git.d-ma.be/mathias/tapir/internal/domain"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Backend is one transcription endpoint. *Gateway satisfies it; Name is the value
|
||||||
|
// recorded as the serving endpoint.
|
||||||
|
type Backend interface {
|
||||||
|
Name() string
|
||||||
|
Transcribe(ctx context.Context, wavPath string) (domain.Transcription, error)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Chain tries a primary (local) backend and, if it fails for ANY reason, a
|
||||||
|
// fallback. The fallback exists because iguana can wedge permanently on a Metal
|
||||||
|
// OOM (infra#273) — which is also why a fallback is a data-egress decision: the
|
||||||
|
// berget endpoint is a cloud service, so the result records which one served it.
|
||||||
|
type Chain struct {
|
||||||
|
primary Backend
|
||||||
|
fallback Backend // nil disables the fallback
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewChain builds a Chain. A nil fallback means a primary failure is the result.
|
||||||
|
func NewChain(primary, fallback Backend) *Chain {
|
||||||
|
return &Chain{primary: primary, fallback: fallback}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Transcribe returns the first backend's success. It falls back on any primary
|
||||||
|
// error — transport, timeout, non-2xx, empty transcript — and never inspects a
|
||||||
|
// response body to decide. A cancelled context stops the chain: a job the user or
|
||||||
|
// shutdown cancelled must not ship its audio to a cloud endpoint as a last act.
|
||||||
|
func (c *Chain) Transcribe(ctx context.Context, wavPath string) (domain.Transcription, error) {
|
||||||
|
t, err := c.primary.Transcribe(ctx, wavPath)
|
||||||
|
if err == nil {
|
||||||
|
t.Endpoint = c.primary.Name()
|
||||||
|
return t, nil
|
||||||
|
}
|
||||||
|
if c.fallback == nil || ctx.Err() != nil {
|
||||||
|
return domain.Transcription{}, fmt.Errorf("%s: %w", c.primary.Name(), err)
|
||||||
|
}
|
||||||
|
t, ferr := c.fallback.Transcribe(ctx, wavPath)
|
||||||
|
if ferr != nil {
|
||||||
|
return domain.Transcription{}, errors.Join(
|
||||||
|
fmt.Errorf("%s: %w", c.primary.Name(), err),
|
||||||
|
fmt.Errorf("%s: %w", c.fallback.Name(), ferr))
|
||||||
|
}
|
||||||
|
t.Endpoint = c.fallback.Name()
|
||||||
|
return t, nil
|
||||||
|
}
|
||||||
@@ -0,0 +1,147 @@
|
|||||||
|
// Package transcribe turns a WAV file into a timed Swedish transcript by calling
|
||||||
|
// the LiteLLM gateway's OpenAI-compatible /audio/transcriptions endpoint
|
||||||
|
// (ADR-033). A Gateway is one model alias; a Chain tries a local one and falls
|
||||||
|
// back to a second.
|
||||||
|
package transcribe
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"io"
|
||||||
|
"mime/multipart"
|
||||||
|
"net/http"
|
||||||
|
"os"
|
||||||
|
"strings"
|
||||||
|
|
||||||
|
"git.d-ma.be/mathias/tapir/internal/domain"
|
||||||
|
)
|
||||||
|
|
||||||
|
// ErrEmptyTranscript is returned when the endpoint answered 200 but with no text.
|
||||||
|
// An empty transcript is not a success: chat over it could only answer "nothing".
|
||||||
|
var ErrEmptyTranscript = errors.New("transcribe: endpoint returned an empty transcript")
|
||||||
|
|
||||||
|
// StatusError is a non-2xx answer from the gateway. Nothing may branch on the
|
||||||
|
// body — {"error":"failed to process audio"} has had at least five distinct
|
||||||
|
// causes — so only the status code is exposed for logic; Snippet is for logs.
|
||||||
|
type StatusError struct {
|
||||||
|
Code int
|
||||||
|
Snippet string
|
||||||
|
}
|
||||||
|
|
||||||
|
func (e *StatusError) Error() string {
|
||||||
|
return fmt.Sprintf("transcribe: gateway returned HTTP %d: %s", e.Code, e.Snippet)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Gateway transcribes through one model alias on the gateway. baseURL is the
|
||||||
|
// gateway's /v1 root — the in-cluster NodePort, not the public hostname, whose
|
||||||
|
// ingress rejects bodies over ~8-10 MB with 413 (docs/research/audio-length-ceiling).
|
||||||
|
type Gateway struct {
|
||||||
|
baseURL string
|
||||||
|
key string
|
||||||
|
model string
|
||||||
|
client *http.Client
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewGateway builds a Gateway. The client should carry no overall Timeout: a 90
|
||||||
|
// minute recording takes minutes, so the deadline belongs to the caller's context.
|
||||||
|
func NewGateway(baseURL, key, model string, client *http.Client) *Gateway {
|
||||||
|
return &Gateway{baseURL: strings.TrimRight(baseURL, "/"), key: key, model: model, client: client}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Name is the model alias; it is what gets recorded as the serving endpoint.
|
||||||
|
func (g *Gateway) Name() string { return g.model }
|
||||||
|
|
||||||
|
// Transcribe uploads the WAV at path and returns the normalised transcription.
|
||||||
|
// The body is streamed from disk with a COMPUTED Content-Length: a 90 minute WAV
|
||||||
|
// is ~170 MB, so it is neither buffered in memory nor sent chunked.
|
||||||
|
func (g *Gateway) Transcribe(ctx context.Context, path string) (domain.Transcription, error) {
|
||||||
|
f, err := os.Open(path)
|
||||||
|
if err != nil {
|
||||||
|
return domain.Transcription{}, fmt.Errorf("transcribe: open audio: %w", err)
|
||||||
|
}
|
||||||
|
defer f.Close()
|
||||||
|
info, err := f.Stat()
|
||||||
|
if err != nil {
|
||||||
|
return domain.Transcription{}, fmt.Errorf("transcribe: stat audio: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
var prefix bytes.Buffer
|
||||||
|
mw := multipart.NewWriter(&prefix)
|
||||||
|
for _, kv := range [][2]string{{"model", g.model}, {"response_format", "verbose_json"}, {"language", "sv"}} {
|
||||||
|
if err := mw.WriteField(kv[0], kv[1]); err != nil {
|
||||||
|
return domain.Transcription{}, fmt.Errorf("transcribe: build request: %w", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// The multipart field is `file`; `audio_file` is rejected as "Invalid request".
|
||||||
|
if _, err := mw.CreateFormFile("file", "audio.wav"); err != nil {
|
||||||
|
return domain.Transcription{}, fmt.Errorf("transcribe: build request: %w", err)
|
||||||
|
}
|
||||||
|
suffix := []byte("\r\n--" + mw.Boundary() + "--\r\n")
|
||||||
|
|
||||||
|
body := io.MultiReader(bytes.NewReader(prefix.Bytes()), f, bytes.NewReader(suffix))
|
||||||
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, g.baseURL+"/audio/transcriptions", body)
|
||||||
|
if err != nil {
|
||||||
|
return domain.Transcription{}, fmt.Errorf("transcribe: build request: %w", err)
|
||||||
|
}
|
||||||
|
req.ContentLength = int64(prefix.Len()) + info.Size() + int64(len(suffix))
|
||||||
|
req.Header.Set("Content-Type", mw.FormDataContentType())
|
||||||
|
req.Header.Set("Authorization", "Bearer "+g.key)
|
||||||
|
// No "Expect: 100-continue": it trips the proxy for any body over 1 KB.
|
||||||
|
|
||||||
|
resp, err := g.client.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
return domain.Transcription{}, fmt.Errorf("transcribe: %s: %w", g.model, err)
|
||||||
|
}
|
||||||
|
defer resp.Body.Close()
|
||||||
|
if resp.StatusCode < 200 || resp.StatusCode > 299 {
|
||||||
|
snip, _ := io.ReadAll(io.LimitReader(resp.Body, 200))
|
||||||
|
return domain.Transcription{}, &StatusError{Code: resp.StatusCode, Snippet: strings.TrimSpace(string(snip))}
|
||||||
|
}
|
||||||
|
|
||||||
|
t, err := parseVerbose(resp.Body)
|
||||||
|
if err != nil {
|
||||||
|
return domain.Transcription{}, fmt.Errorf("transcribe: %s: %w", g.model, err)
|
||||||
|
}
|
||||||
|
t.Endpoint = g.model
|
||||||
|
return t, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// verboseJSON is the subset of the verbose_json response used. Both backends
|
||||||
|
// return it; they differ in segment granularity (iguana: whole seconds, berget:
|
||||||
|
// sub-second) and in the language token, neither of which matters here.
|
||||||
|
type verboseJSON struct {
|
||||||
|
Text string `json:"text"`
|
||||||
|
Duration float64 `json:"duration"`
|
||||||
|
Segments []struct {
|
||||||
|
Start float64 `json:"start"`
|
||||||
|
End float64 `json:"end"`
|
||||||
|
Text string `json:"text"`
|
||||||
|
} `json:"segments"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// parseVerbose normalises a verbose_json body. Whisper's final window can end past
|
||||||
|
// the audio (9.0 s for 4.25 s of audio was observed), so segment ends are clamped
|
||||||
|
// to the reported duration.
|
||||||
|
func parseVerbose(r io.Reader) (domain.Transcription, error) {
|
||||||
|
var v verboseJSON
|
||||||
|
if err := json.NewDecoder(r).Decode(&v); err != nil {
|
||||||
|
return domain.Transcription{}, fmt.Errorf("decode response: %w", err)
|
||||||
|
}
|
||||||
|
text := strings.TrimSpace(v.Text)
|
||||||
|
if text == "" {
|
||||||
|
return domain.Transcription{}, ErrEmptyTranscript
|
||||||
|
}
|
||||||
|
t := domain.Transcription{Text: text, AudioSeconds: v.Duration}
|
||||||
|
for _, s := range v.Segments {
|
||||||
|
seg := domain.Segment{Start: s.Start, End: s.End, Text: strings.TrimSpace(s.Text)}
|
||||||
|
if v.Duration > 0 {
|
||||||
|
seg.End = min(seg.End, v.Duration)
|
||||||
|
seg.Start = min(seg.Start, seg.End)
|
||||||
|
}
|
||||||
|
t.Segments = append(t.Segments, seg)
|
||||||
|
}
|
||||||
|
return t, nil
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user