「生成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 へ抜けます。
[来店客] ── 音声 ──▶ 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 を直接インターネットへ晒さないためです。
壁①レイテンシ：直列処理は必ず破綻する
音声接客のパイプラインは、本質的に重い直列処理です。
録音 → 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（スレッドプール）でこれを実装しました。
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 + 音声）」を区別します。
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：パイプライン内部を並列ファンアウトする
バックグラウンドタスクの内部でも、依存関係のない処理は並列に流します。たとえば「セッション情報の取得」「入力モデレーション」「埋め込み生成」は互いに独立しているので、同時に走らせてクリティカルパスを縮めます。
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）に照合します。
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 チェーンに、検索した上位ドキュメントと直近の会話履歴を渡しつつ、回答長を制約します（音声で読み上げるため、長すぎる回答は接客として失格です）。
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 だけを返させます。
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つです。
- 運用の単純さ: 業務データと埋め込みが同じ RDB に乗る。バックアップ・権限・トランザクションが一元化される。
- コスト: Pinecone 等の月額・別系統の監視を増やさない。
- 整合性: 会話・ドキュメント・検索結果（類似度スコア）を同じトランザクション境界で扱える。
埋め込みは OpenAI text-embedding-3-large を1024次元に切り詰めて使いました。検索精度を確保しつつ、保存・計算コストを抑えるバランス点です。
ここで正直に書いておきます。初期実装の検索は、ドキュメントを全件読み出して 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）と距離演算子に寄せることです。
-- スケール時の移行先: 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 と本番の最大の違いは、**「外した回答を回収して直す仕組みがあるか」**です。データモデルは会話の全文脈を保持するよう設計しました（要約）。
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桁アクセスコードを検証し、
HttpOnlyCookie の JWT を発行（アクセス24h / リフレッシュ7d、CSRF トークン併用）。来店客はログイン操作をしません。 - 運用コンソール: AWS Cognito（SRP 認証）。API Gateway の Cognito オーソライザーで JWT を検証してから内部へ通します。
# キオスク: アクセスコード → 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（軽量タスク、ローリングデプロイで無停止）。
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 経由で出ます。
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 で自動化
# バックエンド: ビルド → 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つの壁」をどう設計で越えるか**です。
- レイテンシは、HTTP即返し（submit-and-poll）＋依存グラフに沿った並列ファンアウトで越える。
- 正確性は、生成（流暢さ）と規則（型番マスタ照合）のハイブリッドで越える。
- 安全性は、生成の前段に入力モデレーションのゲートを置いて越える。
- 改善は、検索根拠と教師データを残し、運用が学習データを生むループで越える。
- 運用は、用途別の二層認証と、Terraform による再現可能インフラ・無停止デプロイで越える。
生成AIプロダクトの価値は「賢いモデルを呼ぶこと」ではなく、モデルの曖昧さを業務の厳密さと噛み合わせる設計にあります。そこを丁寧にやれるかどうかが、PoC を本番に変える分水嶺だと考えています。
実際にこの音声接客キオスクをどう作り、どんな技術判断をしたかは、ケーススタディで詳しく紹介しています。生成AIを「検証」で終わらせず事業に載せたい方は、ぜひお気軽にご相談ください。