# 生成AIの音声接客を『本番運用』するまで：Bedrock × Whisper × Polly × pgvector で無人キオスクを設計する > 店舗の対面接客を代替する生成AI音声エージェントを、PoCではなく本番運用まで持っていくための設計を実コードで解説。リアルタイム音声ループ、非同期・並列推論パイプライン、pgvectorによるRAG、ハルシネーションの構造的排除、AWS本番アーキテクチャまで。 - 公開日: 2026-06-24 - 著者: 友田 陽大 - タグ: AI, RAG, 音声AI, AWS Bedrock, Claude, pgvector, LangChain, Python, AWS, Terraform, アーキテクチャ設計 - URL: https://tomodahinata.com/blog/production-voice-ai-sales-agent-bedrock-pgvector ## 要点 - 音声接客の本番化はレイテンシ・正確性・安全性・改善・運用の5つの壁を設計で越えることが分水嶺になる - レイテンシは HTTP 即返し（submit-and-poll）と依存グラフに沿った並列ファンアウトで対面の沈黙を防ぐ - 型番・サイズは LLM でなく正規表現でマスタ照合し、生成は説明の言葉・マスタは数字の真実と分業させて誤答を排除する - 生成の前段に Claude を低温の分類器とした入力モデレーションゲートを置き、無関係・不適切発話を遮断する - RAG は専用 DB を足さず PostgreSQL + pgvector を選び、まず全件スキャンで正しく動かし HNSW への移行先を明示する --- 「生成AIで店舗の接客を自動化したい」という相談は増えました。デモを作るだけなら、いまや週末で動きます。`getUserMedia` でマイクを取り、Whisper で文字起こしし、LLM に投げ、TTS で読み上げる——これだけです。 しかし、**それを無人の店頭に置き、来店客が何を話すか分からない状態で、毎日壊れずに動かし続ける**となると、話はまったく別物になります。本記事は、専門商材（例: タイヤのような型番・サイズが命の商品）を扱う小売店舗向けに、生成AI音声接客キオスクを設計・実装し、本番運用まで持っていった経験から、「デモと本番の間にある距離」を実コードで埋めていく記録です。 対象読者は、生成AIプロダクトを「検証で終わらせず」事業に載せたいエンジニア・技術責任者です。スタックは Python(Flask) / React(TypeScript) / AWS(Bedrock, ECS Fargate, API Gateway, RDS+pgvector, Cognito, Lambda) / Terraform。関連する実績の詳細は文末のケーススタディに譲り、ここでは設計判断とトレードオフに集中します。 ## デモと本番を分ける5つの壁 音声接客を本番運用するとき、必ずぶつかる壁は次の5つでした。 | 壁 | デモでは無視できる | 本番では致命的 | | --- | --- | --- | | レイテンシ | 5秒待てる | 対面で5秒の沈黙は会話が成立しない | | 正確性 | それっぽければ良い | 型番・サイズの誤りは誤発注・クレーム直結 | | 安全性 | 開発者しか話さない | 来店客は何でも話す（無関係・不適切発話） | | 改善 | 手で直す | 「どの回答が外したか」を運用で回収する仕組みが要る | | 運用 | ローカルで動けば良い | 無停止デプロイ・再現可能なインフラ・監査ログ | 以降、この5つを1つずつ潰していきます。 ## 全体アーキテクチャ：2つのサーフェスを分離する 最初の設計判断は、**「来店客が触る面」と「運営者が触る面」を完全に分ける**ことでした。要求もセキュリティ境界もまるで違うからです。 - **店頭キオスク（来店客向け）**: 音声で対話する。認証は店舗端末に紐づく短いアクセスコード。徹底して「会話の速さ」に最適化する。 - **運用コンソール（運営者向け）**: 会話ログの分析、ナレッジ（RAG）の管理、端末管理。認証は AWS Cognito。徹底して「正確さと監査性」に最適化する。 リクエストフローは次の通りです。運用コンソールは API Gateway の Cognito オーソライザーで JWT を検証してから、VPC Link 経由で内部の ECS へ抜けます。 ```text [来店客] ── 音声 ──▶ CloudFront ──▶ ALB ──▶ ECS(Flask) ──▶ RDS(pgvector) / S3 │ ├─▶ Bedrock (Claude 3.5 Sonnet) ├─▶ OpenAI (Whisper / Embeddings) └─▶ Polly (TTS) [運営者] ── JWT ──▶ API Gateway ──(Cognito Authorizer)──▶ VPC Link ──▶ NLB ──▶ ALB ──▶ ECS ``` 「なぜ運用コンソールだけ API Gateway を挟むのか」は後述しますが、要点は **Cognito 認証をマネージドに寄せ、内部の ECS を直接インターネットへ晒さない**ためです。 ## 壁①レイテンシ：直列処理は必ず破綻する 音声接客のパイプラインは、本質的に重い直列処理です。 ```text 録音 → STT(文字起こし) → 入力チェック → ベクトル検索 → LLM生成 → TTS(音声合成) → 返却 ``` これを素直に直列で書くと、各ステップ数百ms〜数秒が積み上がり、平気で5秒を超えます。対面接客で5秒の沈黙は「フリーズした」と判断される時間です。ここを2段階で攻めました。 ### 段階1：HTTPを即座に返し、生成はバックグラウンドへ 音声をアップロードする `POST /api/conversation` は、**文字起こし結果と `taskId` だけを即座に返し**、重い生成処理はバックグラウンドタスクに逃がします。クライアントは `taskId` を使って結果をポーリングします。 この設計には実利が2つあります。1つは体感レイテンシ（ユーザーには「聞き取った内容」がすぐ返る）。もう1つは、**API Gateway の29秒タイムアウト制約を構造的に回避できる**ことです。LLM 生成が遅い日でも、HTTP レイヤーは詰まりません。 Flask では `flask-executor`（スレッドプール）でこれを実装しました。 ```python class ConversationResource(Resource): @jwt_required() def post(self): executor = current_app.config["EXECUTOR_CLIENT"] audio_file = self._decode_audio(request) # 文字起こしと認証情報の取得は互いに独立 → 並列に流す futures = { executor.submit( openai.audio.transcriptions.create, model="whisper-1", file=audio_file, language="ja", ): "transcription", executor.submit(self.get_access_code): "access_code", } results = {} for future in as_completed(futures): results[futures[future]] = future.result() transcript = results["transcription"].text task = BackgroundTaskResult(encoded_audio=None) db.session.add(task) db.session.commit() # 重い生成処理はバックグラウンドへ。HTTPはここで即返す。 executor.submit(self.background_task, transcript, results["access_code"], task.id) return {"transcript": transcript, "taskId": task.id}, 202 ``` ポーリング側は「まだ処理中（202）」と「完了（200 + 音声）」を区別します。 ```python class BackgroundTaskResultResource(Resource): def get(self, task_id): task = BackgroundTaskResult.query.get_or_404(task_id) if task.encoded_audio is None: return {"status": "in_progress"}, 202 return {"status": "completed", "audio": task.encoded_audio}, 200 ``` > 補足: スレッドプールを選んだのは、このパイプラインが I/O バウンド（外部API待ち）が支配的で、Gunicorn を `eventlet` ワーカーで動かしているためです。CPU バウンドな前処理が増えるなら、ここは SQS + 専用ワーカーへ寄せるのが次の一手になります。「まず動かし、ボトルネックが見えたら正しい場所へ移す」という順序を守るのが重要です。 ### 段階2：パイプライン内部を並列ファンアウトする バックグラウンドタスクの内部でも、依存関係のない処理は並列に流します。たとえば「セッション情報の取得」「入力モデレーション」「埋め込み生成」は互いに独立しているので、同時に走らせてクリティカルパスを縮めます。 ```python def background_task(self, transcript, access_code, task_id): executor = current_app.config["EXECUTOR_CLIENT"] # フェーズ1: 互いに独立 → ファンアウト futures = { executor.submit(self.get_session, access_code): "session", executor.submit(self.filter_question_content, transcript): "moderation", executor.submit(self.get_embedding, transcript): "embedding", } phase1 = {futures[f]: f.result() for f in as_completed(futures)} if phase1["moderation"] == "REJECT": return self._store_rejection(task_id) # 生成に進ませない # フェーズ2: 検索・履歴取得・QAチェーン初期化も独立 → ファンアウト futures = { executor.submit(self.retrieve_documents, phase1["embedding"]): "docs", executor.submit(self.get_past_conversations, phase1["session"].id): "history", executor.submit(load_qa_chain, self.claude_llm): "qa_chain", } phase2 = {futures[f]: f.result() for f in as_completed(futures)} answer = self._generate(phase2["qa_chain"], phase2["docs"], phase2["history"], transcript) encoded = self.encode_audio(answer) # Polly → base64 self._persist(task_id, encoded, answer) ``` ポイントは「全部を非同期にする」ことではなく、**依存グラフを描いて、独立な辺だけを同時に流す**ことです。やみくもな並列化はバグの温床になります。 ## 壁②正確性：生成と規則のハイブリッドで誤答を排除する 専門商材で最も怖いのは「流暢な誤答」です。LLM は自信たっぷりに間違った型番を言います。タイヤサイズ `225/60R15` を `225/65R15` と言い間違えれば、誤発注に直結します。 ここでの設計原則は明快です。**「曖昧さが許される箇所は生成に、誤りが致命的な箇所は規則に倒す」**。 商品番号・サイズの抽出は LLM に任せず、発話テキストから正規表現で決定的に取り出し、マスタテーブル（`normalized_data`）に照合します。 ```python TIRE_SIZE = re.compile(r"(\d{3})\s*[-/／]?\s*(\d{2})\s*[-/Rr]?\s*(\d{2})") def resolve_product_number(self, transcript: str) -> str | None: m = TIRE_SIZE.search(transcript) if not m: return None w, aspect, rim = m.groups() row = NormalizedData.query.filter_by( digit_1=w, digit_2=aspect, digit_3=rim ).first() return row.product_number if row else None ``` 照合できれば、その確定値を回答の制約として LLM に渡します。**生成AIは「説明の言葉」を作る役、マスタは「数字の真実」を握る役**に分業させるわけです。これだけでクレーム級の誤答はほぼ消えます。 LLM 側の生成も、無制約には使いません。LangChain の QA チェーンに、検索した上位ドキュメントと直近の会話履歴を渡しつつ、**回答長を制約**します（音声で読み上げるため、長すぎる回答は接客として失格です）。 ```python prompt = ( "あなたは店舗の接客スタッフです。以下の参考情報と会話履歴だけを根拠に、" "200文字以内で、確認できない事項は断定せず案内してください。\n" f"# 確定した商品番号: {product_number or '未確定'}\n" f"# 直近の会話:\n{recent_history}\n" f"# 質問: {transcript}" ) answer = qa_chain.run({"input_documents": documents, "question": prompt}) ``` ## 壁③安全性：生成の前に発話をゲートする 無人キオスクは「誰が何を話すか」を制御できません。無関係な雑談、不適切発話、攻撃的なプロンプトが飛んできます。そこで**生成に進む前に、入力を分類して遮断するゲート**を置きました。Claude 3.5 Sonnet を低温（`temperature=0.1`）の分類器として使い、`ACCEPT` / `REJECT` だけを返させます。 ```python def filter_question_content(self, transcript: str) -> str: body = json.dumps({ "anthropic_version": "bedrock-2023-05-31", "max_tokens": 5, "temperature": 0.1, "messages": [{ "role": "user", "content": ( "次の発話が店舗接客として適切なら ACCEPT、" "成人向け・政治・誹謗中傷・商材と無関係なら REJECT のみを返答:\n" f"{transcript}" ), }], }) res = self.bedrock.invoke_model( modelId="anthropic.claude-3-5-sonnet-20240620-v1:0", body=body, ) verdict = json.loads(res["body"].read())["content"][0]["text"].strip() return "REJECT" if "REJECT" in verdict else "ACCEPT" ``` 同じ Bedrock の Claude を「回答生成」「入力モデレーション」「画像・動画を見せるべきかの判定」の3用途で共通利用しているのもコスト・運用上の工夫です。プロバイダを増やさず、モデル更新も1箇所で済みます。 ## RAG基盤：なぜ pgvector か、そして全件スキャンの限界 ナレッジ検索（RAG）の土台には、専用ベクトルDBを足さず **PostgreSQL + pgvector** を選びました。理由は3つです。 1. **運用の単純さ**: 業務データと埋め込みが同じ RDB に乗る。バックアップ・権限・トランザクションが一元化される。 2. **コスト**: Pinecone 等の月額・別系統の監視を増やさない。 3. **整合性**: 会話・ドキュメント・検索結果（類似度スコア）を同じトランザクション境界で扱える。 埋め込みは OpenAI `text-embedding-3-large` を**1024次元に切り詰めて**使いました。検索精度を確保しつつ、保存・計算コストを抑えるバランス点です。 ここで正直に書いておきます。初期実装の検索は、**ドキュメントを全件読み出して Python 側で内積を計算する素朴な方式**でした。 ```python def retrieve_documents(self, query_embedding): cur = self.conn.cursor() cur.execute("SELECT id, content, vector FROM documents") scored = [ (np.dot(query_embedding, vec), content, doc_id) for doc_id, content, vec in cur.fetchall() ] scored.sort(key=lambda x: x[0], reverse=True) top = scored[:10] return [LangchainDocument(page_content=c) for _, c, _ in top], top ``` 店舗あたりのカタログ規模では、これで十分実用的に動きます。**早すぎる最適化を避け、まず正しく動かす**という判断です。ただしこれは O(n) のフルスキャンであり、ドキュメントが数万を超えると破綻します。スケール時の正しい移行先は、pgvector のインデックス（HNSW / IVFFlat）と距離演算子に寄せることです。 ```sql -- スケール時の移行先: DB側でANN検索を完結させる CREATE INDEX ON documents USING hnsw (vector vector_cosine_ops); SELECT id, content FROM documents ORDER BY vector <=> %(query)s -- コサイン距離。インデックスが効く LIMIT 10; ``` 「いまの規模で正しく動かし、限界とその先の道筋を明示しておく」——これがハッタリのない RAG 運用だと考えています。検索結果は類似度スコアごと `vector_search_results` に保存し、**後から『なぜその回答になったか』を追跡できる**ようにしています。これは次の「改善ループ」で効いてきます。 ## 壁④改善：運用しながら精度を上げるループ PoC と本番の最大の違いは、**「外した回答を回収して直す仕組みがあるか」**です。データモデルは会話の全文脈を保持するよう設計しました（要約）。 ```text User ─< Terminal ─< Session ─< Conversation ─< Chat │ ├─ vector_search_results (検索根拠 + 類似度) └─ documents (pgvector埋め込み) ─< attachments(画像/動画) normalized_data (型番マスタ) background_task_results (生成音声の非同期ハンドル) ``` `Conversation` には `failure_reason`（音声認識失敗・モデレーション拒否などのコード）、`process_time`（レイテンシ実測）、さらに**運営者が後付けする教師データ**として `failure_reason_should_be` と `system_response_message_should_be` を持たせています。 運用コンソールで会話を見たオペレーターが「この回答は本来こうあるべきだった」を入力すると、それを**埋め込み直してナレッジに再投入**できます。つまり、運用そのものが学習データを生むループになっています。`process_time` を貯めれば、レイテンシのリグレッションも検知できます。 ## 壁⑤運用：二層認証と、再現可能なインフラ ### 認証は用途別に分ける 来店客と運営者を同じ認証で扱うのは筋が悪い。そこで二層に分けました。 - **キオスク**: 店舗端末に紐づく6桁アクセスコードを検証し、`HttpOnly` Cookie の JWT を発行（アクセス24h / リフレッシュ7d、CSRF トークン併用）。来店客はログイン操作をしません。 - **運用コンソール**: AWS Cognito（SRP 認証）。API Gateway の Cognito オーソライザーで JWT を検証してから内部へ通します。 ```python # キオスク: アクセスコード → JWT(Cookie) class AccessCodeResource(Resource): def post(self): code = AccessCodeSchema().load(request.get_json())["accessCode"] terminal = Terminal.query.filter_by(access_code=code, is_active=True).first() if terminal is None: return {"message": "invalid code"}, 401 access = create_access_token(identity={"token": code}, expires_delta=timedelta(hours=24)) resp = make_response({"ok": True}) set_access_cookies(resp, access, max_age=24 * 60 * 60) # HttpOnly + CSRF return resp ``` 入力は Marshmallow スキーマで境界検証し、DB アクセスは SQLAlchemy の ORM 経由（パラメータ化クエリ）。S3 上のカタログ画像・動画は署名付き URL でのみ配布し、バケットを公開しません。 ### インフラは Terraform で全面コード化 `staging` と `production` を**同一構成**で再現できるよう、VPC からアプリまでを Terraform で管理しました。バックエンドは ECS Fargate（軽量タスク、ローリングデプロイで無停止）。 ```hcl resource "aws_ecs_service" "backend" { name = "${var.env}-backend" cluster = aws_ecs_cluster.backend.id task_definition = aws_ecs_task_definition.backend.arn launch_type = "FARGATE" desired_count = 1 deployment_configuration { maximum_percent = 200 # 新タスクを立ててから minimum_healthy_percent = 100 # 旧タスクを落とす = 無停止 } } ``` RAG のドキュメント取り込みは、重い処理（埋め込み生成）なので VPC アタッチした Lambda に分離しました。RDS へ直接書き込みつつ、OpenAI へは NAT 経由で出ます。 ```hcl resource "aws_lambda_function" "insert_rag_document" { function_name = "${var.env}-insert-rag-document" runtime = "python3.11" timeout = 300 # 大きめのPDFも完走させる vpc_config { # RDS(pgvector)へ到達するためVPC内に置く subnet_ids = [var.lambda_subnet_id] security_group_ids = [var.lambda_sg_id] } } ``` なぜ運用コンソールだけ「API Gateway → VPC Link → NLB → ALB → ECS」と層が多いのか。それは **Cognito 認証を API Gateway のオーソライザーにオフロード**しつつ、内部の ECS をパブリックに晒さないためです。VPC Link は接続先に NLB を要求するため、ALB の前段に NLB を置く構成になっています。 ### デプロイは GitHub Actions で自動化 ```yaml # バックエンド: ビルド → ECRへpush → ECSローリングデプロイ - run: docker build -t $ECR_REPO:latest -f backend/Dockerfile.prod ./backend - run: docker push $ECR_REPO:latest - run: | aws ecs update-service --cluster $CLUSTER --service $SERVICE \ --force-new-deployment ``` フロントは S3 同期 + CloudFront 無効化。SPA のルーティングは CloudFront の「404 → `/index.html`（200）」で吸収しています。 ## 可観測性・回復性・冪等性をどう担保したか 最後に、本番運用で効く「地味だが重要な」設計を整理します。 - **可観測性**: 会話ごとに `process_time` と `failure_reason` を記録。ECS / RDS は CloudWatch Logs。「遅い・外した」を会話単位で後追いできる。 - **回復性**: 生成失敗・モデレーション拒否を例外で落とさず、`failure_reason` 付きで必ず1レコードに収束させる。来店客には常に何らかの音声を返す（無言で固まらせない）。 - **冪等性**: 非同期結果は `taskId` をキーに `background_task_results` へ書き込むため、ポーリングの再送やネットワーク再試行で二重生成が起きない。 - **テスト容易性**: 商品番号の抽出・正規化のような決定的ロジックは純粋関数に切り出し、外部API・DBなしで単体テストできる。「曖昧なLLM部分」と「厳密なロジック部分」を分けたことが、テスト戦略でも効いてくる。 ## まとめ：デモは才能、本番は設計 音声接客のデモは、もはや誰でも作れます。差がつくのは**「本番の5つの壁」をどう設計で越えるか**です。 1. **レイテンシ**は、HTTP即返し（submit-and-poll）＋依存グラフに沿った並列ファンアウトで越える。 2. **正確性**は、生成（流暢さ）と規則（型番マスタ照合）のハイブリッドで越える。 3. **安全性**は、生成の前段に入力モデレーションのゲートを置いて越える。 4. **改善**は、検索根拠と教師データを残し、運用が学習データを生むループで越える。 5. **運用**は、用途別の二層認証と、Terraform による再現可能インフラ・無停止デプロイで越える。 生成AIプロダクトの価値は「賢いモデルを呼ぶこと」ではなく、**モデルの曖昧さを業務の厳密さと噛み合わせる設計**にあります。そこを丁寧にやれるかどうかが、PoC を本番に変える分水嶺だと考えています。 --- 実際にこの音声接客キオスクをどう作り、どんな技術判断をしたかは、ケーススタディで詳しく紹介しています。生成AIを「検証」で終わらせず事業に載せたい方は、ぜひお気軽にご相談ください。