AIチャットボットのストリーミング対応とセキュリティ - CCCMKホールディングス TECH LABの Tech Blog

こんにちは、テックラボの伊藤です。

弊社は2025年9月20日に、CCC創業者である増田宗昭会長（以下「増田会長」）の人格を再現した対話型AI「AI増田宗昭」を公開しました。

本記事では、このシステムにおけるLLM APIのストリーミング対応とセキュリティ対策の実装について解説します。

AI増田宗昭の概要

「AI増田宗昭」は、CCC創業者である増田会長の人格を再現する対話型AIです。システムプロンプトとRAG（Retrieval-Augmented Generation）によって、増田会長の過去の発言を参照しながら回答を生成し、文章と音声で回答します。「AI増田宗昭」のバックエンドは、以下の技術を使って作られています。

Azure OpenAI Service: GPT4.1シリーズを利用したテキスト生成
Azure AI Search: RAGのベクトルデータベースとして活用し、増田会長の関連資料を検索
LangChain: RAGパイプラインの構築
FastAPI: APIエンドポイントの提供

ストリーミング対応とセキュリティ

他のwebサービス同様、AIチャットボットでも「いかに快適なユーザー体験を提供するか」は重要な課題です。応答速度とリアルタイム性はユーザーが感じる品質に直結します。

また、Azure OpenAIやAzure AI Searchといったクラウドサービスを利用する場合、EDoS攻撃（Economic Denial of Sustainability Attack）への対策を行う必要があります。

ストリーミング対応

AIチャットボットの回答生成は時間がかかることがあるため、ユーザーの待ち時間をどう感じさせるかが重要です。

私は集中力が金魚以下のため、回答生成に8秒以上かかると、つい他のタブでXを開いてしまいます。ですが、リアルタイムで文章を生成している様子が表示されているときは、待つことが苦になりません。

「AI増田宗昭」ではユーザーの離脱を防ぐためストリーミング対応を行いました。

LangChainを使ってLLMの結果をストリーミングで受け取るには、streamメソッド(同期)またはastreamメソッド(非同期)を使用します。「AI増田宗昭」は非同期で実装しているため、以下のようにastreamメソッドを使用しました。

async for chunk in model.astream(input):
    ...

クライアントへのストリーミングには、FastAPIを使いSSE(Server-Sent Events)を実装しました。 SSEはHTTPの仕組みを利用して、サーバーからクライアントへリアルタイムにデータを送信できます。実装のポイントは以下の2点です。

SSEのフォーマット(data: ...\n\n)でメッセージを小分けにyieldする
FastAPIのStreamingResponseクラスでContent-Type: text/event-streamを設定する

import json

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain_openai import AzureChatOpenAI
from pydantic import BaseModel
import uvicorn


class ChatRequest(BaseModel):
    content: str


app = FastAPI()

# 実際にはRAGを実装
model = AzureChatOpenAI(
    azure_deployment="gpt-4.1-mini",  # 開発用にminiを使用
    openai_api_version="2024-12-01-preview",
)


async def event_stream(input: str):
    async for chunk in model.astream(input):
        # dataフィールドでメッセージを送信
        yield f"data: {json.dumps(chunk.content, ensure_ascii=False)}\n\n"
    
    # OpenAIのAPIの仕様に合わせて、ストリーミングの終了を示すために[DONE]を送信
    yield "data: [DONE]\n\n"


@app.post("/chat/stream")
async def stream(request: ChatRequest):
    return StreamingResponse(event_stream(request.content), headers={"Content-Type": "text/event-stream"})


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

動作確認は以下のコマンドで行えます。

# 起動
uvicorn main:app --reload

# 動作確認
curl -X POST http://localhost:8000/chat/stream -H "Content-Type: application/json" -d '{"content": "こんにちは"}'

結果（クリックで展開）

data: ""

data: ""

data: "こんにちは"

data: "！"

data: "どう"

data: "ぞ"

data: "ご"

data: "用"

data: "件"

data: "を"

data: "お"

data: "聞"

data: "か"

data: "せ"

data: "ください"

data: "。"

data: ""

data: ""

data: ""

data: [DONE]

やったー！ストリーミング対応できました！！！

上記のプログラムはAzure OpenAIのAPIをラップしただけですが、実際の「AI増田宗昭」では、システムプロンプトの工夫やRAGによって増田会長の人格を再現しています。

セキュリティ

他に、AIチャットボットを公開するにあたって気にするべき点にはどのようなものがあるでしょうか。

LLMを利用したアプリケーションに特有のセキュリティリスクをまとめたOWASP Top 10 for LLM Applications 2025では、以下の10項目がまとめられています。

