เจาะลึกวิศวกรรมการสร้าง Voice AI Agent: เมื่อเสียงเป็นมากกว่าแค่การพิมพ์

# streaming_stt.py
# Prerequisites: Python 3.10+, standard library only
# Run: python streaming_stt.py
 
import asyncio
from dataclasses import dataclass
from enum import Enum
 
class TranscriptEventType(Enum):
    PARTIAL = "transcript.user.delta"   # live, still-changing transcript
    FINAL = "transcript.user"            # confirmed, won't change again
 
@dataclass
class TranscriptEvent:
    event_type: TranscriptEventType
    text: str
    confidence: float = 1.0
 
class MockStreamingSTT:
    """
    Stands in for a real STT WebSocket connection. Real implementations
    send audio chunks and receive these same two event types back --
    partial deltas while the user is mid-utterance, then one final
    event once the model is confident the words are settled.
    """
    def __init__(self, simulated_utterance: str):
        words = simulated_utterance.split()
        self._partial_stages = [" ".join(words[:i]) for i in range(1, len(words) + 1)]
 
async def stream_events(self):
        for stage in self._partial_stages[:-1]:
            yield TranscriptEvent(TranscriptEventType.PARTIAL, stage, confidence=0.7)
            await asyncio.sleep(0)   # yield control, simulating real async I/O
        yield TranscriptEvent(TranscriptEventType.FINAL, self._partial_stages[-1], confidence=0.97)
 
async def consume_transcript_stream(stt: MockStreamingSTT):
    """
    The pattern every voice agent client implements: render partial
    transcripts live for responsiveness, but only act on the FINAL
    event downstream -- partials can and do change before that.
    """
    final_transcript = None
    partial_count = 0
 
async for event in stt.stream_events():
        if event.event_type == TranscriptEventType.PARTIAL:
            partial_count += 1
            print(f"  [partial] '{event.text}' (confidence={event.confidence})")
        elif event.event_type == TranscriptEventType.FINAL:
            final_transcript = event.text
            print(f"  [FINAL]   '{event.text}' (confidence={event.confidence})")
 
return final_transcript, partial_count
 
async def main():
    stt = MockStreamingSTT("My order number is A B 3 7 9 2")
    final_text, n_partials = await consume_transcript_stream(stt)
    print(f"\nFinal transcript used downstream: '{final_text}'")
    print(f"Partial events received before final: {n_partials}")
 
asyncio.run(main())

# turn_detection.py
# Prerequisites: Python 3.10+, standard library only
# Run: python turn_detection.py
 
from dataclasses import dataclass
from enum import Enum
 
class TurnState(Enum):
    LISTENING = "listening"
    SILENCE_PENDING = "silence_pending"   # silence detected, not yet long enough to decide
    END_OF_TURN = "end_of_turn"
 
@dataclass
class AudioFrame:    is_speech: bool
    timestamp_ms: int
 
class TurnDetector:
    """
    Standalone turn-detection state machine -- consumes a stream of
    (is_speech, timestamp) frames and decides when the user has
    finished speaking. Deliberately separate from STT: STT produces
    transcripts; turn detection decides WHEN to stop listening and
    let the agent respond, using the silence pattern in the audio
    stream itself.    """
    def __init__(self, min_silence_ms: int = 600, max_silence_ms: int = 1500):
        self.min_silence_ms = min_silence_ms
        self.max_silence_ms = max_silence_ms
        self._silence_start: int | None = None
        self.state = TurnState.LISTENING
 
def process_frame(self, frame: AudioFrame, utterance_looks_complete: bool = True) -> TurnState:
        if frame.is_speech:
            self._silence_start = None
            self.state = TurnState.LISTENING
            return self.state
 
if self._silence_start is None:
            self._silence_start = frame.timestamp_ms
 
silence_duration = frame.timestamp_ms - self._silence_start
 
if silence_duration >= self.max_silence_ms:
            self.state = TurnState.END_OF_TURN
        elif silence_duration >= self.min_silence_ms and utterance_looks_complete:
            self.state = TurnState.END_OF_TURN
        else:
            self.state = TurnState.SILENCE_PENDING
 
return self.state
 
if __name__ == "__main__":
    print("Complete-sounding utterance -- the minimum threshold applies:")
    detector = TurnDetector(min_silence_ms=600, max_silence_ms=1500)
    frames = [
        AudioFrame(True, 0), AudioFrame(True, 100), AudioFrame(True, 200),
        AudioFrame(False, 300), AudioFrame(False, 400),
        AudioFrame(True, 500), AudioFrame(True, 600),
        AudioFrame(False, 700), AudioFrame(False, 900),
        AudioFrame(False, 1100), AudioFrame(False, 1300),
    ]
 
for f in frames:
        state = detector.process_frame(f)
        print(f"  t={f.timestamp_ms:>5}ms speech={f.is_speech!s:>5} -> {state.value}")
 
print("\nUtterance that still sounds unfinished -- the ceiling applies:")
    trailing_detector = TurnDetector(min_silence_ms=600, max_silence_ms=1500)
    trailing_frames = [AudioFrame(True, 0)] + [AudioFrame(False, t) for t in range(100, 1800, 400)]
 
