エージェント基盤への登録が成功し、テキスト画面から呼び出せる。ところが音声コンパニオンへつなぐと、最初の返答まで無言になったり、割り込んだ後に古い回答が読み上げられたりする——この差は、LLMの賢さではなく登録時の能力宣言と、音声ターンで実際に観測できる挙動を同一視したことから生じます。
本記事では、Gemini側のエージェントをTencent Conversational AIへ接続する前段に、実通信で適格性を判定するVoice Readiness Gateを置きます。登録済みかどうかではなく、次の3段階へ振り分けるのが目的です。
-
VOICE_READY: リアルタイム音声ターンへ投入できる -
TEXT_ONLY: 呼び出せるが、音声では待機表示や明示的な再試行が必要 -
BLOCKED: 相関不能、到達不能などにより利用させない
結論：Agent Cardは発見に使い、音声への投入可否は実通信で決める
音声AIに必要なのは、単に最終回答が返ることではありません。
- アプリが定めた時間内に最初のイベントが届く
- 1回の要求から複数の増分が届く
- すべてのイベントを同じ
requestIdへ関連付けられる - 新しいユーザー発話が始まったら、古いイベントを配信経路から排除できる
- 失敗時に、再試行・テキスト継続・有人対応のどれへ移るかが決まっている
エージェントが自称するstreaming=trueのような宣言は、ルーティング候補を作る材料にはなります。しかし、本番の音声経路へ入れる条件にはしません。デプロイ後のプローブで宣言と実挙動を照合します。
前提：A2A、RTC、LLM、TTSを一つの接続に見立てない
今回の責任分界は次のとおりです。
利用者の音声
↓
RTC／音声入出力
↓
音声認識とターン検出
↓
Voice Readiness Gate
↓
自作Adapter ── 認証・形式変換 ── Gemini側A2Aエージェント
↓
読み上げ可能な応答
↓
音声合成
Tencent Conversational AIは、複数のLLMプロバイダーを利用したリアルタイム音声対話の構成を扱います。全体像は公式のConversational AI overviewで確認できます。
また、公式のLarge Language Model configurationでは、OpenAI互換モデルやDify、Cozeなどのエージェント基盤との接続、リクエスト識別子を使ったルーティング・観測が説明されています。
ただし、ここから「任意のA2AエンドポイントをGemini経由で直接指定できる」とは判断しません。少なくとも本記事で参照するTencent RTC公式資料だけでは、その直接互換性は確認できないためです。そこで、次の境界を持つAdapterを自分たちで用意します。
- 南向き: Gemini側エージェントの認証とA2A通信を担当
- 北向き: Tencent Conversational AIのLLM設定が要求する形式に合わせる
- 内部:
requestId、期限、ストリーム、失敗理由を正規化する
このAdapterがあると、Gemini側の登録方式や認証方式が変わっても、音声ターン制御まで巻き込まずに済みます。
判定に使うデータモデル
能力宣言と観測結果を分離します。
data class AgentManifest(
val agentId: String,
val declaredStreaming: Boolean,
val environment: String
)
data class ProbeEvidence(
val reachable: Boolean,
val correlated: Boolean,
val observedIncremental: Boolean,
val firstEventMs: Long?,
val error: String? = null
)
enum class Admission {
VOICE_READY,
TEXT_ONLY,
BLOCKED
}
AgentManifestはデプロイ設定や登録情報から取得する宣言です。一方、ProbeEvidenceは対象エンドポイントへ実際に要求を送り、受信した結果だけで作ります。
この2種類を一つのcapabilitiesオブジェクトへ混ぜると、「宣言上はストリーミング対応だから正常」という誤判定が起きます。
手順1：Kotlinの検証プロジェクトを作る
次の構成を作成します。
voice-readiness-gate/
├── build.gradle.kts
├── settings.gradle.kts
└── src/main/kotlin/Main.kt
settings.gradle.kts:
rootProject.name = "voice-readiness-gate"
build.gradle.kts:
plugins {
kotlin("jvm") version "2.0.21"
application
}
repositories {
mavenCentral()
}
dependencies {
implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.9.0")
}
application {
mainClass.set("MainKt")
}
ここで指定したバージョンは再現用の一例です。本番では組織の依存関係ポリシーに合わせて固定してください。
手順2：実通信でストリームを検査する
Main.ktへ次のコードを保存します。
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.produceIn
import java.util.UUID
// 登録情報やデプロイ設定から得る「宣言」
data class AgentManifest(
val agentId: String,
val declaredStreaming: Boolean,
val environment: String
)
data class AgentRequest(
val requestId: String,
val text: String
)
data class AgentChunk(
val requestId: String,
val text: String
)
// 実通信から得る「観測結果」
data class ProbeEvidence(
val reachable: Boolean,
val correlated: Boolean,
val observedIncremental: Boolean,
val firstEventMs: Long?,
val error: String? = null
)
data class ProbeBudget(
val firstEventMs: Long,
val secondEventMs: Long
)
enum class Admission {
VOICE_READY,
TEXT_ONLY,
BLOCKED
}
interface AgentClient {
fun stream(request: AgentRequest): Flow<AgentChunk>
}
suspend fun probe(
client: AgentClient,
budget: ProbeBudget
): ProbeEvidence = coroutineScope {
val request = AgentRequest(
requestId = UUID.randomUUID().toString(),
text = "接続確認です。短く応答してください。"
)
val startedAt = System.nanoTime()
val channel = client.stream(request).produceIn(this)
try {
val first = withTimeoutOrNull(budget.firstEventMs) {
channel.receiveCatching().getOrNull()
}
val elapsedMs = (System.nanoTime() - startedAt) / 1_000_000
if (first == null) {
return@coroutineScope ProbeEvidence(
reachable = false,
correlated = false,
observedIncremental = false,
firstEventMs = null,
error = "first_event_timeout"
)
}
val second = withTimeoutOrNull(budget.secondEventMs) {
channel.receiveCatching().getOrNull()
}
val correlated = first.requestId == request.requestId &&
(second == null || second.requestId == request.requestId)
ProbeEvidence(
reachable = true,
correlated = correlated,
observedIncremental = second != null,
firstEventMs = elapsedMs
)
} catch (e: CancellationException) {
throw e
} catch (e: Exception) {
ProbeEvidence(
reachable = false,
correlated = false,
observedIncremental = false,
firstEventMs = null,
error = e::class.simpleName
)
} finally {
// 少なくとも自アプリへの配送を停止する。
// 上流エージェントの計算停止まで保証するものではない。
channel.cancel()
}
}
fun decide(
manifest: AgentManifest,
evidence: ProbeEvidence
): Admission = when {
!evidence.reachable -> Admission.BLOCKED
!evidence.correlated -> Admission.BLOCKED
!manifest.declaredStreaming -> Admission.TEXT_ONLY
!evidence.observedIncremental -> Admission.TEXT_ONLY
else -> Admission.VOICE_READY
}
class FakeAgent(
private val chunks: List<String>,
private val delayMs: Long,
private val corruptRequestId: Boolean = false
) : AgentClient {
override fun stream(request: AgentRequest): Flow<AgentChunk> = flow {
for (part in chunks) {
delay(delayMs)
emit(
AgentChunk(
requestId = if (corruptRequestId) "other-request" else request.requestId,
text = part
)
)
}
}
}
fun main() = runBlocking {
val budget = ProbeBudget(
firstEventMs = 800,
secondEventMs = 800
)
val manifest = AgentManifest(
agentId = "gemini-agent-adapter",
declaredStreaming = true,
environment = "staging"
)
val incrementalAgent = FakeAgent(
chunks = listOf("こんにちは。", "何をお手伝いしましょうか。"),
delayMs = 100
)
val finalOnlyAgent = FakeAgent(
chunks = listOf("こんにちは。何をお手伝いしましょうか。"),
delayMs = 100
)
val wrongCorrelationAgent = FakeAgent(
chunks = listOf("こんにちは。", "続きです。"),
delayMs = 100,
corruptRequestId = true
)
listOf(
"incremental" to incrementalAgent,
"final-only" to finalOnlyAgent,
"wrong-correlation" to wrongCorrelationAgent
).forEach { (name, agent) ->
val evidence = probe(agent, budget)
println("$name: ${decide(manifest, evidence)} / $evidence")
}
}
実行します。
./gradlew run
判定結果は次の種類になります。
incremental: VOICE_READY
final-only: TEXT_ONLY
wrong-correlation: BLOCKED
実測時間は環境によって変わるため、出力値そのものは固定しません。
なぜ最終回答だけのエージェントをBLOCKEDにしないのか
呼び出し自体は成立しているからです。用途を次のように分けられます。
| 判定 | 音声UIでの扱い | テキストUIでの扱い |
|---|---|---|
VOICE_READY |
通常の音声ターンへ投入 | 利用可能 |
TEXT_ONLY |
待機を明示するモード、または利用しない | 利用可能 |
BLOCKED |
ルーティングしない | 原因解消まで停止 |
TEXT_ONLYを無理に音声へ投入すると、画面では許容できる待ち時間が「無言」に変わります。モデルの回答品質が高くても、この会話体験は改善しません。
手順3：割り込み時は上流の停止より先に、古い配送を止める
A2Aクライアントがキャンセルを受け付けても、上流のモデル計算やツール処理が即座に止まるとは限りません。そのため、音声アプリ側では古いターンのイベントを必ず捨てます。
class VoiceTurnRunner(
private val scope: CoroutineScope,
private val client: AgentClient,
private val emitToSpeech: suspend (String) -> Unit
) {
private var generation = 0L
private var activeJob: Job? = null
fun startUserTurn(text: String) {
generation += 1
val myGeneration = generation
activeJob?.cancel()
val request = AgentRequest(
requestId = UUID.randomUUID().toString(),
text = text
)
activeJob = scope.launch {
client.stream(request).collect { chunk ->
if (myGeneration != generation) return@collect
if (chunk.requestId != request.requestId) return@collect
emitToSpeech(chunk.text)
}
}
}
fun interrupt() {
generation += 1
activeJob?.cancel()
activeJob = null
}
}
このコードが保証するのは、キャンセル後の古いイベントをTTSへ渡さないことです。すでにエージェントが開始した外部操作を取り消す保証ではありません。
予約、購入、投稿、機器操作などの副作用を持つエージェントでは、別途次の仕組みが必要です。
- 実行前の明示確認
- 操作ごとの冪等キー
- 最小権限の認証情報
- 実行済みかどうかを確認する業務状態
- 人が停止・修正できる管理経路
手順4：Tencent Conversational AIへ接続する位置
実装時は、Voice Readiness Gateで合格したAdapterの北向きエンドポイントを、Tencent Conversational AIのLLM接続先として構成します。具体的な設定項目と要求形式は、Large Language Model configurationを基準にしてください。
重要なのは、設定画面へGemini側のURLを貼ること自体ではなく、次の情報がAdapter境界を往復できることです。
{
"requestId": "turn-または要求を識別するID",
"agentId": "ルーティング対象",
"input": "音声認識後のテキスト",
"deadlineAt": "アプリが定めた期限",
"stream": true
}
これはTencent RTC固有のAPI形式を示したものではなく、Adapter内部で保持すべき論理データの例です。実際のHTTPフィールドは公式ドキュメントに合わせて変換してください。
また、音声コンパニオンやキャラクター対話はソーシャル体験と結びつきやすい領域です。ユースケースの位置付けは、公式のSocial Entertainment solutionでも確認できます。
確認方法：登録成功ではなく、5つの故障を注入する
ステージング環境で以下を順番に試します。
1. 認証情報を無効化する
期待結果:
- プローブは
BLOCKED - ユーザーの音声をそのエージェントへ送らない
- 失敗理由を画面または運用ログで確認できる
認証失敗をLLMに説明させてはいけません。アプリが決定論的な案内へ切り替えます。
2. 最終回答だけ返す
期待結果:
- 宣言がストリーミング対応でも、観測結果により
TEXT_ONLY - 通常のリアルタイム音声ルートへ昇格しない
3. 異なるrequestIdを返す
期待結果:
BLOCKED- 他ターンの回答を読み上げない
4. 回答途中で新しいユーザー発話を開始する
期待結果:
- 古い
generationのチャンクがTTSへ渡らない - 新しいターンだけが読み上げ対象になる
5. 登録情報は残したままエンドポイントを停止する
期待結果:
- 登録済みという事実に引きずられず
BLOCKED - 再試行を無制限に繰り返さない
しきい値は製品仕様として決める
サンプルでは最初と次のイベントをそれぞれ800ミリ秒待っていますが、これは性能目標ではありません。次を実測し、自分たちの会話設計に合わせて決めます。
- ユーザー発話終了から最初のエージェントイベントまで
- 最初と2番目のイベント間隔
- 割り込みから最後の古いイベント破棄まで
-
TEXT_ONLYまたはBLOCKEDへ落ちた割合 - Adapter、エージェント、ツール処理のどこで期限を超えたか
短すぎる期限は正常なエージェントを排除し、長すぎる期限は無言時間を増やします。単一の値を全ターンへ適用せず、雑談、検索、外部操作などの用途別に決める余地があります。
AIが改善できる範囲と、人が決める範囲
Geminiなどのエージェント基盤は、自然言語から適切なスキルを選び、複数の情報源を使って回答する処理を改善できます。一方、次の判断までモデル能力から自動的に導かれるわけではありません。
- 何秒の無言を許容するか
- 応答不能時にテキストへ切り替えるか
- どの外部操作に確認を求めるか
- どのデータを音声認識・LLM・ログへ残すか
- 利用者が会話や保存を停止できるUIをどこに置くか
「エージェントとして登録できた」は実証可能な能力です。しかし、「音声コンパニオンとして任せられる」は、遅延、割り込み、復旧、権限、プライバシーを含む人間側の製品判断です。Voice Readiness Gateは、その判断を曖昧な期待ではなく、観測可能な条件へ変えるための境界です。
注意点とトレードオフ
- プローブも実リクエストなので、実行頻度に応じて処理量や費用が増えます。デプロイ時、設定変更時、定期監視時を分けてください。
-
channel.cancel()が保証するのはローカル配送の停止です。上流の推論やツール実行が停止したかは、Adapter側で別に観測します。 - プローブ用の入力で実業務ツールを起動しないよう、読み取り専用スキルまたは専用環境を使います。
- 認証情報はクライアントアプリへ置かず、サーバー側Adapterで管理します。
- 音声、文字起こし、会話履歴の保存範囲を明示し、同意撤回と削除経路を用意します。
- 本番では
generationの更新を単一ディスパッチャーやMutexで直列化し、並行呼び出しによる競合を避けます。
関係開示: 筆者はTencent RTCと関係があり、本記事の実装上の参照資料としてTencent RTC公式ドキュメントを使用しています。