初めに
業務で英語を使う機会があり、即時通訳してくれるAIエージェントがいてほしいなぁ
ーー＞よっしゃ自前で作ろう！という企画です。翻訳AIエージェント構築してみました！
やりたいこと/ゴール
動画は記事上に載せられないので、翻訳処理をしているGIF画像を掲載します。
英語を検知したら翻訳した日本語を、日本語を検知したら翻訳した英語を画面上で表示するタスクを、いかに体感の待ち時間を抑えて実現できるかが勝負です。
想定読者
- StrandsAgents,BedrockAgentCore等の活用に興味がある方
- Nova 2 SonicをAmazon Connect以外で利用してみたい方
- 音声AIエージェントに興味がある方
BidiエージェントとAmazon Nova 2 Sonic
今回の翻訳アプリの裏側では、BidiAgent を活用して音声をストリーミング処理し、フロントエンドとWebSocketで通信する構成をとっています。
AIエージェントとして Amazon Nova 2 Sonic を利用しました。
Amazon Nova 2 SonicはS2S(Speech-to-Speech)モデルで、音声を入力として受け取り、音声をレスポンスすることができる生成AIモデルです。
Outputの音声に関しては、日本語が非対応(公式ドキュメント参照)であることに注意が必要です。
このモデルはフレームワークStrandsAgentsのBidiAgentを利用することで、駆動させます。
BidiAgentでは双方向のストリーミング処理を行うことができ、音声、テキスト、画像に対応しています。
そのため適切なInput,Outputのevent形式を設定することで、入力は音声、出力はテキストを実現することができます。
入力編_発話音声をinput
BidiAudioInputEventという下記イベント形式で入力することができます。
音声をbase64でエンコードすることで、テキスト形式のイベントとして取り扱うことができます。
import base64
from strands.experimental.bidi.types.events import BidiAudioInputEvent
audio_bytes = record_audio() # Your audio capture logic
audio_base64 = base64.b64encode(audio_bytes).decode('utf-8')
await agent.send(BidiAudioInputEvent(
audio=audio_base64,
format="pcm",
sample_rate=16000,
channels=1
))
出力編_翻訳結果のテキストをoutput
BidiTranscriptStreamEventという下記イベント形式でレスポンスされます。
{
"type": "bidi_transcript_stream",
"delta": {"text": "Hello"},
"text": "Hello",
"role": "assistant",
"is_final": True,
"current_transcript": "Hello world"
}
ここで注目したい重要なキーは以下の2つです。
-
role:userがユーザー発話の文字起こし、assistantが翻訳結果を指します。 -
is_final:falseなら翻訳の途中結果（partial）、trueなら確定結果（final）となります。
したがって、フロントエンド側では以下のレスポンスパターンに注意しながら、表示の実装を行う必要があります。
- ユーザーの音声入力内容を受領したことを伝えるレスポンス(要するに発話の文字起こし)
- エージェントが翻訳した内容のレスポンス (途中結果)
- is_finalがTrueであり、発話自体がひとまずの区切り & 発話全文の翻訳をレスポンス
// ユーザーの音声入力の文字起こし（確定前）
{
"type": "bidi_transcript_stream",
"role": "user",
"is_final": false,
"current_transcript": "we have no physical branches..."
}
// モデルからのAI翻訳結果（確定前）
{
"type": "bidi_transcript_stream",
"role": "assistant",
"is_final": false,
"current_transcript": "私たちは物理的な支店を持っていません。"
}
実際のBidiエージェント実装内容解説
ベースは公式サンプル実装を参考にしましたが、入力時にコールバック関数(lambda式)を渡して、キューを挟む観点は工夫しました。
とにかく応答速度を上げるために、入力と出力を疎結合にしたいと考え、
- websocketからの受信を止めずに読み続ける
- その裏で agent に入力を渡し続ける
- agent から出る途中結果を即座に返し続ける
上記を実現できるような実装を心がけました。
app = FastAPI()
async def receive_from_frontend(
websocket: WebSocket,
input_queue: asyncio.Queue,
) -> None:
while True:
payload = await websocket.receive_json()
await input_queue.put(payload)
async def read_from_queue(input_queue: asyncio.Queue) -> dict:
return await input_queue.get()
async def send_to_frontend(websocket: WebSocket, payload: dict) -> None:
await websocket.send_json(payload)
@app.websocket("/ws/bidi")
async def websocket_bidi(websocket: WebSocket):
await websocket.accept()
agent = BidiAgent(
model=BidiNovaSonicModel(
model_id="amazon.nova-2-sonic-v1:0",
provider_config={"audio": {"voice": "tiffany"}},
),
system_prompt=(
"あなたはリアルタイム翻訳エージェントです。"
"日本語は英語に、英語は日本語に逐次翻訳してください。"
"返答は翻訳文のみとします。"
),
tools=[],
)
input_queue = asyncio.Queue()
frontend_task = asyncio.create_task(
receive_from_frontend(websocket, input_queue)
)
agent_task = asyncio.create_task(
agent.run(
inputs=[lambda: read_from_queue(input_queue)],
outputs=[lambda payload: send_to_frontend(websocket, payload)],
)
)
フロントエンド側での送受信
早い段階で返ってくる partial な結果を UI にはやく反映させるため、フロントエンド側での工夫も紹介します。
重複表示の防止
ストリーミングUIで一番苦労したのが「テキストの重複対応」です。
partial は文の大きな塊ごと再送されてくることがあるため、無邪気に追加すると同じ内容が何度も表示されてしまいます。
力技な部分もありますが、以下の2つの仕組みを取り入れました。
差分のみを追加する
直前のテキストと重なる部分を取り除き、純粋な差分だけを抽出して追加するロジック getAppendedText を実装しました。
// 簡易コード例
const getAppendedText = (previousText: string, nextText: string) => {
if (nextText.startsWith(previousText)) {
return nextText.slice(previousText.length);
}
// 後方一致で重複部分(overlap)を探し、差分だけを返す
const maxOverlap = Math.min(previousText.length, nextText.length);
for (let overlap = maxOverlap; overlap > 0; overlap -= 1) {
if (previousText.slice(-overlap) === nextText.slice(0, overlap)) {
return nextText.slice(overlap);
}
}
return nextText;
};
既出文のスキップ (Setによる管理)
差分計算だけでは対応しきれない、下記のような事例がありました。
- Bidiエージェントから「
is_final=falseで来た文が、その後is_final=trueのタイミングでまったく同じ全文として再送される」
これを防ぐため、文を「。 . ! ?」などで分割して正規化し、それを Set に入れて管理することで、既出の文をスキップ（除外）する仕組みを加えました。
Bidiエージェントは執筆時点ではPythonのみ対応
フロントエンドはNext.js, バックエンドはPython(FastAPI)で作りました。
フロントエンドでNext.jsのApp Router等を採用すると、
- クライアント端末にて実行されるクライアントコンポーネント
- ホスティングサーバー側で実行されるサーバーコンポーネント
を同一のアプリケーションディレクトリで管理できます。
フロントもバックエンドもTypeScriptでまとめたかったのですが、検証時点(2026年3月中旬)では,Typescript版のStrandsAgentsがBidirectional_streamingに非対応でした。バックエンドはPythonで構築せざるを得ませんでした。
Typescript版のStrandsAgentsも高頻度でアップデートが入ってる印象なので、Bidiエージェント対応にも期待ですね！
最後に
ストリーミングレスポンスのおかげで翻訳待ち時間を長く感じづらい(個人の感想)ので、音声翻訳タスクをAIエージェントに解決させるのは結構オススメできるかなと思いました!！
参考文献
StrandsAgents_Bidiエージェント関連のevent情報
Amazon Nova 2 Sonic対応言語情報
音声対話エージェントをStrands × AgentCore × Nova 2 Sonicで動かしてみる！