for f in trailing_frames:
        state = trailing_detector.process_frame(f, utterance_looks_complete=False)
        print(f"  t={f.timestamp_ms:>5}ms speech={f.is_speech!s:>5} -> {state.value}")

# sentence_chunker.py
# Prerequisites: Python 3.10+, standard library only
# Run: python sentence_chunker.py
 
import asyncio
import re
 
SENTENCE_END_PATTERN = re.compile(r'(?<=[.!?])\s+')
 
async def mock_llm_token_stream(text: str):
    for word in text.split(" "):
        yield word + " "
        await asyncio.sleep(0)
 
async def stream_sentences(token_stream) -> list[str]:
    buffer = ""
    sentences = []
 
async for token in token_stream:
        buffer += token
        match = SENTENCE_END_PATTERN.search(buffer)
        while match:
            sentence = buffer[:match.start() + 1].strip()
            sentences.append(sentence)
            print(f"  [sentence ready for TTS] '{sentence}'")
            buffer = buffer[match.end():]
            match = SENTENCE_END_PATTERN.search(buffer)
 
if buffer.strip():
        sentences.append(buffer.strip())
        print(f"  [final fragment flushed] '{buffer.strip()}'")
 
return sentences
 
async def main():
    text = (
        "Let me check that for you. Your order shipped yesterday and "
        "should arrive Thursday. Is there anything else I can help with"
    )
    sentences = await stream_sentences(mock_llm_token_stream(text))
    print(f"\nTotal sentences yielded: {len(sentences)}")
 
asyncio.run(main())

# bargein_detector.py
# Prerequisites: Python 3.10+, standard library only
# Run: python bargein_detector.py
 
from dataclasses import dataclass
 
@dataclass
class AudioChunk:
    energy_dbfs: float
    voice_confidence: float
    timestamp_ms: int
 
class BargeInDetector:
    def __init__(
        self,        energy_threshold_dbfs: float = -40.0,
        voice_confidence_threshold: float = 0.6,
        min_duration_ms: int = 250,
    ):
        self.energy_threshold = energy_threshold_dbfs
        self.voice_threshold = voice_confidence_threshold
        self.min_duration_ms = min_duration_ms
        self._candidate_start_ms: int | None = None
 
def process_chunk(self, chunk: AudioChunk) -> bool:
        passes_energy = chunk.energy_dbfs > self.energy_threshold
        passes_voice = chunk.voice_confidence > self.voice_threshold
 
if not (passes_energy and passes_voice):
            self._candidate_start_ms = None
            return False
 
if self._candidate_start_ms is None:
            self._candidate_start_ms = chunk.timestamp_ms
 
sustained_duration = chunk.timestamp_ms - self._candidate_start_ms
        return sustained_duration >= self.min_duration_ms
 
if __name__ == "__main__":
    detector_1 = BargeInDetector()
    real_interruption = [AudioChunk(-30, 0.9, t) for t in range(0, 350, 50)]
    fires_1 = [detector_1.process_chunk(c) for c in real_interruption]
    print(f"Real interruption (sustained 300ms):      fired={any(fires_1)}")
 
detector_2 = BargeInDetector()
    cough = [AudioChunk(-28, 0.8, 0), AudioChunk(-50, 0.1, 50), AudioChunk(-50, 0.1, 100)]
    fires_2 = [detector_2.process_chunk(c) for c in cough]
    print(f"Single cough (<100ms):                    fired={any(fires_2)}")

# tool_result_buffer.py
# Prerequisites: Python 3.10+, standard library only
# Run: python tool_result_buffer.py
 
from dataclasses import dataclass
from enum import Enum
 
class TurnOutcome(Enum):
    CLEAN_COMPLETION = "clean_completion"
    INTERRUPTED = "interrupted"
 
@dataclass
class PendingToolResult:
    call_id: str
    result: dict
 
class ToolResultBuffer:
    def __init__(self):
        self._pending: list[PendingToolResult] = []
 
def accumulate(self, call_id: str, result: dict) -> None:
        self._pending.append(PendingToolResult(call_id, result))
 
def resolve_turn(self, outcome: TurnOutcome) -> list[PendingToolResult]:
        pending = list(self._pending)
        self._pending.clear()
        if outcome == TurnOutcome.CLEAN_COMPLETION:
            return pending
        return []
 
if __name__ == "__main__":
    buffer_1 = ToolResultBuffer()
    buffer_1.accumulate("call_abc123", {"temp_c": 22, "condition": "sunny"})
    flushed_1 = buffer_1.resolve_turn(TurnOutcome.CLEAN_COMPLETION)
    print(f"Clean completion:  {len(flushed_1)} result(s) sent -> {flushed_1}")
 
buffer_2 = ToolResultBuffer()
    buffer_2.accumulate("call_def456", {"confirmation_code": "CONF7821"})
    flushed_2 = buffer_2.resolve_turn(TurnOutcome.INTERRUPTED)
    print(f"Interrupted turn:  {len(flushed_2)} result(s) sent (correctly discarded)")