音声制御型 AI エージェントの構築パイプラインを解説
本文の状態
日本語全文を表示中
詳細モードで約28分の本文を読めます。
同じ出来事の情報源
この情報源を基点に整理
KDnuggets
本記事は音声 AI エージェント構築において、従来の逐次処理アーキテクチャがもたらす遅延問題を解決し、ストリーミング処理と高度なオーケストレーション技術の重要性を詳述する。
AI深層分析を開く2026年7月31日 23:24
AI深層分析
キーポイント
逐次処理アーキテクチャの限界
従来の STT-LLM-TTS の順次接続方式は各工程が完了まで待機するため遅延が蓄積し、2026 年の生産環境では実用的な会話体験を提供できない。
ストリーミング処理の採用
STT が部分転写を逐次出力し、LLM がトークンをストリーム化して TTS に渡す方式が、自然な対話を実現する生産標準となっている。
音声特有のエンジニアリング課題
音声エージェントの本質的な難しさは言語モデルそのものではなく、遅延管理、ターン取り、ツール呼び出し、および中断処理（バージイン）などのオーケストレーションにある。
実装コンポーネントの分解
記事はストリーミング音声認識、ターン検出、ストリーミング生成、中断処理、音声制約下でのツール呼び出しという各コンポーネントの責任と破綻点を個別に分析する。
逐次パターンの遅延問題
STT、LLM、TTS の各工程が完了するまで待機するため、遅延が積み重なり実用的な応答速度を達成できない。
重要な引用
It's also not the production-standard pattern in 2026, because it's far too slow for anything that needs to feel like a real conversation.
The actual hard part isn't the prompt, and it isn't even the model. It's orchestration: latency, turn-taking, tool calls, and interruption handling.
Voice is a turn-taking problem, not a transcription problem.
Response delays beyond 500ms feel noticeably slow, and delays beyond 3 seconds cause most users to disengage or assume the system is broken.
編集コメントを表示
編集コメント
音声 AI の実装において、モデルの性能以上にシステム全体の設計がユーザー体験を左右するという視点は極めて重要である。開発者は従来の直感的なパイプライン構築から脱却し、非同期処理と高度な制御ロジックへの転換を迫られていると言える。
Source Article
元記事を日本語で読む
本文に関係しない購読案内、埋め込み通知、サイト内プロモーションは除いています。
イントロダクション
音声エージェントの構築というと、多くの人は「音声認識（STT）」「大規模言語モデル（LLM）」「音声合成（TTS）」の 3 つをつなぎ合わせるイメージを抱きます。これらを接続すれば完了だと思いがちです。
この考え方は間違いではありませんし、最もシンプルなアーキテクチャを正しく表しています。各工程が前の工程の完了を待ってから次の工程へ進むという、直列処理のパターンです。
しかし、2026 年においてこれが実用レベルの標準とはなりえません。なぜなら、この方式では遅延が大きすぎて、あたかも生身の人間と会話しているかのような感覚を得ることは不可能だからです。
真に難しいのはプロンプトやモデルそのものではありません。重要なのは「オーケストレーション」、つまり遅延の最小化、ターン（発言権）の制御、ツール呼び出しの実行、そして割り込み処理です。これらを基本的な STT-LLM-TTS のチェーンの上にどう積み上げるかが問われます。
音声エージェントの本質的なエンジニアリング課題とは何かといえば、それは「文字起こし」の問題ではなく、「ターンテイク（会話の受け渡し）」の問題です。自然に感じるエージェントと、チャットボットを電話案内システムにくっつけたような不自然なシステムの差を生むのは、以下の技術的要素にあります。
- 文脈に基づいた発言終了の検出
- バージイン（話中割り込み）によるキャンセル処理
- ストリーミング処理
- 最初のトークン生成までの時間（Time to First Token）
この記事では、音声制御 AI エージェントの構成要素を分解し、ストリーミング音声認識、ターン検知、ストリーミング生成、割り込み処理、そして音声制約下でのツール呼び出しがそれぞれ何を担当し、どこで失敗しやすいかを解説します。また、各責任を具体的に示すテスト済みのコードスニペットも掲載しています。ここで紹介するコードは、ライブマイクや有料 API キーを必要とせず、それぞれのコンポーネントを独立して動作させることで実証されています。これは、システムに必要な機能を決定する前に、実際にどのように思考すべきかを理解するためのものです。
逐次処理のパターンが機能しない理由
まず、他のすべての要素の基盤となるアーキテクチャの選択について考えましょう。これが決まるかどうかが、この記事で扱う課題があなたのシステムに適用されるかどうかを左右します。
逐次処理では、ユーザーが発話すると STT（音声認識）が全文を文字起こしし、LLM が応答をすべて生成し、TTS（音声合成）が完全な音声を出力してから、ようやくユーザーが内容を聞くことができます。このパターンは構築も推論も最も簡単ですが、その分最も遅いのが欠点です。各工程が前の工程の完了を待ってアイドル状態になるため、遅延が積み重なってしまうからです。
ストリーミングパターンが、実際の生産現場では標準となっています。各処理段階で出力を逐次次の工程へ流すことで、STT（音声認識）は不完全な文字起こし結果を LLM に送り、LLM はトークンを TTS（音声合成）にストリーミングします。そして TTS は、LLM がまだ生成を続けている最中にも、最初の文が完成した段階ですぐに音声を合成・再生します。
この仕組みは実装が非常に難しく、中断処理やバッファ管理、部分的な状態の扱いなど、細心の注意が必要です。しかし、この記事ではまさにその点を詳しく解説していきます。なぜなら、これこそが実用的な応答遅延の許容範囲を満たす唯一のパターンだからです。
この許容範囲は単なる理想論ではありません。人間の会話では、話者間の自然な間隔は 200〜300 ミリ秒程度です。応答が遅れて 500 ミリ秒を超えると、明らかに遅く感じられます。さらに 3 秒を超えると、利用者の多くが会話を中断するか、「システムが故障した」と誤解してしまいます。
現在、主要なプロバイダーの音声対話システムは、最初のトークンが出力されるまでの時間が 0.8〜3 秒に集中しています。つまり、実際の応答内容が何もない段階で、アーキテクチャの設計次第で「自然に感じる」領域に入るか、「通話者が切ってしまう」領域に陥るかが決まってしまうのです。
# ストリーミング音声認識
音声エージェントにおける最初のコンポーネントの役割は、「この音声ファイルを文字起こしする」ことではありません。重要なのは、ユーザーが話している最中に着信した音声ストリームを継続的に処理し、逐次テキストを出力して、ユーザーの話が終わったと確信できた時点でシグナルを送ることです。
音声エージェント向けのリアルタイム文字起こし（STT）は、永続的な WebSocket 接続上で動作します。音声データは約 50 ミリ秒ごとの小さなチャンクで送信され、ストリーミング形式で文字起こしのイベントが返されます。これは最後に一度だけテキストを返すブロッキング呼び出しとは異なります。
この区別が重要なのは、ストリーム途中での文字起こし内容がどのように変化するかに関係しています。実際のストリーミング STT エンジンは、新しい音声データが届くたびにモデルが最良の推測を更新する「部分的なイベント」を逐次出力します。そして、単語の内容が確定したと確信できた時点で、最終的なイベントを一つだけ返します。
ここで重要なのは、注文番号や電話番号、固有名詞といったエンティティの精度です。これらがわずかに聞き間違えられるだけで、後続の関数呼び出しが完全に失敗してしまいます。これは人間が「確認してください」と一言声をかけることで防げたはずのミスです。
streaming_stt.py
前提条件：Python 3.10 以上、標準ライブラリのみ使用
実行方法：python streaming_stt.py
import asyncio
from dataclasses import dataclass
from enum import Enum
class TranscriptEventType(Enum):
PARTIAL = "transcript.user.delta" # リアルタイムでまだ変更中の文字起こし
FINAL = "transcript.user" # 確定した、二度と変わらない文字起こし
@dataclass
class TranscriptEvent:
event_type: TranscriptEventType
text: str
confidence: float = 1.0
class MockStreamingSTT:
"""
実際の STT WebSocket 接続の代わりとなるモック実装です。本番環境ではオーディオチャンクを送信し、ユーザーが発話している最中は部分的な更新（デルタ）を、モデルが単語確定と判断した時点で最終的なイベントを受け取ります。
"""
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) # コントロールを返還し、非同期 I/O をシミュレート
yield TranscriptEvent(TranscriptEventType.FINAL, self._partial_stages[-1], confidence=0.97)
async def consume_transcript_stream(stt: MockStreamingSTT):
"""
音声エージェントクライアントが採用すべき基本パターンです。応答性を高めるために部分的な文字起こしをリアルタイムで表示しますが、実際の処理は最終イベント（FINAL）のみに対して行います。発話途中の部分は確定前に変更される可能性があるためです。
"""
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())
How to run**: python streaming_stt.py, no dependencies required.
The downstream code only ever acts on the single FINAL event, even though nine partial transcripts streamed in before it as the simulated utterance built up word by word. That separation — render partials for live feedback, act only on the confirmed final — is what every real streaming STT client implements, whether it's AssemblyAI's Voice Agent API or any other production endpoint.
# Turn Detection: Deciding When the User Is Actually Done
このコンポーネントは、音声認識（STT）の処理の一部のように見えてしまい、つい軽視されがちです。しかし実際には別個の問題として扱うべきものであり、それを独立した機能として捉えることが調整可能性の鍵となります。
ターン検出とは、通話者が話し終わったと判断し、エージェントが応答を開始するタイミングをシステムが決定するための専用ロジックです。これは文字起こしのテキスト内容ではなく、音声ストリーム内の沈黙パターンを分析して動作します。そのため、STT とは明確に異なる処理として実装される必要があります。
この検出の閾値設定を誤ると、会話の流れが破綻する原因になります。検出が早すぎれば、話者が思考のために一瞬止まった瞬間にエージェントが割り込み、会話が不自然になります。逆に遅すぎれば、すべての応答間に不自然な沈黙が生じ、LLM 自体の応答速度が速いにもかかわらずシステム全体が遅く感じられてしまいます。
実運用システムでは、この挙動を2 つの数値で制御します。まず「最小沈黙時間」です。これは通話者が話し終わったと判断するまでの最低限の無音時間を指し、通常は 600ms 程度に設定されます。ただし、これだけで終了を決定するのではなく、文字起こしの内容も文脈的に完了していることを確認した上でターンを終了します。もう一つが「最大沈黙時間（上限）」です。これは曖昧な沈黙が生じた場合でも、強制的に応答を開始させるための閾値で、通常は 1500ms 程度に設定されます。
介護や医療といった慎重な対話が必要な文脈では、この上限を 2500ms 程度まで引き上げる必要があります。一方、会話のテンポが速い場面では、最小沈黙時間を 300ms 程度まで下げることで、よりスムーズな応答を実現できます。
これはシステムアーキテクチャに固定された定数ではなく、ユースケースに応じて調整可能なポリシー決定事項です。
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" # 沈黙を検知したが、まだ判断するほど十分ではない
END_OF_TURN = "end_of_turn"
@dataclass
class AudioFrame:
is_speech: bool
timestamp_ms: int
class TurnDetector:
"""
単独で動作するターン検出状態機械。 (is_speech, timestamp) のフレームストリームを消費し、ユーザーの発話がいつ終了したかを判断します。
STT（音声認識）とは意図的に分離されています。STT は文字起こしを行いますが、ターン検出は音声ストリーム内の沈黙パターンを用いて、いつ聴取を停止してエージェントが応答するべきかを決定します。
"""
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:
"""
utterance_looks_complete は、通訳側から伝わる意味的な信号です。つまり、ユーザーがこれまでに話した内容が一つのまとまった思考として完了しているかどうかを示します。この最小閾値は、その信号が合致した場合のみターンを終了させます。一方、最大閾値では、信号の有無に関わらず強制的に終了します。
"""
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), # brief pause -- a thinking pause
AudioFrame(True, 500), AudioFrame(True, 600), # speaker resumes
AudioFrame(False, 700), AudioFrame(False, 900),
AudioFrame(False, 1100), AudioFrame(False, 1300), # silence clock reaches 600ms here
]
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}")
How to run**: python turn_detection.py, no dependencies required.
t=300ms から t=500ms の間の一時停止は、話者が沈黙の最小閾値に達する前に発話を再開するため、「沈黙待機中」の状態から脱出することはありません。これは文脈を途切れさせるべきではない、文中での一瞬の思考のための一時停止です。一方、t=700ms で実際に話者が止まった場合、カウントは中断されずに続き、正しく t=1300ms で「発話終了」イベントがトリガーされます。
2 回目の実行では、上限値（ceiling）の重要性が浮き彫りになります。意味的な信号が発話がまだ完了していないことを示している場合、最小閾値は無視され、沈黙時間がハードな上限に達するまで発話は終了しません。これが min_silence_ms と max_silence_ms を単一の固定されたタイムアウトではなく、個別に調整可能な 2 つの数値として分離することの最大の価値です。
# Streaming the Response Into Text-to-Speech
**
このセクションでは、最も重要なハンドオフポイントにおいて、前節で説明したストリーミングアーキテクチャを具体的に実装します。発話が検出されたら、LLM は完全な応答が生成されるのを待たずにトークンを逐次ストリームし、TTS（テキスト読み上げ）も応答全体が終わるのを待つのではなく、最初の文が完成した時点で音声合成を開始すべきです。
LLM のストリームから TTS エンジンへ実際に渡される単位は、トークンでも完全な応答でもありません。それは蓄積されたバッファ内で境界が検出された瞬間に識別される「完全な文」です。
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):
"""
実際のストリーミング LLM 呼び出しの代わりを務める関数です。1 トークン（単語）ずつ生成し、レスポンスが一度にすべて表示されるのではなく、断続的に到着する様子をシミュレートします。
"""
for word in text.split(" "):
yield word + " "
await asyncio.sleep(0)
async def stream_sentences(token_stream) -> list[str]:
"""
LLM のストリーミング出力と TTS（音声合成）をつなぐハンドオフユニットです。生トークンではなく、完結した文単位で処理します。
蓄積バッファ内で文の終端が検出された瞬間、その文を即座に返却して TTS が発話を開始できるようにし、LLM は引き続き次の生成を行います。
"""
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)
ストリーミングが終了した後に残った断片も、終端記号がなくても TTS に渡す必要があります。
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())
実行方法：python sentence_chunker.py を実行するだけで、追加の依存関係は不要です。
3 つの文が出力されます。最初の文「Let me check that for you」は、LLM が 3 つ目の文を生成し終わるよりもずっと前に TTS（音声合成）による読み上げを開始できるようになります。この早期の引き渡しこそが、ストリーミング TTS に応答性を感じさせる理由そのものです。ユーザーは、完全なレスポンスが完成するのを待たずに、LLM が生成を開始してから数ミリ秒以内にエージェントの話し声を聞くことができます。
状態を壊さず割り込みに対応する方法
**
バージイン（Barge-in）は、音声エージェントエンジニアリングにおいて最も難しい課題の一つと広く認識されており、ここには特に細心の注意を払う価値があります。バージインを実現するには、以下の 4 つの処理が同時に実行される必要があります：TTS の再生停止、進行中の TTS 生成のキャンセル、LLM 生成のキャンセル、そしてストリーム状態のリセット。これら 4 つのうちどれか一つでも見落とすと、エージェントがユーザーの話しかぶせたり、より混乱を招くことに、中断された後に古い思考を声に出して完了させたりすることになります。これは外側からは診断しにくい不具合のように感じられますが、実際には 4 つのステップを個別に確認していれば原因を特定できるものです。
**
バージイン（話中割り込み）が単に存在するだけでなく信頼できるかどうかを決定するのは、誤検知の防止です。音声活動検出器 (VAD) が背景ノイズや咳、横での会話を本当の割り込みと誤認し、ユーザーには理由が見えないままエージェントが途中で発言を中断してしまう現象を「誤バージイン」と呼びます。
これを防ぐためには、3 つの信号を組み合わせて判断します。まず、フルスケールに対する相対値で -45〜-35 デシベル (dBFS) 程度のエネルギー閾値です。次に、実際の音声とノイズを区別する音声分類器（Silero VAD や WebRTC VAD など）の活用。そして、バージインを実際に発火させる前に、200〜300 ミリ秒にわたって持続的な音声を検出する「最小時間ガード」です。
大きな咳一つでエージェントが途中で止まるべきではありません。まさにそのための機能が、この時間ガードなのです。
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 # signal energy in dBFS
voice_confidence: float # 0.0-1.0, output of a voice classifier like Silero VAD
timestamp_ms: int
class BargeInDetector:
"""
3 つの信号を組み合わせ、ユーザーが本当にエージェントの発言を中断しているのか、それとも背景ノイズや咳、横での会話が誤って音声と認識されているのかを判断します。この 3 つのチェックのうちどれか一つでも欠けると、誤検知（false-barge-in）が発生します。
"""
def __init__(
self,
energy_threshold_dbfs: float = -40.0, # 生産環境では -45 から -35 の範囲内
voice_confidence_threshold: float = 0.6,
min_duration_ms: int = 250, # 生産環境では 200〜300ms の範囲内
):
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:
"""
3 つの条件がすべて連続して少なくとも min_duration_ms の間満たされた瞬間に、真の割り込み（barge-in）が発生したと判断し True を返します。
"""
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__":
# A genuine interruption: strong, sustained voice signal for 300ms
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)}")
A single short cough: high energy but drops immediately, never sustains
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)}")
Loud background noise: passes the energy threshold but fails voice classification
detector_3 = BargeInDetector()
background_noise = [AudioChunk(-32, 0.25, t) for t in range(0, 400, 50)]
fires_3 = [detector_3.process_chunk(c) for c in background_noise]
print(f"Loud non-voice background noise: fired={any(fires_3)}")
How to run: python bargein_detector.py, no dependencies required.
検出器は、持続的な中断に対して正しく作動し、一瞬の咳や音量は大きいものの声に似た背景雑音に対しては適切に沈黙を保ちます。特に注目すべきはこの三つ目のケースです。エネルギー閾値だけを見れば十分な音量であっても、騒がしい部屋では誤ってバージイン（割り込み）が発生し続ける恐れがあります。そのため、音声分類器によるチェックが存在するのです。これは音量だけに頼るのではなく、独立した第二のゲートとして機能しています。
次に進む前に、実務で報告された失敗事例を率直に指摘しておく必要があります。バージインが解除されるタイミングが危険になるケース です。これは、中断信号が発火する前に、下流の自動化処理（予約パイプラインの呼び出しや、すでに実行中のデータベースへの書き込みなど）がトリガーされてしまった場合です。
LLM のトークン生成を途中でキャンセルするのは安全です。単にトークンの出力が止まるだけです。しかし、システムから既に支払いが完了してしまった取引をキャンセルするのは全く別の問題であり、それが次のセクションで「ツール結果のバッファリング」が必要とされる理由です。
# 会話中のツール呼び出し
音声コンテキストにおけるツール呼び出しには、テキストベースのチャットインターフェースでは存在しない特有の問題があります。それは、ツール呼び出しが実行されてから結果が届くまでの間に生じる「間」が、ユーザーに聞こえてしまう点です。
電話通話中に無音状態が続くと、ユーザーは通話が切断されたと誤解し、話し始めてしまいます。その瞬間、まだ処理中のツール呼び出しを邪魔してしまい、エラーを引き起こす原因となります。一方、テキストチャットでは関数が実行されている間の数秒の停止は視覚的に確認できませんが、電話通話では「応答性がある」と感じさせるか、「壊れている」と感じさせるかの決定的な違いになります。
この問題に対する既知の解決策には「プレアム（前文）テクニック」という名前がついています。これは、ツール呼び出しを実行する前後に、モデルが自分の行動をナレーションするように指示する手法です。「それをお調べします」「少々お待ちください」などの言葉を発することで、実際の関数実行中も会話が途切れないように保ちます。
これはコードの修正ではなくプロンプトのパターンですが、音声特有の問題であり、このセクションで扱うもう一つの重大な課題と密接に関連しているため、ここで名前を挙げておく価値があります。
2 つ目の問題は、ツール結果がユーザーに届く前に中断された場合にどうなるかです。ドキュメントに記載されている本番環境のパターンでは、ツール結果が届き次第蓄積し、現在のターンが正常に完了した時点で初めて送信します。もしターンが中断された場合は、保留中の結果をすべて破棄します。
https://www.assemblyai.com/blog/raw-websocket-voice-agent-voice-agent-api
すでにその結果から先へ進んでしまった会話に、古いツール結果を送り込むことは、前のセクションで説明したバージイン（割り込み）処理が防止しようとしている状態の混乱を招くだけです。
tool_result_buffer.py
前提条件：Python 3.10+、標準ライブラリのみ
実行方法：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)")
1 つのターンで複数のツール呼び出しを並行して実行し、まとめて解決する
buffer_3 = ToolResultBuffer()
buffer_3.accumulate("call_weather", {"temp_c": 18})
buffer_3.accumulate("call_calendar", {"next_slot": "2026-06-22T14:00"})
flushed_3 = buffer_3.resolve_turn(TurnOutcome.CLEAN_COMPLETION)
print(f"Parallel tool calls, clean completion: {len(flushed_3)} result(s) sent")
実行方法**: python tool_result_buffer.py（依存ライブラリ不要）
中断されたシナリオでは、ツール呼び出し自体は正常に完了し、有効な確認コードを生成していても、結果として返されるデータはゼロになります。これは意図的な設計です。その結果が到着する頃にはユーザーはすでに会話を別の方向へ進めてしまっており、無理に結果を挿入しても、もはや問われている質問に対する回答になっていないからです。
3 つ目のシナリオでは、このパターンが並行ツール呼び出しでも成り立つことが確認されています。それらの呼び出しが含まれるターンが正常に完了すると、両方の結果が一括して出力されます。これは重要です。なぜなら、最新の音声モデルは並行ツール呼び出しをサポートしており、1 つのターン内で複数のツールを同時に起動できるからです。
コンポーネント間の実際の接続方法
「部品を再結合する」——音声入力を受け取ると、ストリーミング型音声認識（STT）は単語が届くたびに逐次トランスクリプトを出力します。一方、ターン検知はこの同じ音声ストリーム内の沈黙パターンを見守り、「ユーザーが本当に話し終わったか」を判断します。LLM は応答をストリーミングで生成する一方で、TTS（音声合成）は残りの文章が完成する前に、最初の完全な文からすでに発話を開始します。バージイン機能により、ユーザーが再び話し始めた瞬間以降であれば、いつでも会話を中断できます。また、モデルが情報を検索したり行動を実行したりする必要が生じた際、ツール呼び出しは単に沈黙を生むのではなく、そのフローの途中に「前置きとバッファリング」という迂回ルートを挿入します。
ベンダー各社は、この一連のプロセスを単一の WebSocket エンドポイントに統合する傾向が強まっています。AssemblyAI の Voice Agent API や OpenAI の Realtime API、および同様のサービスは、STT、LLM オーケストレーション、TTS、ターン検知、バージインといった機能をサーバー側で単一接続上で処理します。そのため、2026 年に音声エージェントを構築するチームの多くが、本記事で解説した 5 つのコンポーネントを手作業で実装するのではなく、これらのサービスの一つに統合するのが合理的な選択となっています。これは妥当なデフォルトと言えます。
しかし、「音声エージェント」が全体として処理しているという漠然とした理解ではなく、各コンポーネントが具体的に何を行うかを把握することが重要です。これこそが、「エージェントが壊れているように感じる」という謎を、デバッグ可能な問題へと変える鍵となります。例えば、バージインの閾値が過度に敏感であること、ツール呼び出しが遅延する際に前置きが欠落していること、あるいはわずかな VAD（音声活動検知）の設定変更で検出できたはずの注文番号のトランスクリプトエラーなどが該当します。
# 結論
音声エージェントは、片端にマイクをテープで貼り付け、もう片端にスピーカーを取り付けたチャットボットではありません。テキストベースの会話には存在しない 5 つの問題を解決する 5 つのコンポーネントから成り立っています。
具体的には、ブロック型の呼び出しではなくストリーミングトランスクリプションを行うこと、固定されたタイムアウトではなく個別に調整可能なポリシーとしてのターン検出、LLM から TTS への文レベルでのハンドオフ（応答全体を待たない）、単一のノイズ閾値ではなく 3 つの信号を組み合わせて構築されるバージイン検出、そして結果が到着する前にユーザーが次の行動に移ることを考慮したツール呼び出し結果のバッファリングです。
これらすべての基盤となるレイテンシ予算は容赦がありません。500 ミリ秒ほどが「自然に感じるか」「明らかに遅いと感じるか」の境界線であり、各コンポーネントはこの予算を守るか、あるいは静かにそれを破るかのどちらかです。
ほとんどのチームは、これら 5 つをゼロから実装するのではなく、統合されたリアルタイム API を基盤に構築するのが現実的な選択でしょう。しかし、各パーツが具体的に何の責任を負っているかを正確に理解しておくことが、不自然な音声エージェントを単に再起動して運を待つのではなく、実際に修正可能にするための鍵となります。
リソース:
- AssemblyAI の Voice Agent API を使って 5 分で音声エージェントを構築する
- AssemblyAI の Voice Agent API を使った生 WebSocket による音声エージェント
原文を表示
**
# Introduction
Most people picture building a voice agent as stitching three things together: speech-to-text (STT), a large language model (LLM), and text-to-speech (TTS). Wire them up, and you're done. That picture is correct as far as it goes, and it describes the simplest architecture, where each stage waits for the previous one to fully complete before starting. It's also not the production-standard pattern in 2026, because it's far too slow for anything that needs to feel like a real conversation.
The actual hard part isn't the prompt, and it isn't even the model. It's orchestration: latency, turn-taking, tool calls, and interruption handling, layered on top of that basic STT-LLM-TTS chain. This is the actual engineering challenge precisely: voice is a turn-taking problem, not a transcription problem; semantic end-of-turn detection, barge-in cancellation, streaming, and time-to-first-token are the levers that separate a voice agent that feels natural from one that feels like a phone tree with a chatbot bolted onto it.
This article breaks the pipeline into its real components — streaming speech recognition, turn detection, streaming generation, interruption handling, and tool calling under voice constraints — and shows what each one is responsible for, where it actually breaks, and includes a tested code excerpt that makes the responsibility concrete. None of the code here needs a live microphone or a paid API key to run; each component is demonstrated in isolation, the way you'd actually reason about it before deciding what your system needs.
# Why the Sequential Pattern Doesn't Work
Start with the architecture choice underneath everything else, because it determines whether the rest of this article's concerns even apply to your system.
In the sequential pattern, the user speaks, STT transcribes the full utterance, the LLM generates the full response, TTS synthesizes the full audio, and only then does the user hear anything. It's the simplest pattern to build and reason about. It's also the slowest, because every stage sits idle waiting for the one before it to fully finish, and those delays stack on top of each other.
The streaming pattern is the production standard instead: each stage streams its output to the next incrementally. STT streams partial transcripts to the LLM, the LLM streams tokens to TTS, and TTS synthesizes and plays audio from the first complete sentence while the LLM is still generating everything after it. This is genuinely harder to build; it demands careful handling of interruptions, buffering, and partial state, which is exactly what the rest of this article walks through, but it's the only pattern that hits a usable latency budget.
That budget isn't a vague aspiration. Human conversation has a natural 200 to 300ms gap between speakers. Response delays beyond 500ms feel noticeably slow, and delays beyond 3 seconds cause most users to disengage or assume the system is broken. Current speech-to-speech systems cluster in the 0.8 to 3 second time-to-first-token range across leading providers, which means the architecture decision alone is what determines whether your agent lands in the "feels natural" zone or the "caller hangs up" zone, before a single word of the actual response has been considered.
# Streaming Speech-to-Text
The first component's job in a voice agent is not "transcribe this audio file." It's continuously processing an incoming audio stream and emitting transcripts as the user is still speaking, then signaling once it's confident they've finished. Production STT for voice agents runs over a persistent WebSocket connection. Audio goes out in small chunks, roughly 50ms at a time, and streaming transcript events come back — not a single blocking call that returns text once at the very end.
This distinction matters because of how the transcript actually changes mid-stream. A real streaming STT engine emits partial events that update as more audio arrives and the model revises its best guess, followed by one final event once it's confident the words have settled. Accuracy on entities — order numbers, phone numbers, and proper nouns — matters disproportionately here, because a single misheard digit breaks a downstream function lookup entirely, in a way that a human listener would have caught by simply asking the caller to confirm.
# 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())How to run**: python streaming_stt.py, no dependencies required.
The downstream code only ever acts on the single FINAL event, even though nine partial transcripts streamed in before it as the simulated utterance built up word by word. That separation — render partials for live feedback, act only on the confirmed final — is what every real streaming STT client implements, whether it's AssemblyAI's Voice Agent API or any other production endpoint.
# Turn Detection: Deciding When the User Is Actually Done
**
This component is easy to skip mentally because it feels like it should just be part of the STT step. It isn't, and treating it as a separate concern is what makes it tunable. Turn detection is the system's specific method for deciding when the caller has finished speaking and the agent should respond, and it consumes the audio stream's silence pattern, not the transcript's text content, which is why it's a distinct piece of logic from STT.
Get this wrong in either direction, and the conversation breaks differently. Too eager, and the agent interrupts a speaker who paused mid-thought to think. Too slow, and every single exchange carries an awkward dead-air gap that makes the whole system feel sluggish even when the LLM itself responds instantly. Production systems control this with two numbers: a minimum silence duration before declaring end-of-turn, commonly around 600ms, which ends the turn only when the transcript side also suggests the utterance sounds finished, and a maximum silence ceiling that forces a response even on an ambiguous pause, often around 1500ms. Deliberate-speech contexts like eldercare or healthcare warrant raising that ceiling toward 2500ms; fast-paced conversational contexts warrant dropping the minimum toward 300ms. This is a tunable policy decision specific to your use case, not a fixed constant baked into the architecture.
# 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:
"""
utterance_looks_complete carries the semantic signal from the
transcript side -- whether what the user has said so far sounds
like a finished thought. The minimum threshold ends the turn only
when that signal agrees; the maximum threshold ends it regardless.
"""
if frame.is_speech:
# Any speech resets the silence clock entirely
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 # hard ceiling -- force a response
elif silence_duration >= self.min_silence_ms and utterance_looks_complete:
self.state = TurnState.END_OF_TURN # confident enough silence has settled
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), # brief pause -- a thinking pause
AudioFrame(True, 500), AudioFrame(True, 600), # speaker resumes
AudioFrame(False, 700), AudioFrame(False, 900),
AudioFrame(False, 1100), AudioFrame(False, 1300), # silence clock reaches 600ms here
]
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}")How to run**: python turn_detection.py, no dependencies required.
The pause between t=300ms and t=500ms never escalates past silence_pending, because the speaker resumes before the silence clock crosses the minimum threshold — exactly the kind of mid-sentence thinking pause that shouldn't end the turn. Once the speaker actually stops at t=700ms, the clock runs uninterrupted and correctly fires end_of_turn at t=1300ms. The second run is where the ceiling earns its place: with the semantic signal saying the utterance still sounds unfinished, the minimum threshold is ignored entirely and the turn ends only once silence hits the hard ceiling. That's the entire value of separating min_silence_ms and max_silence_ms as two distinct, tunable numbers rather than a single fixed timeout.
# Streaming the Response Into Text-to-Speech
**
This section makes the streaming architecture from the first section concrete at the handoff point that matters most. Once a turn is detected, the LLM should stream tokens as they're generated rather than waiting for the full response, and TTS should begin synthesizing audio from the first complete sentence rather than waiting for the entire reply. The unit that actually gets handed from the LLM stream to the TTS engine isn't a token and isn't the full response; it's a complete sentence, detected the instant its boundary appears in the accumulating buffer.
# 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):
"""
Stands in for a real streaming LLM call. Yields one token (word) at
a time, simulating tokens arriving incrementally rather than the
full response appearing all at once.
"""
for word in text.split(" "):
yield word + " "
await asyncio.sleep(0)
async def stream_sentences(token_stream) -> list[str]:
"""
The handoff unit between LLM streaming and TTS synthesis: complete
sentences, not raw tokens. The instant a sentence boundary appears
in the accumulated buffer, that sentence is yielded so TTS can start
speaking it while the LLM is still generating what comes after it.
"""
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)
# Whatever remains once the stream ends is the final fragment --
# still needs to be flushed to TTS even without terminal punctuation.
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())How to run**: python sentence_chunker.py, no dependencies required.
Three sentences come out, and the first one, "Let me check that for you," is ready for TTS to start speaking well before the LLM has finished composing the third. That early handoff is the entire reason streaming TTS feels responsive: the user hears the agent start talking within a few hundred milliseconds of the LLM beginning to generate, instead of waiting for the complete response to finish first.
# Handling Interruption Without Breaking State
**
Barge-in is widely treated as the single hardest part of voice agent engineering, and it's worth spending the most care on here too. Barge-in requires four things to happen together: stopping TTS playback, canceling in-flight TTS generation, canceling LLM generation, and resetting stream state. Miss any one of these, and the agent either talks over the user or, more confusingly, finishes its old thought out loud after being interrupted, which feels broken in a way that's hard to diagnose from the outside if you don't already know to look at all four steps individually.
The piece that determines whether barge-in is reliable rather than just present is false-positive prevention. False-barge-in fires when the voice activity detector (VAD) mistakes background noise, a cough, or a side conversation for a genuine interruption, and the agent cuts itself off mid-sentence for no reason the user can perceive. Prevention combines three signals: an energy threshold, typically -45 to -35 decibels relative to full scale (dBFS), a voice classifier such as Silero VAD or WebRTC VAD** that distinguishes actual speech from noise, and a minimum-duration guard requiring 200 to 300ms of sustained voice before the barge-in actually fires. A single loud cough should never stop the agent mid-sentence; that's specifically what the duration guard exists to prevent.
# 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 # signal energy in dBFS
voice_confidence: float # 0.0-1.0, output of a voice classifier like Silero VAD
timestamp_ms: int
class BargeInDetector:
"""
Combines three signals to decide whether the user is genuinely
interrupting the agent, or whether background noise, a cough, or
a side conversation is being mistaken for real speech. Missing any
one of these three checks is what causes false-barge-in.
"""
def __init__(
self,
energy_threshold_dbfs: float = -40.0, # within the -45 to -35 production range
voice_confidence_threshold: float = 0.6,
min_duration_ms: int = 250, # within the 200-300ms production range
):
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:
"""
Returns True the instant a real barge-in should fire -- i.e. all
three conditions have held continuously for at least min_duration_ms.
"""
passes_energy = chunk.energy_dbfs > self.energy_threshold
passes_voice = chunk.voice_confidence > self.voice_threshold
if not (passes_energy and passes_voice):
# Signal dropped below threshold -- reset the candidate window so a
# brief loud noise can't accumulate duration across separate bursts.
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__":
# A genuine interruption: strong, sustained voice signal for 300ms
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)}")
# A single short cough: high energy but drops immediately, never sustains
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)}")
# Loud background noise: passes the energy threshold but fails voice classification
detector_3 = BargeInDetector()
background_noise = [AudioChunk(-32, 0.25, t) for t in range(0, 400, 50)]
fires_3 = [detector_3.process_chunk(c) for c in background_noise]
print(f"Loud non-voice background noise: fired={any(fires_3)}")How to run: python bargein_detector.py, no dependencies required.
The detector fires on the genuine sustained interruption and correctly stays silent on both the brief cough and the loud-but-not-voice-like background noise. That third case is the one worth dwelling on: noise that's loud enough to pass the energy threshold alone would trigger a false barge-in constantly in a noisy room, which is exactly why the voice classifier check exists as a second, independent gate rather than relying on volume alone.
One production-reported failure mode is worth naming plainly before moving on: barge-in teardown becomes genuinely dangerous when downstream automation has already triggered before the interruption fires — a booking pipeline call, or a database write that's already in flight. Canceling LLM token generation mid-stream is safe; the tokens just stop. Canceling a payment that's already left your system is a different problem entirely, and it's the reason the next section's tool-result buffering exists.
# Tool Calling Mid-Conversation
**
Tool calling in a voice context has a problem that simply doesn't exist in a text-based chat interface: the gap between a tool call firing and its result arriving is audible. Dead air during a phone call makes users assume the call dropped, which prompts them to start talking and interrupt the tool call that's still in progress. In a text chat, a three-second pause while a function executes is invisible. On a phone call, it's the difference between feeling responsive and feeling broken.
The documented fix has a name: the preamble technique, instructing the model to narrate what it's doing before and during a tool call, saying something like "Let me check that for you" or "One moment while I pull that up," which keeps the conversation audibly alive while the function actually executes. It's a prompting pattern, not a code pattern, but it solves a problem that's specific to voice and worth naming here because it pairs directly with the second hard problem this section covers.
That second problem is what happens to a tool result if the user interrupts before it ever arrives. The documented production pattern is to accumulate tool results as they come in, and only actually send them once the current turn finishes cleanly, discarding any pending results entirely if the turn was interrupted instead. Sending a stale tool result into a conversation that's already moved on past it creates exactly the kind of state confusion the previous section's barge-in handling exists to prevent in the first place.
# 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:
"""
Implements the documented production pattern: accumulate tool
results as they arrive mid-turn, but only send them once the
current turn finishes cleanly. If the turn was interrupted
instead, discard everything pending -- sending a stale tool
result into a conversation that already moved on is worse
than not responding to the tool call at all.
"""
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]:
"""
Called when the current conversational turn ends. Flushes every
pending result downstream on a clean completion, or discards all
of them on an interruption -- there's no partial-credit path here.
"""
pending = list(self._pending)
self._pending.clear()
if outcome == TurnOutcome.CLEAN_COMPLETION:
return pending
return [] # interrupted -- discard everything, send nothing
if __name__ == "__main__":
# Tool call resolves, turn completes cleanly -- result is sent
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}")
# Tool call resolves, but user interrupts before the turn completes --
# the result must be discarded, not sent into a conversation that moved on
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)")
# Multiple parallel tool calls in one turn, resolved together
buffer_3 = ToolResultBuffer()
buffer_3.accumulate("call_weather", {"temp_c": 18})
buffer_3.accumulate("call_calendar", {"next_slot": "2026-06-22T14:00"})
flushed_3 = buffer_3.resolve_turn(TurnOutcome.CLEAN_COMPLETION)
print(f"Parallel tool calls, clean completion: {len(flushed_3)} result(s) sent")How to run**: python tool_result_buffer.py, no dependencies required.
The interrupted scenario sends zero results, even though the tool call itself completed successfully and produced a perfectly valid confirmation code. That's deliberate: the user has already moved the conversation somewhere else by the time that result would arrive, and injecting it anyway would be answering a question that's no longer the one being asked. The third scenario confirms the pattern holds for parallel tool calls too — both results flush together once the turn that contained them resolves cleanly, which matters because modern voice models support parallel tool calling, meaning multiple tools can fire simultaneously within a single turn.
# How the Components Actually Connect
**
Putting the pieces back together: audio comes in, streaming STT emits partial transcripts as the words arrive, turn detection watches the silence pattern in that same audio stream and decides when the user has actually finished, the LLM streams a response while TTS begins speaking the first complete sentence well before the rest has been generated, barge-in can interrupt at any point downstream of the user starting to talk again, and a tool call, when the model needs to look something up or take an action, inserts a preamble-and-buffer detour into the middle of that flow rather than just leaving dead air.
Vendors increasingly bundle this entire chain into a single WebSocket endpoint: AssemblyAI's Voice Agent API, OpenAI's Realtime API**, and similar offerings handle STT, LLM orchestration, TTS, turn detection, and barge-in server-side over one connection, which is why most teams building voice agents in 2026 reasonably integrate against one of these rather than hand-rolling all five components covered in this article. That's a sound default. But understanding what each component does specifically, not just that "the voice agent" handles it, is what turns "the agent feels broken" from a mystery into a debuggable problem: a too-eager barge-in threshold, a missing preamble during a slow tool call, a transcript error on an order number that a slightly different VAD tuning would have caught.
# Conclusion
**
A voice agent is not a chatbot with a microphone taped to one end and a speaker to the other. It's five components solving five problems that don't exist at all in text-based conversation: streaming transcription instead of a blocking call, turn detection as its own tunable policy rather than a fixed timeout, sentence-level handoff from the LLM to TTS instead of waiting for the full response, barge-in detection built from three combined signals rather than a single noise threshold, and tool-call result buffering that accounts for the user moving on before the result arrives.
The latency budget underneath all of it is unforgiving — 500ms is roughly the line between feeling natural and feeling noticeably slow — and every one of these components either protects that budget or quietly breaks it. Most teams will reasonably build on a bundled realtime API rather than implementing all five from first principles. But knowing precisely what each piece is responsible for is what makes it possible to actually fix a voice agent that feels wrong, instead of just restarting it and hoping.
Resources**:
- Build a Voice Agent in 5 Minutes with AssemblyAI's Voice Agent API
- Raw WebSocket Voice Agent with AssemblyAI's Voice Agent API
Shittu Olumide is a software engineer and technical writer passionate about leveraging cutting-edge technologies to craft compelling narratives, with a keen eye for detail and a knack for simplifying complex concepts. You can also find Shittu on Twitter.
関連記事
今日のまとめ
AIデイリーブリーフで今日の重要ニュースをまとめ読み