本記事は「Serverless Advent Calendar 2025」7 日目の記事です。
チャットアプリケーションのストリーミング表示の方法について
基盤モデルを使って Web ベースのチャットアプリケーションを作成するケースはよく見受けられますよね。 私も基盤モデルを使用するアプリケーションのデモやサンプルとして、よく作ります。
その際、「レスポンスのメッセージのストリーミングで表示したい」 というニーズも多いと思います。
世の中には色々な SDK がありますが、基盤モデルを直接呼び出す API でも、AI エージェントをコントロールする API でも、ストリーミングでレスポンスを得る API は、だいたい用意されています。
なので、それらを使えばいい、ということになるのですが、チャットアプリケーションの構造によって適用できる技術やフレームワーク、ランタイムが変わってきます。
例えば、Streamlit を使用し、そのコードの中でストリーミングのレスポンスを得る API を使用して実装し、コンテナにしてサーバーサイドで動作させるだけなら、至極シンプルで、何の問題もありません。
では、サーバーレスのコンピューティング環境として AWS Lambda を使いたい場合はどうでしょうか？
AWS Lambda を単体で使うなら、関数 URL を活用することもできます。
ただ、Amazon API Gateway と AWS Lambda を組み合わせた構成の場合はどうでしょうか？
そういった場合に役立つ Amazon API Gateway の Update が 2025年11月 に発表されました。
「Amazon API Gateway の REST API で AWS Lambda 関数と統合した場合のレスポンスストリームをサポート」
これは とてもよい Update だと思いますし、すでに多くの人に試され、その感触などが様々なブログ記事で公開されています。
ただ、AWS Lambda 関数と組み合わせる場合は、AWS Lambda の InvokeWithResponseStream API を使うことになり、ランタイムは Node.js が前提です。
もし、Node.js 以外のランタイムを使いたい場合は、Web Adapter の使用などを検討することになります。
では、AWS Lambda の関数 ランタイムとして Node.js 以外を使いたい、 Web Adapter を使いたくない、という場合はどうでしょうか？
その場合は、Amazon API Gateway の WebSocket API と AWS Lambda 関数を組み合わせる という方法もあります。
今回は、この方法を試してみたいと思います。
今回作成するアプリケーションの構成
- Amazon API Gateway で sendtext というハンドラを用意して、AWS Lambda 関数と統合します。
- AWS Lambda 関数は、受け取ったメッセージを Amazon Bedrock に対して converse_stream という API で送信します。
- そのレスポンスをストリームで受け取り、Amazon API Gateway へ返信します。
- フロントエンドでは、Amazon API Gateway の WebSocket API のエンドポイントの URL に対して WebSocket の接続を維持して、sendtext のハンドラが実行されるようにメッセージを送信し、レスポンスを受信します。
フロントエンドに Streamlit を使うことも検討したのですが、WebSocket の接続を維持するためのコードが複雑になったので、Next.js での実装に切り替えました。
AWS Lambda 関数
- sendtxt ハンドラとなる AWS Lambda 関数では、Amazon Bedrock の基盤モデルに対して converse_stream API でメッセージを送信します。
- 基盤モデルからのストリームレスポンスは、Amazon API Gateway に対して post_to_connection API でメッセージを返信します。
import boto3 import json import os API_ENDPOINT = os.environ["API_ENDPOINT"] STAGE = os.environ["STAGE"] MODEL_ID = os.environ["MODEL_ID"] # 推論パラメータの値 temperature = 0.5 # 推論パラメータの値の設定 inference_config = {"temperature": temperature} # システムプロンプト system_prompts = [{"text": "あなたは優秀なアシスタントです。問い合わせ内容に丁寧に応答して下さい。"}] def lambda_handler(event,context): # AWS SDK のクライアントオブジェクトの作成 brt = boto3.client(service_name='bedrock-runtime') apigw_management = boto3.client('apigatewaymanagementapi', endpoint_url=f"{API_ENDPOINT}/{STAGE}") # 接続 ID の取得 connectionId = event.get('requestContext', {}).get('connectionId') # メッセージの取得 body_content = json.loads(event.get('body', {})) # 基盤モデルへのリクエストメッセージの構成 text = body_content.get('text') message_1 = { "role": "user", "content": [{"text": f"{text}"}] } messages = [] messages.append(message_1) # Bedrock の基盤モデルの ID を指定 modelId = MODEL_ID # Bedrock の基盤モデルへリクエストを送信 try: response = brt.converse_stream(modelId=modelId,messages=messages,system=system_prompts,inferenceConfig=inference_config) except Exception as e: return { "statusCode": 500, "body": f"{e}" } # レスポンスを取得し、Amazon API Gateway へ返信 for event in response.get('stream'): if 'contentBlockDelta' in event: chunk = event['contentBlockDelta']['delta']['text'] try: apigw_management.post_to_connection(ConnectionId=connectionId, Data=json.dumps(chunk)) except Exception as e: return { "statusCode": 500, "body": f"{e}" } return { "statusCode": 200 }
この AWS Lambda 関数と Amazon API Gateway の WebSocket API と、その統合を作成する AWS SAM テンプレートはこちらの GitHub リポジトリ から参照できです。
フロントエンド
ストリーム表示に関わる部分だけフォーカスして説明します。
下記では、useEffect で WebSocket の接続を作成・維持を行っています。また、ws.onmessage でメッセージを細切れに受け取り、useState で管理している messages へ格納しています。
export default function Chat() { const [messages, setMessages] = useState<Message[]>([]); const [input, setInput] = useState(""); const [isConnected, setIsConnected] = useState(false); const wsRef = useRef<WebSocket | null>(null); const currentResponseRef = useRef<string>(""); const currentMessageIdRef = useRef<number | null>(null); useEffect(() => { const ws = new WebSocket(process.env.NEXT_PUBLIC_WebSocket_URL!); wsRef.current = ws; ws.onopen = () => setIsConnected(true); ws.onclose = () => setIsConnected(false); ws.onerror = () => setIsConnected(false); ws.onmessage = (event) => { const data = JSON.parse(event.data); const messageText = String(data); currentResponseRef.current += messageText; if (currentMessageIdRef.current) { setMessages(prev => prev.map(msg => msg.id === currentMessageIdRef.current ? { ...msg, text: currentResponseRef.current } : msg ) ); } }; return () => ws.close(); }, []);
下記は WebSocket でユーザーのメッセージを送信しています。
wsRef.current.send(JSON.stringify({ action: "sendtext", text: promptWithHistory }));
下記は useState で管理している messages を表示している部分です。
<div className="flex-1 overflow-y-auto p-4 space-y-4"> {messages.map((message) => ( <div key={message.id} className={`flex ${message.isUser ? "justify-start" : "justify-end"}`} > <div className={`max-w-md px-4 py-2 rounded-lg break-words ${ message.isUser ? "bg-yellow-200 text-black" : "bg-green-200 text-black" }`} > {message.text} </div> </div> ))} </div>
上記以外のフロントエンドのプロジェクトのリソースは こちらのGitHub のリポジトリ にまとめています。 フロントエンドのコードは、Amazon Q Developer の力を借りて作成したので、1時間ほどで作成できました。
完成イメージ
注意点
Amazon API Gateway の WebSocket API は、リクエスト数だけでなく接続時間にも課金されます。 これは REST API とは異なりますので注意しましょう。
最後に
基盤モデルを使ったチャットアプリケーションのようにレスポンスをストリーム表示するような場合で AWS のサーバーレスのサービスを使う場合は、いくつか方法はありますが、Amazon API Gateway の WebSocket API を活用することで、統合する AWS Lambda 関数のコードをシンプルにできるというメリットを感じました。
もちろん、2025年11月に Update された Amazon API Gateway の REST API のレスポンスストリーム対応も役立ちますが、WebSocket API の場合は、AWS Lambda 関数のランタイムを Node.js にする必要がなく、Web Adapter も不要であることはメリットといえると思います。
WebSocket API 料金は意識しておく必要はありますが、サーバーレスでの実装パターンの一つとして今後も活用していきたいと思います。