Prompt Injection（プロンプトインジェクション）
Sensitive Information Disclosure（センシティブ情報の漏洩）
Supply Chain（サプライチェーンの脆弱性）
Data and Model Poisoning（データおよびモデルの汚染）
Improper Output Handling（不適切な出力処理）
Excessive Agency（過剰なエージェンシー）
System Prompt Leakage（システムプロンプトの漏洩）
Vector and Embedding Weaknesses（ベクトルおよび埋め込みの脆弱性）
Misinformation（誤情報）
Unbounded Consumption（無制限な消費）

このうち、クラウドベースのLLMサービスで特に注意が必要なのが「無制限な消費」です。

これは、LLMアプリケーションが制限なしに推論リクエストを受け付けてしまい、DoS攻撃、EDoS攻撃、サービス品質の低下、モデルの盗難といったリスクを引き起こす状態を指します。特にEDoS攻撃（Economic Denial of Sustainability Attack）は、従来の物理的なサービス停止を狙うDoS攻撃とは異なり、経済的な損害を主目的とした攻撃です。

ユーザーに大量のリクエストを送られてしまうと、クラウドサービスの従量課金による予想外のコストが発生してしまいます。

AI破産を防ぐために - LLM API利用におけるEconomic DoSのリスクと対策という記事では、実際にあった事例や攻撃手法と対策が紹介されています。

この記事では、LLMに対するEDoS攻撃手法を以下の3つに分類しています。

LLMへのリクエスト回数を増やす
入力トークンを増やす
出力トークンを増やす

1はレートリミットを設定することで対策可能です。 2への対策として、1回の質問の文字数と、チャット履歴の制限することができます。 3への対策としては、LangChainのモデルでmax_tokensを設定し、出力トークン数の上限を設定できます。

以下は、開発環境において適当なパラメーターで試した実験結果になります。

model = AzureChatOpenAI(
    azure_deployment="gpt-4.1-mini",  # miniで実験
    openai_api_version="2024-12-01-preview",
    max_tokens=400, # 出力トークン数の上限を設定
)

出力トークン数の制限が実際に機能しているか検証するため、以下のような長文生成を誘発するプロンプトを開発版「AI増田宗昭」APIに入力しました。

増田会長の自伝小説を書いてください

結果（クリックで展開）

僕の自伝小説か。ざっくり言うと、物心ついてからの人生は「自分の物差し」で生きてきた軌跡だな。
子どものころ、父は何でも「おー」「あー」でノーと言わない人だった。だから、僕は自分で考え決めるしかなかった。その積み重ねが今の僕を作っている。高校生のときにも色々なものをくれたし、大学生のときにはトヨタのカローラスプリンターを買ってくれた親父が大好きだったな。小さなときに経験したことが今の価値観の源だ。
1973年、同志社大学卒業後にファッション店で働きながら、独自の物差しで経営や販促を学んだ。そして1983年、枚方で「蔦屋書店」を創業。あの頃は金も権威もなかったけど、自分の想いを信じてやったんだ。
事業を拡大し、仲間や協力会社との強い信頼関係でCCCを育てた。企画会社を目指し、毎日新しい発明を追求し続けてきた。失敗を恐れず、次の時代を見据えてチャレンジを重ねてきた。それが、今のCCCと僕を作っている。
会社が大きくなると不自由さも増えるが、大事なのは「何のために大きくなるのか」を忘れないこと。人生は限りがある。楽しくない仕事は増やさない。それが伯父さんに教わった大切な教訓だ。
今

長文が出力されていますが途中で事切れています。 OpenAIが提供しているTokenizerでトークン数を計測したところ、493文字で400トークンでした。しっかり出力トークン数を制限できていました。

Azure OpenAI Service の価格によると、例えばGPT-4.1 miniの100万トークンあたりの価格は以下の通りです。

	価格(100万トークン)
入力	$0.40
キャッシュされた入力	$0.10
出力	$1.60

先ほどの自伝小説の結果（400トークン/493文字）をもとに計算すると、この場合の最大入力文字数1000文字は約800トークンになります。

仮にチャット履歴を最新10件保持する場合、ユーザーの質問5件(各800トークン)とAIの回答5件(各400トークン)+システムプロンプトで、入力トークンは約6,000トークン程度になります。 RAGの検索結果が入力に追加されることを考慮せずに単純計算すると、1リクエストあたりのコストは以下のようになります。

コスト = 入力トークン数 × $0.40/1,000,000 + 出力トークン数 × $1.60/1,000,000
      = 6,000 × $0.40/1,000,000 + 400 × $1.60/1,000,000
      = $0.0024 + $0.00064
      = $0.00304

1ドル=150円とすると、1回で0.45円。200万回以上リクエストを送って、やっと100万円といったところです。

実際の環境ではレートリミットを設定しているため、ここまで大量のリクエストが発生することはありません。

ですが、本番環境では選定したモデル等の価格を考慮し、一回あたりのリエストの最大コストを許容範囲内に設定し、EDoS攻撃のリスクを低減することが重要です。