feat: per-call LLM timing, speaker ID, cancel captures on leave
- llm.py: log each Ollama call's caller + total/load/prompt/gen durations so a slow voice turn is attributable to a specific internal call (router/enrichment/digest/main); a RELOAD marker flags cold reloads. - voice.ts: track in-flight Opus captures and abort them on session destroy(); drop any utterance that finishes after the user left, so no trailing post-leave VAD turns are reported. - userbot.ts: show the speaker's Discord user ID on each transcript line (answered and dropped) so it's clear whose audio produced the turn. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -53,6 +53,14 @@ test("formatTurnMessage reports a dropped turn with only the listening stage", (
|
|||||||
expect(msg).not.toContain("🔊 TTS");
|
expect(msg).not.toContain("🔊 TTS");
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("formatTurnMessage shows the speaker user ID when provided", () => {
|
||||||
|
const answered = formatTurnMessage({ user: "12345", transcript: "안녕", reply: "하이" });
|
||||||
|
expect(answered).toContain("👤 12345");
|
||||||
|
const dropped = formatTurnMessage({ user: "67890", transcript: "", reply: "", note: "음성 아님(VAD 차단)" });
|
||||||
|
expect(dropped).toContain("👤 67890");
|
||||||
|
expect(dropped).toContain("❌ 음성 아님(VAD 차단)");
|
||||||
|
});
|
||||||
|
|
||||||
test("formatTurnMessage falls back to the plain line when no timing is present", () => {
|
test("formatTurnMessage falls back to the plain line when no timing is present", () => {
|
||||||
const msg = formatTurnMessage({ transcript: "안녕", reply: "하이" });
|
const msg = formatTurnMessage({ transcript: "안녕", reply: "하이" });
|
||||||
expect(msg).toBe('🎤 들음 → 🗣️ "안녕"\n🤖 답변: 하이');
|
expect(msg).toBe('🎤 들음 → 🗣️ "안녕"\n🤖 답변: 하이');
|
||||||
|
|||||||
@@ -35,6 +35,9 @@ async function loadSelfbot(): Promise<any> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
export interface TurnInfo {
|
export interface TurnInfo {
|
||||||
|
/** Discord user ID of the speaker, so the transcript shows whose audio
|
||||||
|
* produced each turn (and which user a dropped/VAD turn belongs to). */
|
||||||
|
user?: string;
|
||||||
transcript: string;
|
transcript: string;
|
||||||
reply: string;
|
reply: string;
|
||||||
note?: string;
|
note?: string;
|
||||||
@@ -69,9 +72,10 @@ function durSec(a?: number, b?: number): string | null {
|
|||||||
* timing breakdown (listening / LLM / TTS) with start→end wall-clock times and
|
* timing breakdown (listening / LLM / TTS) with start→end wall-clock times and
|
||||||
* durations, so it's obvious what took long. Pure + exported for testing. */
|
* durations, so it's obvious what took long. Pure + exported for testing. */
|
||||||
export function formatTurnMessage(info: TurnInfo): string {
|
export function formatTurnMessage(info: TurnInfo): string {
|
||||||
|
const who = info.user ? `👤 ${info.user} ` : "";
|
||||||
const head = info.transcript
|
const head = info.transcript
|
||||||
? `🎤 들음 → 🗣️ "${info.transcript}"\n🤖 답변: ${(info.reply || "").trim() || "(무응답)"}`
|
? `${who}🎤 들음 → 🗣️ "${info.transcript}"\n🤖 답변: ${(info.reply || "").trim() || "(무응답)"}`
|
||||||
: `🎤 들음 → ❌ ${info.note || "무시됨"}`;
|
: `${who}🎤 들음 → ❌ ${info.note || "무시됨"}`;
|
||||||
|
|
||||||
const lines: string[] = [];
|
const lines: string[] = [];
|
||||||
const listen = durSec(info.listenStartMs, info.listenEndMs);
|
const listen = durSec(info.listenStartMs, info.listenEndMs);
|
||||||
@@ -120,7 +124,7 @@ async function joinAndListen(client: AnyClient, channelId: string): Promise<void
|
|||||||
// joinVoiceChannel (it exposes id, guild.id and guild.voiceAdapterCreator).
|
// joinVoiceChannel (it exposes id, guild.id and guild.voiceAdapterCreator).
|
||||||
const session = await joinChannel(channel as unknown as VoiceBasedChannel);
|
const session = await joinChannel(channel as unknown as VoiceBasedChannel);
|
||||||
session.onTurn = (info) => {
|
session.onTurn = (info) => {
|
||||||
console.log(`🗣️ ${info.transcript || "(" + (info.note || "empty") + ")"}\n🤖 ${info.reply}`);
|
console.log(`👤 ${info.user || "?"} 🗣️ ${info.transcript || "(" + (info.note || "empty") + ")"}\n🤖 ${info.reply}`);
|
||||||
// Mirror every heard utterance (and the reply / drop reason) to a text
|
// Mirror every heard utterance (and the reply / drop reason) to a text
|
||||||
// channel so you can see what the bot understood even when it doesn't answer.
|
// channel so you can see what the bot understood even when it doesn't answer.
|
||||||
void postTranscript(client, info);
|
void postTranscript(client, info);
|
||||||
|
|||||||
@@ -66,6 +66,13 @@ export class VoiceSession {
|
|||||||
private connection: VoiceConnection;
|
private connection: VoiceConnection;
|
||||||
private player: AudioPlayer;
|
private player: AudioPlayer;
|
||||||
private listening = new Set<string>();
|
private listening = new Set<string>();
|
||||||
|
/** Set once the session is torn down (user left / leave command). In-flight
|
||||||
|
* captures check this so we don't run STT/reply or post a transcript for
|
||||||
|
* audio that arrived after the user already left the channel. */
|
||||||
|
private destroyed = false;
|
||||||
|
/** Opus subscriptions currently capturing, so leave() can end them
|
||||||
|
* immediately instead of waiting out the silence timeout. */
|
||||||
|
private activeStreams = new Set<{ destroy: () => void }>();
|
||||||
/** Pending reply clips. Played one at a time so concurrent speakers don't
|
/** Pending reply clips. Played one at a time so concurrent speakers don't
|
||||||
* cut each other's replies off. */
|
* cut each other's replies off. */
|
||||||
private playQueue: Buffer[] = [];
|
private playQueue: Buffer[] = [];
|
||||||
@@ -158,11 +165,14 @@ export class VoiceSession {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private async captureUtterance(userId: string): Promise<void> {
|
private async captureUtterance(userId: string): Promise<void> {
|
||||||
|
// Don't start a new capture once we're tearing down (user left).
|
||||||
|
if (this.destroyed) return;
|
||||||
// "듣기 시작": the moment we begin capturing this speaker's utterance.
|
// "듣기 시작": the moment we begin capturing this speaker's utterance.
|
||||||
const listenStartMs = Date.now();
|
const listenStartMs = Date.now();
|
||||||
const opusStream = this.connection.receiver.subscribe(userId, {
|
const opusStream = this.connection.receiver.subscribe(userId, {
|
||||||
end: { behavior: EndBehaviorType.AfterSilence, duration: config.silenceMs },
|
end: { behavior: EndBehaviorType.AfterSilence, duration: config.silenceMs },
|
||||||
});
|
});
|
||||||
|
this.activeStreams.add(opusStream);
|
||||||
const decoder = new prism.opus.Decoder({
|
const decoder = new prism.opus.Decoder({
|
||||||
frameSize: 960,
|
frameSize: 960,
|
||||||
channels: DISCORD_CHANNELS,
|
channels: DISCORD_CHANNELS,
|
||||||
@@ -173,6 +183,11 @@ export class VoiceSession {
|
|||||||
pcmStream.on("data", (c: Buffer) => chunks.push(c));
|
pcmStream.on("data", (c: Buffer) => chunks.push(c));
|
||||||
|
|
||||||
await new Promise<void>((resolve) => pcmStream.once("end", () => resolve()));
|
await new Promise<void>((resolve) => pcmStream.once("end", () => resolve()));
|
||||||
|
this.activeStreams.delete(opusStream);
|
||||||
|
// If the user left while we were capturing, drop this utterance entirely —
|
||||||
|
// don't run STT/reply or report a (usually empty/VAD-blocked) turn for
|
||||||
|
// audio that trailed in after they left.
|
||||||
|
if (this.destroyed) return;
|
||||||
// "듣기 종료": end of speech (silence detected). Anything after this is
|
// "듣기 종료": end of speech (silence detected). Anything after this is
|
||||||
// STT + reply + TTS on the brain side.
|
// STT + reply + TTS on the brain side.
|
||||||
const listenEndMs = Date.now();
|
const listenEndMs = Date.now();
|
||||||
@@ -264,6 +279,18 @@ export class VoiceSession {
|
|||||||
}
|
}
|
||||||
|
|
||||||
destroy() {
|
destroy() {
|
||||||
|
this.destroyed = true;
|
||||||
|
// End any in-flight captures now so their pcmStream resolves immediately
|
||||||
|
// and the post-capture `destroyed` check drops them (no trailing
|
||||||
|
// post-leave VAD turns).
|
||||||
|
for (const s of this.activeStreams) {
|
||||||
|
try {
|
||||||
|
s.destroy();
|
||||||
|
} catch {
|
||||||
|
/* already ended */
|
||||||
|
}
|
||||||
|
}
|
||||||
|
this.activeStreams.clear();
|
||||||
try {
|
try {
|
||||||
this.connection.destroy();
|
this.connection.destroy();
|
||||||
} catch {
|
} catch {
|
||||||
|
|||||||
@@ -3,9 +3,19 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
from typing import Optional, Any, Dict, List, Generator, Callable
|
from typing import Optional, Any, Dict, List, Generator, Callable
|
||||||
import os
|
import os
|
||||||
|
import sys
|
||||||
import requests
|
import requests
|
||||||
import json
|
import json
|
||||||
|
|
||||||
|
|
||||||
|
def _caller_name() -> str:
|
||||||
|
"""Best-effort name of the function that invoked the LLM wrapper, used to
|
||||||
|
label per-call timing (router / enrichment / digest / main)."""
|
||||||
|
try:
|
||||||
|
return sys._getframe(2).f_code.co_name
|
||||||
|
except Exception:
|
||||||
|
return "?"
|
||||||
|
|
||||||
from .debug import debug_log
|
from .debug import debug_log
|
||||||
|
|
||||||
|
|
||||||
@@ -25,6 +35,33 @@ class ToolsNotSupportedError(Exception):
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def _log_ollama_timing(data: Dict[str, Any], num_ctx: int, caller: str) -> None:
|
||||||
|
"""Emit a one-line per-call latency breakdown so a slow voice turn can be
|
||||||
|
attributed to a specific internal LLM call (router / enrichment / digest /
|
||||||
|
main) instead of just a total. ``load_duration`` > 0 means the model was
|
||||||
|
cold-reloaded for this call — the single most expensive thing to avoid.
|
||||||
|
"""
|
||||||
|
if not isinstance(data, dict):
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
ns = 1e9
|
||||||
|
total = data.get("total_duration", 0) / ns
|
||||||
|
load = data.get("load_duration", 0) / ns
|
||||||
|
peval = data.get("prompt_eval_duration", 0) / ns
|
||||||
|
pcount = data.get("prompt_eval_count")
|
||||||
|
gen = data.get("eval_duration", 0) / ns
|
||||||
|
gcount = data.get("eval_count")
|
||||||
|
reload_flag = " RELOAD" if load > 0.5 else ""
|
||||||
|
print(
|
||||||
|
f" ⏱️ llm[{caller}] ctx={num_ctx} total={total:.2f}s "
|
||||||
|
f"load={load:.2f}s{reload_flag} prompt={peval:.2f}s({pcount}t) "
|
||||||
|
f"gen={gen:.2f}s({gcount}t)",
|
||||||
|
flush=True,
|
||||||
|
)
|
||||||
|
except Exception: # pragma: no cover - logging must never break a reply
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
def call_llm_direct(base_url: str, chat_model: str, system_prompt: str, user_content: str, timeout_sec: float = 10.0, thinking: bool = False, num_ctx: int = OLLAMA_NUM_CTX, temperature: Optional[float] = None) -> Optional[str]:
|
def call_llm_direct(base_url: str, chat_model: str, system_prompt: str, user_content: str, timeout_sec: float = 10.0, thinking: bool = False, num_ctx: int = OLLAMA_NUM_CTX, temperature: Optional[float] = None) -> Optional[str]:
|
||||||
"""Direct LLM call without temporal context, location, or other ask_coach features.
|
"""Direct LLM call without temporal context, location, or other ask_coach features.
|
||||||
|
|
||||||
@@ -62,10 +99,12 @@ def call_llm_direct(base_url: str, chat_model: str, system_prompt: str, user_con
|
|||||||
"keep_alive": "30m",
|
"keep_alive": "30m",
|
||||||
}
|
}
|
||||||
|
|
||||||
|
caller = _caller_name()
|
||||||
try:
|
try:
|
||||||
with requests.post(f"{base_url.rstrip('/')}/api/chat", json=payload, timeout=timeout_sec) as resp:
|
with requests.post(f"{base_url.rstrip('/')}/api/chat", json=payload, timeout=timeout_sec) as resp:
|
||||||
resp.raise_for_status()
|
resp.raise_for_status()
|
||||||
data = resp.json()
|
data = resp.json()
|
||||||
|
_log_ollama_timing(data, num_ctx, caller)
|
||||||
|
|
||||||
if isinstance(data, dict):
|
if isinstance(data, dict):
|
||||||
content = extract_text_from_response(data)
|
content = extract_text_from_response(data)
|
||||||
@@ -233,10 +272,12 @@ def chat_with_messages(
|
|||||||
if tools and isinstance(tools, list) and len(tools) > 0:
|
if tools and isinstance(tools, list) and len(tools) > 0:
|
||||||
payload["tools"] = tools
|
payload["tools"] = tools
|
||||||
|
|
||||||
|
caller = _caller_name()
|
||||||
try:
|
try:
|
||||||
with requests.post(f"{base_url.rstrip('/')}/api/chat", json=payload, timeout=timeout_sec) as resp:
|
with requests.post(f"{base_url.rstrip('/')}/api/chat", json=payload, timeout=timeout_sec) as resp:
|
||||||
resp.raise_for_status()
|
resp.raise_for_status()
|
||||||
data = resp.json()
|
data = resp.json()
|
||||||
|
_log_ollama_timing(data, OLLAMA_NUM_CTX, caller)
|
||||||
if isinstance(data, dict):
|
if isinstance(data, dict):
|
||||||
return data
|
return data
|
||||||
except requests.exceptions.Timeout:
|
except requests.exceptions.Timeout:
|
||||||
|
|||||||
Reference in New Issue
Block a user