From 1347a77d33b817e57a252cd21b64974fe7272b51 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Tue, 27 Oct 2026 21:22:14 +0800 Subject: [PATCH] feat(loop): wire speak tool into the loop (M57) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - speak_tool.go: the speak tool itself — rune-derived timeout budget, structured failure codes, persona/emotion stamped on the synthesis ctx, audio persisted through the store seam, structured result for the model. - voice_deps.go: the wiring seam (VoiceSynth / VoiceStore + VoiceStoreFunc adapter), voiceEnabled, withSpeakSpec (appended after every schema shrink), and captureVoice (success beats failure, last one wins). - options.go / loop.go: VoiceSynth, VoiceStore, SpeakBudget injection points. - request.go: advertise speak on both toolSchemas return paths. - tools.go / tools_invoke.go: carry the batch voice outcome; split invokeOne / emitToolExecution / spanStatusForToolPhase / invokeParallel into their own file (tools.go was at the 500-line ceiling) and let speak run without a dispatcher. - run.go: keep the voice outcome across tool batches (a later batch without speak must not erase it). Known, deliberate: speak bypasses the capacity allow-set and skill-round narrowing (it belongs to no built-in toolset), so it is also advertised in quiet/awake; execution still passes the tool security gate, where speak is not registered and non-owner actors are denied fail-closed. --- internal/runtime/loop/loop.go | 11 +- internal/runtime/loop/options.go | 9 + internal/runtime/loop/request.go | 4 +- internal/runtime/loop/run.go | 5 + internal/runtime/loop/speak_tool.go | 336 ++++++++++++++++++++++++++ internal/runtime/loop/tools.go | 98 +------- internal/runtime/loop/tools_invoke.go | 107 ++++++++ internal/runtime/loop/voice_deps.go | 152 ++++++++++++ 8 files changed, 628 insertions(+), 94 deletions(-) create mode 100644 internal/runtime/loop/speak_tool.go create mode 100644 internal/runtime/loop/tools_invoke.go create mode 100644 internal/runtime/loop/voice_deps.go diff --git a/internal/runtime/loop/loop.go b/internal/runtime/loop/loop.go index 9afd957..cbe5124 100644 --- a/internal/runtime/loop/loop.go +++ b/internal/runtime/loop/loop.go @@ -143,7 +143,13 @@ type Loop struct { // 逐字节一致。它由装配期一次性挂上，之后只读；每会话的状态在 SkillFace 内部 // 按会话分桶并自带锁，因此循环的并发 Run 契约不变。 skillFace *SkillFace - log *slog.Logger + // voiceSynth / voiceStore 是 speak 的注入依赖（M57）。两者都非 nil 时 + // 循环向模型广告 speak 并自行承载它的执行（见 voice_deps.go）； + // speakBudget 是超时预算的替换公式，nil 用内置公式。 + voiceSynth VoiceSynth + voiceStore VoiceStore + speakBudget func(int) int + log *slog.Logger } // New 由 opts 构建 Loop 并应用默认值。MaxRounds 非正时 @@ -216,6 +222,9 @@ func New(opts Options) *Loop { imageResolver: opts.Vision, documentResolver: opts.DocumentResolver, documentInjectChars: docInjectChars, + voiceSynth: opts.VoiceSynth, + voiceStore: opts.VoiceStore, + speakBudget: opts.SpeakBudget, log: sourcedLog(opts.Log), } } diff --git a/internal/runtime/loop/options.go b/internal/runtime/loop/options.go index 2ff8d0f..61f95c3 100644 --- a/internal/runtime/loop/options.go +++ b/internal/runtime/loop/options.go @@ -214,6 +214,15 @@ type Options struct { // 或重排模型所见的消息。它是循环中与 // OpenClaw 的 transformContext 等价之物（design §1.3）。为 nil = 不做投影。 TransformContext func(ctx context.Context, messages []model.Message, round int) []model.Message + // VoiceSynth 与 VoiceStore 一起接线时（**两个都非 nil**），循环向模型 + // 广告 speak 工具（M57）：她可以自己决定这一条回复用语音还是文本。 + // 任一为 nil = 不广告、speak 不可调用，模型面与接线前逐字节一致。 + VoiceSynth VoiceSynth + VoiceStore VoiceStore + // SpeakBudget 可替换 speak 的超时预算公式（文本 rune 数 → 毫秒）。 + // 为 nil 或返回值非正时回落内置公式 speakBudgetMs。它是给测试与 + // 部署调优用的注入点，不是能力开关（开关是上面两个依赖）。 + SpeakBudget func(runes int) int // Log 是基础 logger。为 nil 时替换为标记来源 // "runtime/loop" 的丢弃 logger。 Log *slog.Logger diff --git a/internal/runtime/loop/request.go b/internal/runtime/loop/request.go index f0377d1..0c54db6 100644 --- a/internal/runtime/loop/request.go +++ b/internal/runtime/loop/request.go @@ -87,7 +87,7 @@ func (l *Loop) toolSchemas(ctx context.Context, messages []model.Message) []mode } schemas := d.ToolSchemas() if l.capacity == nil || !l.capacity.Enabled() { - return l.narrowToolSchemas(ctx, l.restrictSchemas(ctx, d, schemas)) + return l.withSpeakSpec(l.narrowToolSchemas(ctx, l.restrictSchemas(ctx, d, schemas))) } names := make([]string, 0, len(schemas)) @@ -124,7 +124,7 @@ func (l *Loop) toolSchemas(ctx context.Context, messages []model.Message) []mode out = append(out, s) } } - return l.narrowToolSchemas(ctx, l.restrictSchemas(ctx, d, out)) + return l.withSpeakSpec(l.narrowToolSchemas(ctx, l.restrictSchemas(ctx, d, out))) } // capacityTexts 为 capacity 判定器提取近期对话文本： diff --git a/internal/runtime/loop/run.go b/internal/runtime/loop/run.go index 9f9a02e..fed6728 100644 --- a/internal/runtime/loop/run.go +++ b/internal/runtime/loop/run.go @@ -187,6 +187,11 @@ func (l *Loop) driveTurn(ctx context.Context, sessionID string, msgs []model.Mes } msgs = batch.messages res.ToolCallsExecuted += batch.executed + // speak 的出站结论逐批回填（M57）：后面批次里没有 speak 时保留 + // 前面那批的结论 —— 这条回复想用语音这件事发生在哪一批都算数。 + if batch.voice != nil { + res.Voice = batch.voice + } l.saveCheckpoint(ctx, sessionID, step, msgs, agent.StopReasonToolCalls) // 关键工具循环安全网（或终止钩子）不等真正的最终答案 diff --git a/internal/runtime/loop/speak_tool.go b/internal/runtime/loop/speak_tool.go new file mode 100644 index 0000000..0f98d1d --- /dev/null +++ b/internal/runtime/loop/speak_tool.go @@ -0,0 +1,336 @@ +package loop + +// 本文件实现模型侧的 speak 工具（M57）。主人 2026-10-27 的口径是「他可以自己 +// 选择某次消息是拿文字发 还是去请求 TTS 接口发音频」——**决策是她的**，因此这里 +// 只提供一个可调用的工具，绝不替她决定什么时候用语音（那会变成幕后置规则）。 +// 调了 speak = 这条想用语音；不调 = 发文本。 +// +// 三条执行口径： +// - 超时预算按**文本 rune 数**推导（speakBudgetMs）：短句拿最低预算，长文拿 +// 上限，中间线性 —— 越长合成越慢，但绝不无限等； +// - 成功 = 音频副本**已落盘**（经 VoiceStore 接缝，接线方走 documents.SaveAudio +// 的内容寻址 0600 口径）+ 结构化结果（name/mime/size/durationMs/budgetMs/costMs）； +// - 超时/失败 = 结构化错误（code + 预算 + 实际耗时），**回合绝不因此失败**， +// 她能在后续输出里看到「这次语音没成」，客户端也能据此特殊显示。 + +import ( + "context" + "errors" + "fmt" + "log/slog" + "strings" + "time" + "unicode/utf8" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + mvoice "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/multimodal/voice" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool" + voicetts "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/voice/tts" + "git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk" +) + +const ( + // speakToolName 是模型侧工具名。它同时是 Result.Voice 回填时的匹配键。 + speakToolName = "speak" + + // speakBudgetBaseMs 是任何一次合成的最低预算：即使一个字的请求也要给 + // 上游留出建连 + 首包的时间，低于它的预算只会把"网络慢"误判成"内容长"。 + speakBudgetBaseMs = 3000 + // speakBudgetPerRuneMs 是每个 rune 追加的预算（毫秒）。25ms/rune 约等于 + // 每秒 40 个字符的合成速度，比实测慢一截，留出余量。 + speakBudgetPerRuneMs = 25 + // speakBudgetMaxMs 是预算上限：再长的文本也等这么久，超了就退回文本， + // 绝不让她的一条消息把整个回合挂住。 + speakBudgetMaxMs = 20000 + // speakMaxRunes 是文本长度硬上限。按上面的公式，约 680 rune 就顶到预算 + // 上限，再长的文本必然超时 —— 早拒绝能省一次上游调用与一次落盘。 + speakMaxRunes = 1200 + + // speakAudioOrigName 是落盘时给 documents.SaveAudio 的原始名，扩展名 + // 由上游实际格式拼接（白名单见 voiceref.MediaExtAllowed）。 + speakAudioOrigName = "speak." + + // 降级/失败原因码（Result.Voice.FallbackReason 与错误正文共用同一套取值， + // 客户端据此决定怎么显示「这条本来想用语音」）。 + speakReasonTimeout = "tts_timeout" + speakReasonFailed = "tts_failed" + speakReasonNoAudio = "tts_no_audio" + speakReasonStoreFailed = "tts_store_failed" + speakReasonTooLong = "text_too_long" + speakReasonInvalidArgs = "invalid_args" +) + +// speakTool 实现 sdk.Tool。依赖全部由循环注入（不自己 new），因此叶子测试 +// 可以塞桩：合成桩决定成功/超时，落盘桩决定存储名，预算桩决定超时长短。 +type speakTool struct { + synth VoiceSynth + store VoiceStore + // budget 把 rune 数映射为预算（毫秒）；nil 用内置公式。 + budget func(runes int) int + // persona 解析本回合的生效人格 id，供 TTS 的发音人解析链使用 + // （发音人绑人格，链本身不动）。 + persona func(context.Context) string + log *slog.Logger +} + +// 编译期断言：speak 就是一个普通的模型侧工具，接线方式与其他工具一致。 +var _ sdk.Tool = speakTool{} + +// Name 返回工具标识。 +func (speakTool) Name() string { return speakToolName } + +// Description 说明**何时**用语音。它是能力描述，不是身份/权限判定 —— +// 谁能用语音仍由渠道层 actor 元数据 + 工具可达性标记决定。 +func (speakTool) Description() string { + return "用语音（合成音频）把这段话念出来，让这一条回复变成声音。" + + "适合用语音：轻声细语、撒娇、唱歌、长一点的心里话或独白、需要语气温度的表达。" + + "适合用文本：代码、列表、链接、报错原文，以及需要被复制或搜索的内容。" + + "语音只改变这一条回复的呈现方式，不改变你的能力或身份。" + + "合成失败或超时会自动退回文本，调用一次就够，不要为同一句话反复重试。" +} + +// Parameters 返回工具输入的 JSON Schema。 +func (speakTool) Parameters() (map[string]any, error) { + return map[string]any{ + "type": "object", + "properties": map[string]any{ + "text": map[string]any{ + "type": "string", + "description": "要念出来的话。写成可直接朗读的口语，不要带 Markdown 标记、代码块或链接。", + }, + "emotion": map[string]any{ + "type": "string", + "description": "可选语气微调（如「温柔」「轻声」「欢快」）。默认语气由你的人格决定，" + + "这个字段只做额外微调，不会换掉发音人。", + }, + }, + "required": []string{"text"}, + }, nil +} + +// speakArgs 是解析后的调用参数。runes 只算一次，预算与日志共用。 +type speakArgs struct { + text string + emotion string + runes int +} + +// Execute 合成一段语音并落盘副本。成功返回结构化结果；失败返回 *speakError， +// 它同时是模型能读懂的错误正文与 Result.Voice 的降级依据。 +func (t speakTool) Execute(ctx context.Context, args any) (any, error) { + in, err := parseSpeakArgs(args) + if err != nil { + return nil, err + } + if t.synth == nil || t.store == nil { + return nil, &speakError{Code: speakReasonFailed, + Err: errors.New("speak: voice synthesis is not configured")} + } + budget := t.budgetMs(in.runes) + synthCtx, cancel := context.WithTimeout(ctx, time.Duration(budget)*time.Millisecond) + defer cancel() + synthCtx = t.stampVoiceContext(synthCtx, ctx, in.emotion) + + started := time.Now() + audio, provider, err := t.synth.TTSWithProvider(synthCtx, in.text) + cost := int(time.Since(started) / time.Millisecond) + if err != nil { + return nil, t.synthesisFailed(synthCtx, err, budget, cost) + } + ext, extOK := speakExt(audio.Format) + if len(audio.Data) == 0 || !extOK { + return nil, &speakError{Code: speakReasonNoAudio, BudgetMs: budget, CostMs: cost, + Err: fmt.Errorf("speak: unusable audio (format %q, %d bytes)", audio.Format, len(audio.Data))} + } + // 落盘用**父 ctx**：预算只约束上游合成，不该让已经合成好的音频因为预算 + // 到点而丢掉（写盘是本地动作，毫秒级，且副本本身就是交付物）。 + ref, err := t.store.SaveAudio(ctx, speakAudioOrigName+ext, audio.Data) + if err != nil { + return nil, &speakError{Code: speakReasonStoreFailed, BudgetMs: budget, CostMs: cost, Err: err} + } + t.logSynthesized(provider, in.runes, budget, cost, len(audio.Data)) + // durationMs 恒为 0：本项目**绝不估算时长**（与 voice.Copy.DurationMs 同一 + // 口径），而解析 WAV 头不在本切片范围内。0 表示「未解析」，不是零长度。 + return map[string]any{ + "name": ref.Name, + "mime": ref.MIME, + "size": ref.Size, + "durationMs": int64(0), + "budgetMs": budget, + "costMs": cost, + }, nil +} + +// stampVoiceContext 把本回合的生效人格与可选语气挂到 ctx 上：TTS 侧按人格 +// 解析发音人（链不动），emotion 只作为逐请求的语气指令透传到底层链。 +func (t speakTool) stampVoiceContext(synthCtx, turnCtx context.Context, emotion string) context.Context { + if t.persona != nil { + if pid := t.persona(turnCtx); pid != "" { + synthCtx = voicetts.WithPersonaID(synthCtx, pid) + } + } + if emotion != "" { + synthCtx = mvoice.WithTTSInstruction(synthCtx, emotion) + } + return synthCtx +} + +// budgetMs 返回本次合成的超时预算；注入值非正时回落内置公式。 +func (t speakTool) budgetMs(runes int) int { + if t.budget != nil { + if ms := t.budget(runes); ms > 0 { + return ms + } + } + return speakBudgetMs(runes) +} + +// synthesisFailed 把上游错误归一化成带原因码的结构化失败。超时判定看三处： +// 上游自己报的 deadline、我们给的预算 ctx 已过期、或实际耗时已经吃满预算 +// （上游把超时包成自己的错误类型时只有最后一处能认出来）。 +func (t speakTool) synthesisFailed(synthCtx context.Context, err error, budget, cost int) error { + code := speakReasonFailed + if errors.Is(err, context.DeadlineExceeded) || + errors.Is(synthCtx.Err(), context.DeadlineExceeded) || cost >= budget { + code = speakReasonTimeout + } + if t.log != nil { + // 只记长度与耗时，**绝不记正文**（与网关出站 TTS 同一口径）。 + t.log.Warn("speak degraded to text", + "reason", code, "budget_ms", budget, "cost_ms", cost, "err", err) + } + return &speakError{Code: code, BudgetMs: budget, CostMs: cost, Err: err} +} + +// logSynthesized 记一条成功合成。只带来源、级别名、长度与耗时，绝不带正文。 +func (t speakTool) logSynthesized(provider string, runes, budget, cost, size int) { + if t.log == nil { + return + } + t.log.Info("speak synthesized", + "provider", provider, "runes", runes, "budget_ms", budget, "cost_ms", cost, "bytes", size) +} + +// speakError 是 speak 的结构化失败。Code 进 Result.Voice.FallbackReason，让 +// 「想用语音但没成」与「本来就用文本」在调用方与客户端都能区分开；预算与实际 +// 耗时让模型下一轮能判断是"该说短点"还是"上游坏了"。 +type speakError struct { + Code string + BudgetMs int + CostMs int + Err error +} + +// Error 渲染成一句可读正文，随 RoleTool 消息回灌给模型。 +func (e *speakError) Error() string { + if e.Err == nil { + return fmt.Sprintf("speak failed: %s (budget_ms=%d cost_ms=%d)", e.Code, e.BudgetMs, e.CostMs) + } + return fmt.Sprintf("speak failed: %s (budget_ms=%d cost_ms=%d): %v", e.Code, e.BudgetMs, e.CostMs, e.Err) +} + +// Unwrap 让 errors.Is 能穿透到原因（含 tool.ErrInvalidArgs 的参数校验分类）。 +func (e *speakError) Unwrap() error { return e.Err } + +// parseSpeakArgs 解析并校验参数。参数错误包 tool.ErrInvalidArgs，使循环把它 +// 归到 argument-validation 而不是执行失败。 +func parseSpeakArgs(args any) (speakArgs, error) { + m, err := speakArgMap(args) + if err != nil { + return speakArgs{}, err + } + text, _ := m["text"].(string) + text = strings.TrimSpace(text) + if text == "" { + return speakArgs{}, &speakError{Code: speakReasonInvalidArgs, + Err: fmt.Errorf("speak: text is required: %w", tool.ErrInvalidArgs)} + } + runes := utf8.RuneCountInString(text) + if runes > speakMaxRunes { + return speakArgs{}, &speakError{Code: speakReasonTooLong, + Err: fmt.Errorf("speak: text has %d runes, limit is %d: %w", runes, speakMaxRunes, tool.ErrInvalidArgs)} + } + emotion, _ := m["emotion"].(string) + return speakArgs{text: text, emotion: strings.TrimSpace(emotion), runes: runes}, nil +} + +// speakArgMap 把调用参数归一化成 map；nil 参数视为空对象（缺 text 由调用方报）。 +func speakArgMap(args any) (map[string]any, error) { + if args == nil { + return map[string]any{}, nil + } + m, ok := args.(map[string]any) + if !ok { + return nil, &speakError{Code: speakReasonInvalidArgs, + Err: fmt.Errorf("speak: arguments must be an object: %w", tool.ErrInvalidArgs)} + } + return m, nil +} + +// speakBudgetMs 把文本长度（rune 数）映射为合成超时预算： +// clamp(speakBudgetBaseMs + speakBudgetPerRuneMs×runes, base, max)。 +func speakBudgetMs(runes int) int { + if runes < 0 { + runes = 0 + } + ms := speakBudgetBaseMs + speakBudgetPerRuneMs*runes + if ms > speakBudgetMaxMs { + return speakBudgetMaxMs + } + if ms < speakBudgetBaseMs { + return speakBudgetBaseMs + } + return ms +} + +// speakExt 把上游实际格式映射成落盘扩展名。空格式按 wav 命名 —— TTS 上游 +// 产物恒为 WAV（见 voice/tts/audio.go 的口径）；白名单外的格式一律判不可用， +// 绝不拿一个读不出 Content-Type 的名字落盘。 +func speakExt(format string) (string, bool) { + switch strings.ToLower(strings.TrimSpace(format)) { + case "": + return "wav", true + case "wav", "mp3", "ogg", "flac", "m4a", "webm", "amr": + return strings.ToLower(strings.TrimSpace(format)), true + default: + return "", false + } +} + +// speakSpec 返回 speak 的模型面声明（工具未接线时为 nil）。 +func (l *Loop) speakSpec() *model.ToolSpec { + if !l.voiceEnabled() { + return nil + } + t := l.speakTool() + params, err := t.Parameters() + if err != nil { + return nil + } + return &model.ToolSpec{Name: t.Name(), Description: t.Description(), Parameters: params} +} + +// speakTool 用循环注入的依赖装配一个 speak 工具值（每次装配都是廉价的）。 +func (l *Loop) speakTool() speakTool { + return speakTool{ + synth: l.voiceSynth, + store: l.voiceStore, + budget: l.speakBudget, + persona: l.personaFor, + log: l.log, + } +} + +// invokeTool 执行单个工具调用。speak 由循环自己承载（它需要本回合的人格与 +// 预算口径，且未注册进能力登记表），其余一律交给 dispatcher —— 它的超时、 +// panic 兜底与允许集软判对普通工具仍然生效。 +func (l *Loop) invokeTool(ctx context.Context, d *tool.Dispatcher, tc agent.ToolCall) (any, error) { + if tc.Name == speakToolName && l.voiceEnabled() { + return l.speakTool().Execute(ctx, tc.Args) + } + if d == nil { + return nil, fmt.Errorf("no tool dispatcher configured for %q", tc.Name) + } + return d.Invoke(tool.WithLegID(ctx, ToolLegIDFrom(ctx)), tc.Name, tc.Args) +} diff --git a/internal/runtime/loop/tools.go b/internal/runtime/loop/tools.go index 902f328..f966d98 100644 --- a/internal/runtime/loop/tools.go +++ b/internal/runtime/loop/tools.go @@ -5,12 +5,10 @@ import ( "encoding/json" "errors" "fmt" - "sync" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool" - "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" ) // Options.ToolExecution 接受的 ToolExecutionMode 取值。为空时选用 @@ -77,6 +75,9 @@ type toolRunResult struct { messages []model.Message executed int terminate bool + // voice 是本批 speak 调用的结论（M57）：nil = 这一批没有要求语音。 + // 由 runToolCalls 在结局钩子之后提取，供 run.go 回填 Result.Voice。 + voice *OutboundVoice } // toolLoopState 携带关键工具循环安全网的每次 run 状态。 @@ -156,6 +157,10 @@ func (l *Loop) runToolCalls(ctx context.Context, messages []model.Message, calls // 结果落定后的两类观察/改写钩子见 applyToolOutcomeHooks。 l.applyToolOutcomeHooks(ctx, calls, outcomes) + // 语音结论必须在钩子之后提取：AfterOutcome 可以改写 speak 的结果， + // 出站结论要看**模型最终看到的那份**。 + res.voice = l.captureVoice(calls, outcomes) + // 钩子可以把每个调用都标记为 terminate 来要求终止本轮 // （OpenClaw 的 shouldTerminateToolBatch）。 if !res.terminate && shouldTerminateBatch(outcomes) { @@ -287,95 +292,6 @@ func (l *Loop) applyToolOutcomeHooks(ctx context.Context, calls []agent.ToolCall } } -// invokeOne 在其每工具截止时间下执行单个工具调用，发出一个 -// tool span、以 tool_execution.start/end 事件流式输出执行进度 -// 并归一化结果。执行全程由单一 phase 决策驱动：start 帧带 -// running，Invoke 返回后按成败取 done/error，同一取值既决定 -// end 帧的 phase，也决定 span 终态（design/19 的 tool 节点 -// running/done/error 状态即由此驱动）。 -func (l *Loop) invokeOne(ctx context.Context, d *tool.Dispatcher, tc agent.ToolCall) ToolOutcome { - if d == nil { - return ToolOutcome{Err: fmt.Errorf("no tool dispatcher configured for %q", tc.Name), ErrKind: tool.KindExecution} - } - _, toolSpan := l.span(ctx, "tool."+tc.Name, trace.SpanKindTool) - l.emitToolExecution(ctx, tc, model.StreamKindToolExecutionStart, model.ToolPhaseRunning, "") - // 路标识桥（design/15 §2.5.7、主人裁决 3）：网关在**每一条在途路**上把互斥 - // 身份经 loop.WithToolLegID 挂在 ctx 上（普通回合 / 并行路 / 队首续跑路都挂， - // 取值 = 该路的 turnID），这里把它桥接成工具层的 tool.WithLegID —— 工具侧 - // 凭 tool.LegIDFrom 读到「我是哪一路」做文件级互斥；未挂载 ⇒ 空串 ⇒ 完全 - // 不做互斥，行为逐字不变（兼容硬要求）。 - // - // 刻意**不**读 ParallelLegFrom：那条通道只有真并行路才有、且只决定 checkpoint - // 归属（checkpoint.go 的 checkpointAgentID），拿它当互斥身份会让普通回合在 - // 「1 普通回合 + 1 并行路」时完全不受保护。 - result, err := d.Invoke(tool.WithLegID(ctx, ToolLegIDFrom(ctx)), tc.Name, tc.Args) - phase := model.ToolPhaseDone - if err != nil { - phase = model.ToolPhaseError - } - l.emitToolExecution(ctx, tc, model.StreamKindToolExecutionEnd, phase, toolEndFrameSummary(result, err)) - out := ToolOutcome{Result: result, Err: err, ErrKind: errorKind(err)} - l.spanProp(toolSpan, "phase", phase) - l.recordToolDone(toolSpan, tc, out) - if err != nil { - l.spanProp(toolSpan, "error", err.Error()) - l.spanProp(toolSpan, "error_kind", out.ErrKind) - } else { - recordToolResult(toolSpan, result) - } - l.finishSpan(toolSpan, spanStatusForToolPhase(phase)) - return out -} - -// emitToolExecution 在循环处于流式路径（配置了 OnStream）时把 -// tool_execution 三事件（start / update / end）转发给前端；否则为 -// 空操作（被静音的内部流程也会跳过它）。phase 是结构化字段 -// （running/done/error），绝不塞进 Delta；工具名走 Tool 字段， -// ID 复用 provider 的 tool call id，使执行帧能与其 tool_call.* 帧关联； -// resultSummary 只在 end 帧、且只在结果真正拿到时非空（口径见 tool_frame_summary.go）。 -func (l *Loop) emitToolExecution(ctx context.Context, tc agent.ToolCall, kind model.StreamKind, phase, resultSummary string) { - if l.onStream == nil || isMutedBroadcast(ctx) { - return - } - ev := model.StreamEvent{Kind: kind, Phase: phase, Tool: tc.Name, ID: tc.ID, ResultSummary: resultSummary} - ev.SessionID = loopSession(ctx) - ev.TurnID = loopTurn(ctx) - l.onStream(ev) -} - -// spanStatusForToolPhase 把工具执行的终态 phase 映射为 span 终态， -// 使 trace/流程图与 WS 帧从同一个 phase 取值得出一致结论。 -// 非终态 phase（running）不会走到这里 —— span 只在执行结束时收尾。 -func spanStatusForToolPhase(phase string) trace.SpanStatus { - if phase == model.ToolPhaseError { - return trace.SpanStatusError - } - return trace.SpanStatusOK -} - -// invokeParallel 并发运行 idxs 处的调用，上限为循环的 -// 并行度，并把每个结果写回 out 的对应调用索引处，使调用方 -// 能按原始顺序渲染结果。信号量限制并发，在途 -// goroutine 完成时释放其槽位，因此被取消的 context -// （它使每个 Invoke 迅速返回）不会泄露 goroutine。 -func (l *Loop) invokeParallel(ctx context.Context, d *tool.Dispatcher, calls []agent.ToolCall, idxs []int, out []ToolOutcome) { - if len(idxs) == 0 { - return - } - sem := make(chan struct{}, l.toolParallelism) - var wg sync.WaitGroup - for _, i := range idxs { - wg.Add(1) - sem <- struct{}{} - go func(i int) { - defer wg.Done() - defer func() { <-sem }() - out[i] = l.invokeOne(ctx, d, calls[i]) - }(i) - } - wg.Wait() -} - // 由 classifyToolBatch 为两种跳过原因返回。 func duplicateCallReason() string { return "duplicate call with identical arguments already issued in this batch" diff --git a/internal/runtime/loop/tools_invoke.go b/internal/runtime/loop/tools_invoke.go new file mode 100644 index 0000000..e4993a7 --- /dev/null +++ b/internal/runtime/loop/tools_invoke.go @@ -0,0 +1,107 @@ +package loop + +// 本文件承载**单个/一批工具调用的执行**（span、流式帧、结果归一化、 +// 有界并行），与 tools.go 的分类、准入与回渲职责分开 —— 从 tools.go +// 抽出，使两边都留在 500 行以内（AGENTS.md §10）。 + +import ( + "context" + "fmt" + "sync" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// invokeOne 在其每工具截止时间下执行单个工具调用，发出一个 +// tool span、以 tool_execution.start/end 事件流式输出执行进度 +// 并归一化结果。执行全程由单一 phase 决策驱动：start 帧带 +// running，Invoke 返回后按成败取 done/error，同一取值既决定 +// end 帧的 phase，也决定 span 终态（design/19 的 tool 节点 +// running/done/error 状态即由此驱动）。 +func (l *Loop) invokeOne(ctx context.Context, d *tool.Dispatcher, tc agent.ToolCall) ToolOutcome { + // speak 由循环自己承载（见 invokeTool），没有 dispatcher 也能执行；其余 + // 工具缺 dispatcher 时仍按原样短路，不为此多开 span 或流式帧。 + if d == nil && !(l.voiceEnabled() && tc.Name == speakToolName) { + return ToolOutcome{Err: fmt.Errorf("no tool dispatcher configured for %q", tc.Name), ErrKind: tool.KindExecution} + } + _, toolSpan := l.span(ctx, "tool."+tc.Name, trace.SpanKindTool) + l.emitToolExecution(ctx, tc, model.StreamKindToolExecutionStart, model.ToolPhaseRunning, "") + // 分派见 invokeTool：路标识桥（tool.WithLegID）与 speak 的循环内承载都在 + // 那里。路标识（design/15 §2.5.7、主人裁决 3）：网关在**每一条在途路**上把 + // 互斥身份经 loop.WithToolLegID 挂在 ctx 上（普通回合 / 并行路 / 队首续跑路 + // 都挂，取值 = 该路的 turnID），工具层凭 tool.LegIDFrom 读到「我是哪一路」 + // 做文件级互斥；未挂载 ⇒ 空串 ⇒ 完全不做互斥，行为逐字不变（兼容硬要求）。 + // + // 刻意**不**读 ParallelLegFrom：那条通道只有真并行路才有、且只决定 checkpoint + // 归属（checkpoint.go 的 checkpointAgentID），拿它当互斥身份会让普通回合在 + // 「1 普通回合 + 1 并行路」时完全不受保护。 + result, err := l.invokeTool(ctx, d, tc) + phase := model.ToolPhaseDone + if err != nil { + phase = model.ToolPhaseError + } + l.emitToolExecution(ctx, tc, model.StreamKindToolExecutionEnd, phase, toolEndFrameSummary(result, err)) + out := ToolOutcome{Result: result, Err: err, ErrKind: errorKind(err)} + l.spanProp(toolSpan, "phase", phase) + l.recordToolDone(toolSpan, tc, out) + if err != nil { + l.spanProp(toolSpan, "error", err.Error()) + l.spanProp(toolSpan, "error_kind", out.ErrKind) + } else { + recordToolResult(toolSpan, result) + } + l.finishSpan(toolSpan, spanStatusForToolPhase(phase)) + return out +} + +// emitToolExecution 在循环处于流式路径（配置了 OnStream）时把 +// tool_execution 三事件（start / update / end）转发给前端；否则为 +// 空操作（被静音的内部流程也会跳过它）。phase 是结构化字段 +// （running/done/error），绝不塞进 Delta；工具名走 Tool 字段， +// ID 复用 provider 的 tool call id，使执行帧能与其 tool_call.* 帧关联； +// resultSummary 只在 end 帧、且只在结果真正拿到时非空（口径见 tool_frame_summary.go）。 +func (l *Loop) emitToolExecution(ctx context.Context, tc agent.ToolCall, kind model.StreamKind, phase, resultSummary string) { + if l.onStream == nil || isMutedBroadcast(ctx) { + return + } + ev := model.StreamEvent{Kind: kind, Phase: phase, Tool: tc.Name, ID: tc.ID, ResultSummary: resultSummary} + ev.SessionID = loopSession(ctx) + ev.TurnID = loopTurn(ctx) + l.onStream(ev) +} + +// spanStatusForToolPhase 把工具执行的终态 phase 映射为 span 终态， +// 使 trace/流程图与 WS 帧从同一个 phase 取值得出一致结论。 +// 非终态 phase（running）不会走到这里 —— span 只在执行结束时收尾。 +func spanStatusForToolPhase(phase string) trace.SpanStatus { + if phase == model.ToolPhaseError { + return trace.SpanStatusError + } + return trace.SpanStatusOK +} + +// invokeParallel 并发运行 idxs 处的调用，上限为循环的 +// 并行度，并把每个结果写回 out 的对应调用索引处，使调用方 +// 能按原始顺序渲染结果。信号量限制并发，在途 +// goroutine 完成时释放其槽位，因此被取消的 context +// （它使每个 Invoke 迅速返回）不会泄露 goroutine。 +func (l *Loop) invokeParallel(ctx context.Context, d *tool.Dispatcher, calls []agent.ToolCall, idxs []int, out []ToolOutcome) { + if len(idxs) == 0 { + return + } + sem := make(chan struct{}, l.toolParallelism) + var wg sync.WaitGroup + for _, i := range idxs { + wg.Add(1) + sem <- struct{}{} + go func(i int) { + defer wg.Done() + defer func() { <-sem }() + out[i] = l.invokeOne(ctx, d, calls[i]) + }(i) + } + wg.Wait() +} diff --git a/internal/runtime/loop/voice_deps.go b/internal/runtime/loop/voice_deps.go new file mode 100644 index 0000000..8e57457 --- /dev/null +++ b/internal/runtime/loop/voice_deps.go @@ -0,0 +1,152 @@ +package loop + +// 本文件是 speak 的接线面（M57）：两个注入依赖的接口、接线判定、 +// 以及「本回合的出站语音结论」从工具结局到 Result.Voice 的提取。 +// 它们刻意留在循环里而不是工具包里 —— speak 需要本回合的人格与 +// 预算口径，而这两样只有循环持有。 + +import ( + "context" + "errors" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + mvoice "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/multimodal/voice" +) + +// VoiceSynth 是 speak 需要的合成能力（*tts.Service 天然满足）。 +// 接口收窄到**一个方法**：循环只需要「给定文本拿回音频与提供者」， +// 发音人解析、降级链与配额全在实现内部，循环不重复实现它们。 +type VoiceSynth interface { + // TTSWithProvider 合成文本并返回音频与**实际**提供者标识 + // （降级后是备用的那个），供日志与可观测性使用。 + TTSWithProvider(ctx context.Context, text string) (mvoice.Audio, string, error) +} + +// VoiceRef 是音频副本落盘后的交付描述：循环只需要「叫什么、什么类型、 +// 多大」，与存储层的哈希/去重标记/磁盘路径无关。单独定义而不复用 +// documents.Ref 是为了让循环不 import documents 包 —— 该包的测试会 +// import 循环，反向依赖构成测试期的 import cycle。 +type VoiceRef struct { + // Name 是服务端存储名（客户端凭它取回音频）。 + Name string + // MIME 是存储层字节嗅探所得，不是合成侧的声明。 + MIME string + // Size 是该副本的字节数。 + Size int64 +} + +// VoiceStore 是 speak 需要的落盘能力。只要一个方法：音频副本必须走文档 +// 存储的内容寻址 + 权限口径，而不是循环自己写文件。*documents.Store 经 +// VoiceStoreFunc 适配即可接上（见该类型的用法说明）。 +type VoiceStore interface { + // SaveAudio 落盘一份音频副本并返回其交付描述。 + SaveAudio(ctx context.Context, origName string, data []byte) (VoiceRef, error) +} + +// VoiceStoreFunc 把函数适配为 VoiceStore（与 ImageResolveFunc 同一惯例）。 +// 接线方用它把存储层接到本包的接缝上，避免循环反向依赖 documents： +// +// loop.VoiceStoreFunc(func(ctx context.Context, name string, data []byte) (loop.VoiceRef, error) { +// ref, err := docs.SaveAudio(ctx, name, data) +// return loop.VoiceRef{Name: ref.Name, MIME: ref.MIME, Size: ref.Size}, err +// }) +type VoiceStoreFunc func(ctx context.Context, origName string, data []byte) (VoiceRef, error) + +// SaveAudio 实现 VoiceStore。 +func (f VoiceStoreFunc) SaveAudio(ctx context.Context, origName string, data []byte) (VoiceRef, error) { + return f(ctx, origName, data) +} + +// voiceEnabled 报告 speak 是否已接线。两个依赖**都**非 nil 才算 —— +// 只接了合成却没有存储时，调用必然在落盘一步失败，不如干脆不广告， +// 让模型面与接线前逐字节一致。 +func (l *Loop) voiceEnabled() bool { + return l != nil && l.voiceSynth != nil && l.voiceStore != nil +} + +// withSpeakSpec 把 speak 追加到模型面的工具声明末尾。它由 toolSchemas +// 在**所有收缩之后**调用，因此 speak 不受能力允许集（quiet/awake）与 +// 技能轮次的 allowed-tools 裁剪 —— speak 不在任何内置 toolset 里， +// 走那些过滤就等于永远不广告。代价（群聊/陌生人也能看到 speak）由 +// 执行侧的工具安全门兜住：speak 未注册进能力登记表，非主人环境下 +// Policy.Allow 以「未知工具」拒绝，fail-closed。 +func (l *Loop) withSpeakSpec(schemas []model.ToolSpec) []model.ToolSpec { + spec := l.speakSpec() + if spec == nil { + return schemas + } + for i := range schemas { + if schemas[i].Name == spec.Name { + return schemas + } + } + out := make([]model.ToolSpec, 0, len(schemas)+1) + out = append(out, schemas...) + return append(out, *spec) +} + +// captureVoice 从一批调用的结局里提取本回合的出站语音结论。返回 nil +// 表示「这一批没有要求语音」，Result.Voice 因此保持 nil，客户端按纯 +// 文本呈现 —— 与没有 speak 之前逐字节一致。 +// +// 多个 speak 调用时取**最后一个成功的**：成功意味着音频真的落了盘， +// 后续的重试即使失败也不该让已经交付的音频消失。全都失败时取最后 +// 一个失败的原因，使客户端仍能区分「想用语音但没成」与「本来就用 +// 文本」。 +func (l *Loop) captureVoice(calls []agent.ToolCall, outcomes []ToolOutcome) *OutboundVoice { + var lastOK, lastFail *OutboundVoice + n := len(calls) + if len(outcomes) < n { + n = len(outcomes) + } + for i := 0; i < n; i++ { + if calls[i].Name != speakToolName || outcomes[i].Skipped { + continue + } + if outcomes[i].Err == nil { + lastOK = synthesizedVoice(outcomes[i].Result) + continue + } + lastFail = degradedVoice(outcomes[i].Err) + } + if lastOK != nil { + return lastOK + } + return lastFail +} + +// synthesizedVoice 把 speak 成功时的结构化结果翻成 OutboundVoice。 +// 结果被钩子改写成非对象时返回 nil：拿不到存储名就播放不了，此时 +// 按文本呈现是唯一诚实的选择（DurationMs 保持 0 = 未解析）。 +func synthesizedVoice(result any) *OutboundVoice { + m, ok := result.(map[string]any) + if !ok { + return nil + } + v := &OutboundVoice{Requested: true} + v.Name, _ = m["name"].(string) + v.Mime, _ = m["mime"].(string) + v.Size, _ = m["size"].(int64) + v.DurationMs, _ = m["durationMs"].(int64) + v.CostMs, _ = m["costMs"].(int) + v.BudgetMs, _ = m["budgetMs"].(int) + return v +} + +// degradedVoice 把 speak 的失败翻成「想用语音但退回了文本」。原因码 +// 优先取结构化错误的 Code（超时/无音频/落盘失败/文本过长），非结构化 +// 失败一律归 tts_failed，绝不把空原因码交给客户端。 +func degradedVoice(err error) *OutboundVoice { + v := &OutboundVoice{Requested: true, FallbackReason: speakReasonFailed} + var se *speakError + if !errors.As(err, &se) { + return v + } + if se.Code != "" { + v.FallbackReason = se.Code + } + v.BudgetMs = se.BudgetMs + v.CostMs = se.CostMs + return v +}