feat(multimodal): 视觉分层路由(多模态检测→主/视觉/OCR 工具)+ 语音 STT→TTS 降级链(异步)

This commit is contained in:
Aska committed 2026-09-06 05:12:36 +08:00
1 parent 02aae2bbb4
commit 0a936a3202
19 files changed
+2121

No files matched your search

+112
View File
@@ -0,0 +1,112 @@
package voice
import (
"context"
"fmt"
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
)
// Chain is a degrading provider list. STT and TTS walk the steps in order,
// returning the first successful result and moving to the next step when a
// provider is unavailable (nil direction) or fails. It implements STT and TTS so
// it can be used anywhere a single provider is expected.
type Chain struct {
steps []Provider
}
// NewChain builds a degrading chain from the given providers in the order they
// should be tried. Passing a SilentProvider last guarantees the chain always
// succeeds.
func NewChain(steps ...Provider) *Chain {
return &Chain{steps: steps}
}
// STT walks the chain for a transcription.
func (c *Chain) STT(ctx context.Context, audio Audio) (string, error) {
if len(c.steps) == 0 {
return "", ErrProviderFailed
}
var lastErr error
for _, p := range c.steps {
stt := p.STT()
if stt == nil {
continue
}
text, err := stt.STT(ctx, audio)
if err == nil {
return text, nil
}
lastErr = err
}
return "", fmt.Errorf("voice: %w: %v", ErrProviderFailed, lastErr)
}
// TTS walks the chain for synthesized audio.
func (c *Chain) TTS(ctx context.Context, text string) (Audio, error) {
if len(c.steps) == 0 {
return Audio{}, ErrProviderFailed
}
var lastErr error
for _, p := range c.steps {
tts := p.TTS()
if tts == nil {
continue
}
audio, err := tts.TTS(ctx, text)
if err == nil {
return audio, nil
}
lastErr = err
}
return Audio{}, fmt.Errorf("voice: %w: %v", ErrProviderFailed, lastErr)
}
// Option customizes a chain built by New.
type Option func(*chainConfig)
// chainConfig carries the assembled steps for New.
type chainConfig struct {
steps []Provider
}
// WithProviders overrides the provider steps assembled from config. Tests use it
// to inject deterministic mock providers instead of the real adaptation-point
// rungs.
func WithProviders(steps ...Provider) Option {
return func(c *chainConfig) { c.steps = steps }
}
// New assembles a chain from cfg using the fixed degradation order
// cloud → local → edge-tts/espeak → silent.
//
// - cloud is included only when a cloud voice setting is present (today an
// API key) so a mis-configured build degrades quietly.
// - local is a pure adaptation point and is only included when supplied via
// WithProviders; by default it is skipped.
// - the edge-tts or espeak rung is included according to cfg.TTSProvider.
// - silent is always appended last so the chain never fails outright.
//
// opts may supply a fully assembled step list to override this mapping.
func New(cfg config.VoiceConfig, opts ...Option) *Chain {
conf := &chainConfig{}
for _, o := range opts {
o(conf)
}
if len(conf.steps) > 0 {
return NewChain(conf.steps...)
}
steps := make([]Provider, 0, 5)
if cfg.DashscopeAPIKey != "" {
steps = append(steps, CloudProvider(newDashboardSTT(cfg), newDashboardTTS(cfg)))
}
switch cfg.TTSProvider {
case "edge-tts", "edge":
steps = append(steps, EdgeTTSProvider(newEdgeTTS()))
case "espeak":
steps = append(steps, EspeakProvider(newEspeak()))
}
steps = append(steps, SilentProvider())
return NewChain(steps...)
}
@@ -0,0 +1,186 @@
package voice
import (
"context"
"errors"
"testing"
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
)
// stubSTT is a mock STT returning a fixed result.
type stubSTT struct {
text string
err error
}
func (s *stubSTT) STT(_ context.Context, _ Audio) (string, error) { return s.text, s.err }
// stubTTS is a mock TTS returning a fixed result.
type stubTTS struct {
audio Audio
err error
}
func (s *stubTTS) TTS(_ context.Context, _ string) (Audio, error) { return s.audio, s.err }
// stubProvider is a mock Provider carrying optional directions.
type stubProvider struct {
name string
stt STT
tts TTS
}
func (s *stubProvider) Name() string { return s.name }
func (s *stubProvider) STT() STT { return s.stt }
func (s *stubProvider) TTS() TTS { return s.tts }
func TestChain_STT_Degrades(t *testing.T) {
t.Parallel()
tests := []struct {
name string
steps []Provider
want string
wantErr bool
}{
{
name: "first provider succeeds",
steps: []Provider{&stubProvider{name: "cloud", stt: &stubSTT{text: "hello"}}},
want: "hello",
},
{
name: "first fails second succeeds",
steps: []Provider{
&stubProvider{name: "cloud", stt: &stubSTT{err: ErrNotConfigured}},
&stubProvider{name: "local", stt: &stubSTT{text: "degraded"}},
},
want: "degraded",
},
{
name: "nil direction skipped",
steps: []Provider{
&stubProvider{name: "edge", stt: nil},
&stubProvider{name: "silent", stt: &stubSTT{text: "silence"}},
},
want: "silence",
},
{
name: "all fail no silent -> error",
steps: []Provider{
&stubProvider{name: "cloud", stt: &stubSTT{err: ErrNotConfigured}},
},
wantErr: true,
},
{
name: "all fail silent last -> no error",
steps: []Provider{
&stubProvider{name: "cloud", stt: &stubSTT{err: ErrNotConfigured}},
SilentProvider(),
},
want: "",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
got, err := NewChain(tt.steps...).STT(context.Background(), Audio{Format: "wav", Data: []byte("x")})
if tt.wantErr {
if !errors.Is(err, ErrProviderFailed) {
t.Fatalf("err = %v, want ErrProviderFailed", err)
}
return
}
if err != nil {
t.Fatalf("STT error: %v", err)
}
if got != tt.want {
t.Errorf("text = %q, want %q", got, tt.want)
}
})
}
}
func TestChain_TTS_Degrades(t *testing.T) {
t.Parallel()
tests := []struct {
name string
steps []Provider
want Audio
wantErr bool
}{
{
name: "first provider succeeds",
steps: []Provider{&stubProvider{name: "cloud", tts: &stubTTS{audio: Audio{Format: "mp3", Data: []byte("a")}}}},
want: Audio{Format: "mp3", Data: []byte("a")},
},
{
name: "first fails second succeeds",
steps: []Provider{
&stubProvider{name: "cloud", tts: &stubTTS{err: ErrNotConfigured}},
&stubProvider{name: "edge-tts", tts: &stubTTS{audio: Audio{Format: "mp3", Data: []byte("b")}}},
},
want: Audio{Format: "mp3", Data: []byte("b")},
},
{
name: "all fail no silent -> error",
steps: []Provider{
&stubProvider{name: "cloud", tts: &stubTTS{err: ErrNotConfigured}},
},
wantErr: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
got, err := NewChain(tt.steps...).TTS(context.Background(), "hi")
if tt.wantErr {
if !errors.Is(err, ErrProviderFailed) {
t.Fatalf("err = %v, want ErrProviderFailed", err)
}
return
}
if err != nil {
t.Fatalf("TTS error: %v", err)
}
if got.Format != tt.want.Format || string(got.Data) != string(tt.want.Data) {
t.Errorf("audio = %+v, want %+v", got, tt.want)
}
})
}
}
func TestChain_Empty(t *testing.T) {
t.Parallel()
if _, err := NewChain().STT(context.Background(), Audio{}); !errors.Is(err, ErrProviderFailed) {
t.Fatalf("err = %v, want ErrProviderFailed", err)
}
}
func TestNew_ConfigAssembly(t *testing.T) {
t.Parallel()
// With a cloud key and default edge-tts TTS, the chain starts at cloud and
// ends at silent.
c := New(config.VoiceConfig{DashscopeAPIKey: "k"})
names := make([]string, 0, len(c.steps))
for _, p := range c.steps {
names = append(names, p.Name())
}
if len(names) < 2 || names[len(names)-1] != "silent" {
t.Fatalf("steps = %v, want to end at silent", names)
}
if names[0] != "cloud" {
t.Errorf("first step = %q, want cloud", names[0])
}
// Without a key the cloud rung is skipped.
c2 := New(config.VoiceConfig{})
foundCloud := false
for _, p := range c2.steps {
if p.Name() == "cloud" {
foundCloud = true
}
}
if foundCloud {
t.Error("cloud step present without an API key")
}
}
+227
View File
@@ -0,0 +1,227 @@
package voice
import (
"bytes"
"context"
"encoding/base64"
"encoding/json"
"fmt"
"io"
"log/slog"
"net/http"
"strings"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
)
// cloudMaxResponseBytes caps how much of a cloud voice response is read, so an
// oversized or malformed response cannot exhaust memory. The request timeout
// bounds a single call.
const (
cloudMaxResponseBytes = 8 << 20
cloudHTTPTimeout = 180 * time.Second
)
// HTTPSTT transcribes audio by POSTing a JSON body to a cloud endpoint and
// reading {"text": "..."} back. It is the stubbed default STT implementation;
// the real DashScope/whisper-facing adapter is an adaptation point.
type HTTPSTT struct {
endpoint string
apiKey string
client *http.Client
log *slog.Logger
}
// HTTPTTS synthesizes speech by POSTing a JSON body and reading the returned
// base64 audio. It is the stubbed default TTS implementation.
type HTTPTTS struct {
endpoint string
apiKey string
client *http.Client
log *slog.Logger
}
// HTTPOption customizes the client and logger of an HTTPSTT/HTTPTTS.
type HTTPOption func(*httpConfig)
type httpConfig struct {
client *http.Client
log *slog.Logger
}
// WithHTTPClient sets the *http.Client used for the call; nil keeps the default.
func WithHTTPClient(c *http.Client) HTTPOption {
return func(o *httpConfig) {
if c != nil {
o.client = c
}
}
}
// WithLogger sets the logger used on failures; nil keeps a discard logger.
func WithLogger(l *slog.Logger) HTTPOption {
return func(o *httpConfig) {
if l != nil {
o.log = l
}
}
}
// NewHTTPSTT builds an HTTPSTT for the given endpoint and API key.
func NewHTTPSTT(endpoint, apiKey string, opts ...HTTPOption) *HTTPSTT {
h := &httpConfig{client: defaultHTTPClient(), log: discardLogger()}
for _, o := range opts {
o(h)
}
return &HTTPSTT{endpoint: strings.TrimRight(endpoint, "/"), apiKey: apiKey, client: h.client, log: h.log}
}
// NewHTTPTTS builds an HTTPTTS for the given endpoint and API key.
func NewHTTPTTS(endpoint, apiKey string, opts ...HTTPOption) *HTTPTTS {
h := &httpConfig{client: defaultHTTPClient(), log: discardLogger()}
for _, o := range opts {
o(h)
}
return &HTTPTTS{endpoint: strings.TrimRight(endpoint, "/"), apiKey: apiKey, client: h.client, log: h.log}
}
// sttRequest is the JSON body sent to the STT endpoint.
type sttRequest struct {
Audio string `json:"audio"`
Format string `json:"format"`
}
// sttResponse is the transcription returned by the STT endpoint.
type sttResponse struct {
Text string `json:"text"`
}
// STT posts the audio and returns the recognized text. A blank endpoint returns
// ErrNotConfigured so the chain degrades.
func (s *HTTPSTT) STT(ctx context.Context, audio Audio) (string, error) {
if s.endpoint == "" {
return "", ErrNotConfigured
}
if err := ctx.Err(); err != nil {
return "", err
}
payload, err := json.Marshal(sttRequest{Audio: base64.StdEncoding.EncodeToString(audio.Data), Format: audio.Format})
if err != nil {
return "", fmt.Errorf("voice: %w: marshal: %v", ErrProviderFailed, err)
}
resp, err := postJSON(ctx, s.client, s.apiKey, s.endpoint, payload)
if err != nil {
return "", err
}
defer func() { _ = resp.Body.Close() }()
var decoded sttResponse
if err := json.NewDecoder(resp.Body).Decode(&decoded); err != nil {
return "", fmt.Errorf("voice: %w: decode: %v", ErrProviderFailed, err)
}
return decoded.Text, nil
}
// ttsRequest is the JSON body sent to the TTS endpoint.
type ttsRequest struct {
Text string `json:"text"`
}
// ttsResponse is the synthesized audio returned by the TTS endpoint.
type ttsResponse struct {
Audio string `json:"audio"`
Format string `json:"format"`
}
// TTS posts the text and returns the decoded audio. A blank endpoint returns
// ErrNotConfigured so the chain degrades.
func (s *HTTPTTS) TTS(ctx context.Context, text string) (Audio, error) {
if s.endpoint == "" {
return Audio{}, ErrNotConfigured
}
if strings.TrimSpace(text) == "" {
return Audio{}, ErrEmptyText
}
if err := ctx.Err(); err != nil {
return Audio{}, err
}
payload, err := json.Marshal(ttsRequest{Text: text})
if err != nil {
return Audio{}, fmt.Errorf("voice: %w: marshal: %v", ErrProviderFailed, err)
}
resp, err := postJSON(ctx, s.client, s.apiKey, s.endpoint, payload)
if err != nil {
return Audio{}, err
}
defer func() { _ = resp.Body.Close() }()
var decoded ttsResponse
if err := json.NewDecoder(resp.Body).Decode(&decoded); err != nil {
return Audio{}, fmt.Errorf("voice: %w: decode: %v", ErrProviderFailed, err)
}
data, err := base64.StdEncoding.DecodeString(decoded.Audio)
if err != nil {
return Audio{}, fmt.Errorf("voice: %w: decode audio: %v", ErrProviderFailed, err)
}
format := decoded.Format
if format == "" {
format = "mp3"
}
return Audio{Format: format, Data: data}, nil
}
// postJSON POSTs a JSON payload, wires auth, and checks the status code.
func postJSON(ctx context.Context, client *http.Client, apiKey, url string, payload []byte) (*http.Response, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(payload))
if err != nil {
return nil, fmt.Errorf("voice: %w: build: %v", ErrProviderFailed, err)
}
req.Header.Set("Content-Type", "application/json")
if apiKey != "" {
req.Header.Set("Authorization", "Bearer "+apiKey)
}
resp, err := client.Do(req)
if err != nil {
return nil, fmt.Errorf("voice: %w: request: %v", ErrProviderFailed, err)
}
if resp.StatusCode != http.StatusOK {
_ = resp.Body.Close()
return nil, fmt.Errorf("voice: %w: status %d", ErrProviderFailed, resp.StatusCode)
}
return resp, nil
}
// defaultHTTPClient builds a client with a request timeout. A fresh client is
// returned per call rather than a package global, which the constraints forbid.
func defaultHTTPClient() *http.Client {
return &http.Client{Transport: http.DefaultTransport, Timeout: cloudHTTPTimeout}
}
// discardLogger returns a logger that drops all output.
func discardLogger() *slog.Logger {
return slog.New(slog.NewTextHandler(io.Discard, nil))
}
// newDashboardSTT returns the cloud STT for a config. The endpoint is an
// adaptation point and is empty until config gains a voice URL; a call then
// returns ErrNotConfigured and the chain degrades.
func newDashboardSTT(cfg config.VoiceConfig) STT {
return NewHTTPSTT("", cfg.DashscopeAPIKey)
}
// newDashboardTTS returns the cloud TTS for a config.
func newDashboardTTS(cfg config.VoiceConfig) TTS {
return NewHTTPTTS("", cfg.DashscopeAPIKey)
}
// newEdgeTTS returns the edge-tts adapter point (not yet implemented).
func newEdgeTTS() TTS { return notImplementedTTS{} }
// newEspeak returns the espeak adapter point (not yet implemented).
func newEspeak() TTS { return notImplementedTTS{} }
// notImplementedTTS is an adaptation point that always degrades.
type notImplementedTTS struct{}
func (notImplementedTTS) TTS(_ context.Context, _ string) (Audio, error) {
return Audio{}, ErrNotImplemented
}
@@ -0,0 +1,81 @@
package voice
import (
"context"
"encoding/base64"
"encoding/hex"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
)
func TestHTTPSTT(t *testing.T) {
t.Parallel()
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var body map[string]any
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
t.Fatalf("decode: %v", err)
}
if r.Header.Get("Authorization") != "Bearer sk" {
t.Errorf("auth = %q, want Bearer sk", r.Header.Get("Authorization"))
}
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"text":"hello world"}`))
}))
defer srv.Close()
stt := NewHTTPSTT(srv.URL, "sk")
text, err := stt.STT(context.Background(), Audio{Format: "wav", Data: []byte("raw")})
if err != nil {
t.Fatalf("STT error: %v", err)
}
if text != "hello world" {
t.Errorf("text = %q, want hello world", text)
}
}
func TestHTTPSTT_NotConfigured(t *testing.T) {
t.Parallel()
stt := NewHTTPSTT("", "")
if _, err := stt.STT(context.Background(), Audio{}); err != ErrNotConfigured {
t.Fatalf("err = %v, want ErrNotConfigured", err)
}
}
func TestHTTPTTS(t *testing.T) {
t.Parallel()
audioBytes := []byte{1, 2, 3}
b64 := base64.StdEncoding.EncodeToString(audioBytes)
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var body map[string]any
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
t.Fatalf("decode: %v", err)
}
if body["text"] != "hi" {
t.Errorf("text = %v, want hi", body["text"])
}
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"audio":"` + b64 + `","format":"wav"}`))
}))
defer srv.Close()
tts := NewHTTPTTS(srv.URL, "")
audio, err := tts.TTS(context.Background(), "hi")
if err != nil {
t.Fatalf("TTS error: %v", err)
}
if audio.Format != "wav" || hex.EncodeToString(audio.Data) != hex.EncodeToString(audioBytes) {
t.Errorf("audio = %+v, want wav/010203", audio)
}
}
func TestHTTPTTS_EmptyText(t *testing.T) {
t.Parallel()
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
defer srv.Close()
tts := NewHTTPTTS(srv.URL, "")
if _, err := tts.TTS(context.Background(), " "); err != ErrEmptyText {
t.Fatalf("err = %v, want ErrEmptyText", err)
}
}
@@ -0,0 +1,18 @@
package voice
import "errors"
// Sentinel errors returned by the voice package. Each is wrapped with %w so
// callers can test for a category while keeping the cause.
var (
// ErrProviderFailed is returned when every provider in a chain fails.
ErrProviderFailed = errors.New("voice: all providers failed")
// ErrNotConfigured is returned when a provider has no endpoint/setting.
ErrNotConfigured = errors.New("voice: provider not configured")
// ErrNotImplemented is returned by an adaptation-point provider.
ErrNotImplemented = errors.New("voice: provider not implemented")
// ErrEmptyText is returned when TTS receives empty text.
ErrEmptyText = errors.New("voice: empty text")
// ErrClosed is returned when a service call arrives after Close.
ErrClosed = errors.New("voice: service closed")
)
@@ -0,0 +1,79 @@
package voice
import "context"
// Provider is one rung in the voice degradation chain. It may implement zero,
// one, or both directions; a nil return means that direction is not available
// on this rung and the chain moves to the next.
type Provider interface {
// Name returns a stable provider identifier.
Name() string
// STT returns the speech-to-text implementation, or nil when the provider
// cannot transcribe.
STT() STT
// TTS returns the text-to-speech implementation, or nil when the provider
// cannot synthesize.
TTS() TTS
}
// StaticProvider is a Provider with a fixed name whose STT/TTS are set at
// construction. It is used by the concrete providers below.
type StaticProvider struct {
name string
stt STT
tts TTS
}
// Name returns the provider identifier.
func (p *StaticProvider) Name() string { return p.name }
// STT returns the configured STT, or nil.
func (p *StaticProvider) STT() STT { return p.stt }
// TTS returns the configured TTS, or nil.
func (p *StaticProvider) TTS() TTS { return p.tts }
// NewProvider builds a Provider with the given name and optional directions it
// supports.
func NewProvider(name string, stt STT, tts TTS) *StaticProvider {
return &StaticProvider{name: name, stt: stt, tts: tts}
}
// CloudProvider is the cloud rung of the chain. It is an adaptation point: the
// real STT/TTS implementations are wired in via NewProvider with cloud adapters.
func CloudProvider(stt STT, tts TTS) *StaticProvider {
return NewProvider("cloud", stt, tts)
}
// LocalProvider is the local-model rung of the chain.
func LocalProvider(stt STT, tts TTS) *StaticProvider {
return NewProvider("local", stt, tts)
}
// EdgeTTSProvider is the edge-tts rung (text-to-speech only).
func EdgeTTSProvider(tts TTS) *StaticProvider {
return NewProvider("edge-tts", nil, tts)
}
// EspeakProvider is the espeak rung (text-to-speech only).
func EspeakProvider(tts TTS) *StaticProvider {
return NewProvider("espeak", nil, tts)
}
// SilentProvider is the terminal rung. It always succeeds, returning an empty
// transcription and empty audio, so a chain that includes it never errors.
func SilentProvider() *StaticProvider {
return NewProvider("silent", silenceSTT{}, silenceTTS{})
}
// silenceSTT is the silent STT rung; it returns an empty transcription.
type silenceSTT struct{}
func (silenceSTT) STT(_ context.Context, _ Audio) (string, error) { return "", nil }
// silenceTTS is the silent TTS rung; it returns empty audio.
type silenceTTS struct{}
func (silenceTTS) TTS(_ context.Context, _ string) (Audio, error) {
return Audio{}, nil
}
@@ -0,0 +1,160 @@
package voice
import (
"context"
"fmt"
"sync"
)
// Service runs STT/TTS requests on a background worker pool so caller
// goroutines (the agent loop) are never blocked by provider latency. Each
// request is enqueued on a jobs channel and answered on a per-job result
// channel; a canceled caller context or a closed service unblocks the caller
// immediately and no worker goroutine is leaked. It is safe for concurrent use.
type Service struct {
chain *Chain
jobs chan job
bgCtx context.Context
cancel context.CancelFunc
nWorkers int
wg sync.WaitGroup
mu sync.RWMutex
closed bool
}
// ServiceOption customizes a Service.
type ServiceOption func(*Service)
// WithWorkers sets the worker-pool size. Values <= 0 are ignored and the
// default single worker is kept.
func WithWorkers(n int) ServiceOption {
return func(s *Service) {
if n > 0 {
s.nWorkers = n
}
}
}
// jobKind selects the operation a worker performs.
type jobKind int
const (
jobSTT jobKind = iota
jobTTS
)
// job is one request enqueued by the caller.
type job struct {
kind jobKind
audio Audio
text string
res chan jobResult
}
// jobResult carries the outcome written back to the caller.
type jobResult struct {
text string
audio Audio
err error
}
// NewService builds a Service backed by chain and starts the worker pool.
func NewService(chain *Chain, opts ...ServiceOption) *Service {
bg, cancel := context.WithCancel(context.Background())
s := &Service{
chain: chain,
jobs: make(chan job),
bgCtx: bg,
cancel: cancel,
nWorkers: 1,
}
for _, o := range opts {
o(s)
}
for i := 0; i < s.nWorkers; i++ {
s.wg.Add(1)
go s.worker()
}
return s
}
// worker consumes jobs until the service closes.
func (s *Service) worker() {
defer s.wg.Done()
for {
select {
case <-s.bgCtx.Done():
return
case j := <-s.jobs:
s.run(j)
}
}
}
// run executes a job and delivers the result. Providers run under the service's
// background context so Close cancels in-flight work and no worker goroutine is
// leaked; the result is written to a buffered channel so a caller that already
// returned never blocks the worker.
func (s *Service) run(j job) {
res := jobResult{}
switch j.kind {
case jobSTT:
res.text, res.err = s.chain.STT(s.bgCtx, j.audio)
case jobTTS:
res.audio, res.err = s.chain.TTS(s.bgCtx, j.text)
}
j.res <- res
}
// STT enqueues a transcription request and waits for the result. ctx cancels
// the wait and the underlying provider call.
func (s *Service) STT(ctx context.Context, audio Audio) (string, error) {
res, err := s.submit(ctx, job{kind: jobSTT, audio: audio, res: make(chan jobResult, 1)})
if err != nil {
return "", err
}
return res.text, res.err
}
// TTS enqueues a synthesis request and waits for the result. ctx cancels the
// wait and the underlying provider call.
func (s *Service) TTS(ctx context.Context, text string) (Audio, error) {
res, err := s.submit(ctx, job{kind: jobTTS, text: text, res: make(chan jobResult, 1)})
if err != nil {
return Audio{}, err
}
return res.audio, res.err
}
// submit enqueues j and blocks until the result or a cancellation.
func (s *Service) submit(ctx context.Context, j job) (jobResult, error) {
select {
case s.jobs <- j:
case <-ctx.Done():
return jobResult{}, fmt.Errorf("voice: %w", ctx.Err())
case <-s.bgCtx.Done():
return jobResult{}, ErrClosed
}
select {
case r := <-j.res:
return r, nil
case <-ctx.Done():
return jobResult{}, fmt.Errorf("voice: %w", ctx.Err())
case <-s.bgCtx.Done():
return jobResult{}, ErrClosed
}
}
// Close cancels the worker pool and waits for in-flight jobs to finish. It is
// safe to call once; later calls are no-ops. In-flight provider calls are
// bounded by each provider's own context timeout, so Close returns promptly.
func (s *Service) Close() {
s.mu.Lock()
defer s.mu.Unlock()
if s.closed {
return
}
s.closed = true
s.cancel()
s.wg.Wait()
}
@@ -0,0 +1,108 @@
package voice
import (
"context"
"errors"
"sync"
"testing"
"time"
)
func TestService_STTAndTTS(t *testing.T) {
t.Parallel()
chain := NewChain(
NewProvider("cloud", &stubSTT{text: "transcribed"}, &stubTTS{audio: Audio{Format: "mp3", Data: []byte("a")}}),
)
svc := NewService(chain)
defer svc.Close()
text, err := svc.STT(context.Background(), Audio{Format: "wav", Data: []byte("x")})
if err != nil {
t.Fatalf("STT error: %v", err)
}
if text != "transcribed" {
t.Errorf("text = %q, want transcribed", text)
}
audio, err := svc.TTS(context.Background(), "hello")
if err != nil {
t.Fatalf("TTS error: %v", err)
}
if string(audio.Data) != "a" || audio.Format != "mp3" {
t.Errorf("audio = %+v, want mp3/a", audio)
}
}
func TestService_CloseStopsNewCalls(t *testing.T) {
t.Parallel()
svc := NewService(NewChain(SilentProvider()))
svc.Close()
_, err := svc.STT(context.Background(), Audio{})
if !errors.Is(err, ErrClosed) {
t.Fatalf("err = %v, want ErrClosed", err)
}
}
// blockingSTT blocks until its context is canceled, marking that it started.
type blockingSTT struct {
started chan struct{}
once sync.Once
}
func (b *blockingSTT) STT(ctx context.Context, _ Audio) (string, error) {
b.once.Do(func() { close(b.started) })
<-ctx.Done()
return "", ctx.Err()
}
func TestService_CloseCancelsInFlightWorker(t *testing.T) {
t.Parallel()
blk := &blockingSTT{started: make(chan struct{})}
chain := NewChain(NewProvider("cloud", blk, nil))
svc := NewService(chain)
done := make(chan struct{})
go func() {
defer close(done)
_, _ = svc.STT(context.Background(), Audio{Format: "wav", Data: []byte("x")})
}()
select {
case <-blk.started:
case <-time.After(2 * time.Second):
t.Fatal("provider never started")
}
svc.Close()
select {
case <-done:
case <-time.After(2 * time.Second):
t.Fatal("STT call did not return after Close; worker likely leaked")
}
}
func TestService_CallerCancelUnblocks(t *testing.T) {
t.Parallel()
blk := &blockingSTT{started: make(chan struct{})}
svc := NewService(NewChain(NewProvider("cloud", blk, nil)))
defer svc.Close()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
done := make(chan struct{})
go func() {
defer close(done)
_, _ = svc.STT(ctx, Audio{Format: "wav", Data: []byte("x")})
}()
<-blk.started
cancel()
select {
case <-done:
case <-time.After(2 * time.Second):
t.Fatal("caller cancellation did not unblock STT")
}
}
@@ -0,0 +1,35 @@
// Package voice implements asynchronous speech-to-text and text-to-speech for
// the Cyrene runtime, following DevDocs/design/06-多模态.md §1.
//
// Voice input (STT) is transcribed to text and fed into the same agent pipeline
// as a chat message; the reply is synthesized (TTS) back to audio. Both
// directions degrade through a provider chain — cloud → local → edge-tts/espeak
// → silent — so an unavailable provider falls through to the next rung instead
// of failing the whole turn. The stub default implementations are HTTP-based and
// testable with a stdlib httptest server; real cloud/local providers remain
// adaptation points for later milestones.
package voice
import "context"
// Audio is a piece of audio data plus its encoding.
type Audio struct {
// Format is the encoding, e.g. "wav", "mp3", "ogg".
Format string
// Data is the raw audio bytes.
Data []byte
}
// STT converts audio to text.
type STT interface {
// STT transcribes audio and returns the spoken text. ctx controls
// cancellation and must be honored.
STT(ctx context.Context, audio Audio) (string, error)
}
// TTS synthesizes speech audio from text.
type TTS interface {
// TTS synthesizes audio from text. ctx controls cancellation and must be
// honored.
TTS(ctx context.Context, text string) (Audio, error)
}