📘
「React × FastAPI × OpenAI × AWS Kendra」で作る本格RAGチャットボットを作ってみた

okawa_
2025/05/30に公開
 はじめにこんにちは！おおかわです！
皆様は「RAG（Retrieval-Augmented Generation）」というものをご存じでしょうか？

最近、少しずつ耳にするようになってきた技術ですが、簡単に言うと『社内で扱っているドキュメント等をAIが検索し、AIが質問に回答してくれる仕組み』です。

普通の生成AIは、基本的には学習済みの知識だけで回答を作成するのですが、RAGは異なります。

「社内データ」や「インプットしたい知識を記載したPDF」なんかを検索して、その結果を参考にしながらAIが回答してくれるんです！

今回の記事では、このRAGの仕組みを実際に使って、「React + FastAPI + OpenAI + AWS Kendra + DynamoDB」という構成で、UI付きのRAGチャットアプリケーションをゼロから構築していきます！

 課題と背景大学生や専門学生の中には、「レジュメが多すぎて、管理するのがめんどくさい…」「知らない間にドライブのどっかに消え去っている（探す気は皆無）」そんな悩みあったりしませんか？
私自身、授業毎の膨大な量のPDFを管理するのがめんどくさすぎて、結局「もういい！」ってなって、課題直前に友達に送ってもらう、みたいなこともよくありました。
このアプリは、「探す」という行動そのものをなくすことを目指しています。PDF資料やFAQ、先生が配ってくれたレジュメなどを、全部まとめて管理して、「質問すれば答えが返ってくる」という状態をつくります。
つまり…「これどこにあったっけ…？」と悩む時間を減らして、「聞けば出てくる」という、AIとの対話型学習。これを実現するために使っている技術が、今回のRAGチャットアプリケーションです！

 今回の構成フロントエンド（React + Vite)
チャットUIを提供
ユーザーの質問をサーバーへ送信
バックエンド（FastAPI)
クラアントから質問を受け取る
会話履歴をDynamoDBで管理
AWS Kendra & OpenAI API に処理を渡す
生成された回答をフロントエンドに返却
AWS Kendra[1]（検索エンジン）
質問に関連するPDFなどの情報をインデックスから検索
AWS S3[2]（ストレージ）
レジュメやPDFなどのドキュメントをアップロードして管理
Kendraインデックスが参照する知識ベースとして機能
OpenAI API
Kendraから帰ってきた結果をOpenAIが文脈補強
JSONモードで整形された返答を生成[3]
DynamoDB（会話履歴管理）
セッションごとの履歴を保存し、会話を文脈として保持
履歴ベースで自然なやり取りをサポート（ユーザーの『これ』『この』などの入力にも柔軟に対応）

 バックエンド実装（FastAPI）今回もバックエンドにはFastAPIを採用しています。

FastAPIは軽量化かつ、非同期対応でありながら、ドキュメント自動生成や型ヒントサポートが充実しているため、AIチャットのようなAPIサーバーとの相性がいい便利なWebフレームワークです。
ディレクトリ構成
├── app/
│   ├── main.py                # FastAPI起動エントリーポイント
│   ├── routes/ask_route.py    # /askエンドポイントのルーティング
│   ├── services/openai_service.py
│   ├── services/kendra_service.py
│   ├── services/dynamodb_service.py
main.py
from fastapi import APIRouter
from pydantic import BaseModel

# インポート
from app.services.kendra_service import search_kendra
from app.services.openai_service import get_openai_response
from app.services.dynamodb_service import save_message

router = APIRouter()

class AskRequest(BaseModel):
    question: str
    session_id: str

@router.post("/ask")
async def ask(request: AskRequest):
    question = request.question
    session_id = request.session_id
    
    # Kendraで関連情報検索
    kendra_results = await search_kendra(question)
    
    # OpenAIによる回答生成
    openai_answer = await get_openai_response(kendra_results, session_id)

    # DynamoDB に履歴保存
async def save_message(session_id: str, question: str, answer: str):
    dynamodb_service.save_message(session_id, "user", question)
    dynamodb_service.save_message(session_id, "assistant", answer)
    return {"answer": openai_answer}
エンドポイント/askにPOSTリクエストが届くと、質問とセッション ID を受け取り、そこから一連の処理が始まります。
まず、入力された質問に対してsearch_kendra(question)を使い、AWS Kendra で関連情報を検索します。これは、OpenAIが的確な回答を生成するための材料を集めるフェーズです。
次に、その検索結果をもとに get_openai_response(kendra_results, session_id)によって、AIの回答文を生成します。
最後に、そのやり取り全体をsave_message(session_id, question, openai_answer)によって、DynamoDB に保存します。
ユーザーごとのセッション ID を使って、質問・回答・関連情報を記録することで、後から履歴を参照したり、文脈を維持したりできるようになります。
openai_service.py
import json
from typing import List, Dict
import logging

logger = logging.getLogger(__name__)

# 非同期クライアントの作成
from openai import AsyncOpenAI

openai_client = AsyncOpenAI()  # 環境変数からAPIキーを取得

async def generate_answer_with_openai(
    question: str,
    context: str,
    history: List[Dict[str, str]],
    json_mode: bool = True
) -> str:
    try:
        logger.info("OpenAI API リクエストを送信します")

        # OpenAIに送るメッセージ構築
        messages = [
            {
                "role": "system",
                "content": f"""あなたはRAGシステムのアシスタントです。
以下の参考情報を元に、ユーザーの質問に答えてください。
回答は必ず以下のJSON形式で返してください:
{{"answer": "ここに回答内容"}}

参考情報:
{context}

重要: 回答は必ずJSON形式で、他の形式では返さないでください。"""
            },
            *history,
            {"role": "user", "content": question}
        ]

        response = await openai_client.chat.completions.create(
            model="gpt-4.1",
            messages=messages,
            temperature=0,
            response_format={"type": "json_object"} if json_mode else None,
        )

        answer = response.choices[0].message.content
        logger.info(f"OpenAI APIからの回答: {answer}")

        # JSON形式でパース
        parsed = json.loads(answer)
        return parsed.get("answer", "レスポンスに 'answer' キーが存在しません")

    except json.JSONDecodeError as e:
        logger.error(f"JSON解析エラー: {e}")
        return f"JSON形式ではない返答でした: {answer}"
    except Exception as e:
        logger.error(f"OpenAI APIエラー: {e}")
        return f"OpenAI APIでエラーが発生しました: {str(e)}"
        
async def get_openai_response(kendra_context: str, session_id: str) -> str:
    # セッションに基づいて過去の履歴を取得
    history = await load_chat_history(session_id)

    # 過去の履歴を OpenAI に渡す
    return await generate_answer_with_openai(
        question=history[-1]["content"],  # 最新のユーザーの質問
        context=kendra_context,
        history=history
    )
特徴は、回答を JSON 形式で必ず返すよう指示していること。これにより、後続の処理で扱いやすくなります。

また、json_mode を有効にすることで、OpenAI のレスポンスを構造化された JSON 形式に限定でき、安定した結果が得られます。

さらに、例外処理も組み込んでいるため、万が一レスポンスが不正でもログで検知でき、デバッグしやすい構成になっています。

 「temperature」とは？OpenAI API を使う際に登場する「temperature」というパラメータは、生成される文章の “創造性” や “多様性” をどれだけ許容するかを調整する役割を担っています。

数値は 0〜1 の間で設定でき、数値が高いほど出力はランダムに、低いほど決定的になります。
たとえば、temperature を 1 に近づけると、より自由な表現や言い回しが選ばれやすくなり、クリエイティブな文章生成に向いています。

逆に 0 に近づけると、常に最も確率の高い語句が選ばれるため、安定した出力が得られます。
今回は RAG チャットアプリケーションを作成するにあたって、創造性よりも事実に基づいた正確な回答が求められるため、temperature は 「0」 に設定しています。

temperature=0:

決定的・予測可能: 最も確率の高い単語を選び、毎回ほぼ同じ結果になります。

事実ベース・保守的: 正確性や指示への忠実さが求められる場合に適しています。創造性は低めです。


temperature=1:

ランダム・多様: より多くの単語が選択肢に入り、毎回異なる多様な結果になります。

創造的・革新的: 新しいアイデアや表現が欲しい場合に適しています。一貫性や正確性が低下する可能性もあります。

kendra_service.py
import boto3
import os
import logging
from dotenv import load_dotenv

load_dotenv()
logger = logging.getLogger(__name__)

# Kendraクライアントの初期化
kendra_client = boto3.client(
    "kendra",
    region_name=os.getenv("AWS_REGION"),
    aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
    aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
)

def search_kendra(question: str) -> str:
    try:
        response = kendra_client.query(
            IndexId=os.getenv("KENDRA_INDEX_ID"),
            QueryText=question,
            PageSize=10,
            AttributeFilter={
                "EqualsTo": {"Key": "_language_code", "Value": {"StringValue": "ja"}}
            },
        )
        logger.info(f"Kendra検索結果: {response}")

        if response.get("ResultItems"):
            top = response["ResultItems"][0]
            if top.get("DocumentExcerpt") and top["DocumentExcerpt"].get("Text"):
                return top["DocumentExcerpt"]["Text"]

        return "関連情報が見つかりませんでした"

    except Exception as e:
        logger.error(f"Kendra検索エラー: {e}")
        return "Kendra検索でエラーが発生しました"
返ってきた検索結果の中から、最も関連度の高い1件目の抜粋テキスト（DocumentExcerpt）を取り出して返します。もし結果が空だった場合やエラーが起きた場合は、ログに記録しつつ適切なメッセージを返すよう設計されています。

この構成により、ユーザーの質問に対して適切な文脈を持つ情報を Kendra から取り出し、OpenAI に渡す前段の処理として有効に活用できるようにしました。

 会話履歴の保存とセッション管理（DynamoDB）このチャットアプリでは、ユーザーが過去にどんな質問をしたのかをセッション単位で保持したいと考えました。

DynamoDB は NoSQL なので高速アクセスが可能。かつフルマネージドでスケーラブルなため、DB を使うのが初心者の方でも容易に使用することができます！

以下のテーブル構成と、テーブルにデータを保存するサンプルコードを記載しております！

 データの保存処理のコード例import boto3
import uuid
import os
import logging
from datetime import datetime
from typing import List, Dict, Optional
from botocore.exceptions import ClientError
from boto3.dynamodb.conditions import Key

logger = logging.getLogger(__name__)

class DynamoDBService:
    def __init__(self):
        self.dynamodb = boto3.resource("dynamodb")
        self.table = self.dynamodb.Table(os.getenv("DYNAMODB_CHAT_TABLE"))

    def save_message(self, session_id: str, role: str, content: str, username: Optional[str] = None) -> bool:
        try:
            message = {
                "session_id": session_id,
                "timestamp": datetime.utcnow().isoformat(),
                "message_id": str(uuid.uuid4()),
                "role": role,
                "content": content,
            }
            if username:
                message["username"] = username

            self.table.put_item(Item=message)
            logger.info(f"Saved message: {message['message_id']}")
            return True
        except ClientError as e:
            logger.error(f"Save failed: {e}")
            return False

    def get_session_history(self, session_id: str, limit: int = 50) -> List[Dict]:
        try:
            response = self.table.query(
                KeyConditionExpression=Key("session_id").eq(session_id),
                ScanIndexForward=True,
                Limit=limit
            )
            return response.get("Items", [])
        except ClientError as e:
            logger.error(f"Fetch failed: {e}")
            return []

    def get_recent_sessions(self, limit: int = 10) -> List[str]:
        try:
            response = self.table.scan(
                ProjectionExpression="session_id, #ts",
                ExpressionAttributeNames={"#ts": "timestamp"},
                Limit=limit * 10
            )
            sessions = {}
            for item in response.get("Items", []):
                sid = item["session_id"]
                ts = item["timestamp"]
                if sid not in sessions or ts > sessions[sid]:
                    sessions[sid] = ts

            sorted_ids = sorted(sessions.items(), key=lambda x: x[1], reverse=True)
            return [sid for sid, _ in sorted_ids[:limit]]
        except ClientError as e:
            logger.error(f"Recent sessions fetch failed: {e}")
            return []

    def delete_session(self, session_id: str) -> bool:
        try:
            response = self.table.query(
                KeyConditionExpression=Key("session_id").eq(session_id),
                ProjectionExpression="session_id, #ts",
                ExpressionAttributeNames={"#ts": "timestamp"},
            )
            with self.table.batch_writer() as batch:
                for item in response["Items"]:
                    batch.delete_item(Key={"session_id": item["session_id"], "timestamp": item["timestamp"]})
            logger.info(f"Deleted session: {session_id}")
            return True
        except ClientError as e:
            logger.error(f"Delete failed: {e}")
            return False

    def get_message_count(self, session_id: str) -> int:
        try:
            response = self.table.query(
                KeyConditionExpression=Key("session_id").eq(session_id),
                Select="COUNT"
            )
            return response.get("Count", 0)
        except ClientError as e:
            logger.error(f"Count failed: {e}")
            return 0

async def load_chat_history(session_id: str) -> List[Dict[str, str]]:
    return dynamodb_service.get_session_history(session_id)

# インスタンスはシングルトンとして使用
dynamodb_service = DynamoDBService()

 フロントエンドの実装UIの構成は、React + Viteを使って構築しています。
Viteはビルドの速度が速く、開発中のホットリロードが快適です。また、React のコンポーネントベース設計を活かすことで、チャットバブルや入力欄、アニメーション付きの思考中表示などを部品化し、シンプルかつ拡張性の高い設計が可能になります
Chat.txt（チャット画面のサンプル例）
import React, { useState } from "react";
import { v4 as uuidv4 } from "uuid";

const sessionId = uuidv4(); // 毎回セッションIDを生成（固定でも可）

interface Message {
  role: "user" | "assistant";
  content: string;
}

const Chat: React.FC = () => {
  const [question, setQuestion] = useState("");
  const [messages, setMessages] = useState<Message[]>([]);

  const handleSubmit = async (e: React.FormEvent) => {
    e.preventDefault();

    // 質問を一旦 state に追加
    const userMessage: Message = { role: "user", content: question };
    setMessages((prev) => [...prev, userMessage]);

    try {
      const response = await fetch("http://localhost:8000/ask", {
        method: "POST",
        headers: {
          "Content-Type": "application/json",
        },
        body: JSON.stringify({
          question,
          session_id: sessionId,
        }),
      });

      const data = await response.json();
      const assistantMessage: Message = { role: "assistant", content: data.answer };
      setMessages((prev) => [...prev, assistantMessage]);
      setQuestion(""); // 入力欄をクリア
    } catch (error) {
      console.error("エラー:", error);
    }
  };

  return (
    <div style={{ maxWidth: "600px", margin: "0 auto", padding: "1rem" }}>
      <h1>チャットボット</h1>
      <div style={{ border: "1px solid #ccc", padding: "1rem", marginBottom: "1rem" }}>
        {messages.map((msg, index) => (
          <div key={index} style={{ marginBottom: "0.5rem" }}>
            <strong>{msg.role === "user" ? "あなた" : "AI"}:</strong> {msg.content}
          </div>
        ))}
      </div>
      <form onSubmit={handleSubmit}>
        <input
          type="text"
          value={question}
          onChange={(e) => setQuestion(e.target.value)}
          placeholder="質問を入力してください"
          style={{ width: "80%", marginRight: "1rem" }}
        />
        <button type="submit">送信</button>
      </form>
    </div>
  );
};

export default Chat;

 Thinkingアニメーションの工夫チャットでAIからの返答を待っている間、ただ「無音」で待たせると味気ない。

そこで「回答を生成中です…」のアニメーションをちょっとした演出として入れてます。
{isThinking && (
  <div className="flex items-center justify-center mt-4 mb-6">
    <div className="thinking-indicator">
      <div className="relative">
        <div className="w-8 h-8 border-2 loading-spinner rounded-full" />
      </div>
      <ThinkingMessage />
    </div>
  </div>
)}

 補足：好きな画像をくるくる回す演出に変更したい場合<div className="w-8 h-8 animate-spin">
  <img src="/solvio-logo.png" alt="Thinking" className="w-full h-full object-contain" />
</div>
Solvioのロゴマークを入れてくるくるアニメーションをつけてみました



最終的なUIは以下の動画を見てみてください！

https://www.youtube.com/watch?v=zvG5IbJsLBQ

 検証今回は『Solvio株式会社』のホームページに記載されている情報をS3にアップロードし、Kendraに探索させてみました。
実際に質問した際に返ってくる結果をテストしてみます。
例：『Solvio株式会社について教えてください』
input
リソースに対する質問

output
ユーザーの入力したワードに関する情報の返却

例：『この会社の企業概要を教えて』
input
前回の回答に対する質問

ロジック
DynamoDBに登録されているセッションIDを参照

output
質問の『この会社』を『Solvio株式会社』とAIが判断し、適切な回答を返却


 おわりに本記事では、FastAPI・OpenAI・AWS Kendra・DynamoDB を組み合わせたRAG型チャットアプリケーションの構築方法を紹介しました。
実装の中で特に重要となるのは、以下の3点でした：
Kendraによる高精度な情報検索
OpenAI APIを活用した自然な回答生成
DynamoDBを使ったセッション管理と履歴保持
こうした構成により、ユーザーの文脈を理解した対話が可能になり、「これってどういう意味？」といった曖昧な質問にも柔軟に対応できます。
生成AIを個人的利用だけではなく、業務に取り入れる第一歩として、RAG構成のチャットアプリは非常に汎用性が高く、応用もしやすいです。
興味がある方はぜひ参考にしてみてください！

脚注
Kendraのインデックスとは？

検索対象のデータ（ドキュメント類）を登録・管理するための入れ物みたいなもの。

質問が来たら、このインデックス内の情報をもとに検索する。 ↩︎
S3とは？

AWSが提供する「ファイル置き場」。PDFや画像、動画、HTMLなどのあらゆるファイルを保存可能で、Kendraのインデックスにリソースとして登録できます。 ↩︎
※なんでAIの返答を「JSON形式」で返させるの？

理由は、プログラム側で扱いやすくするためです。

項目ごとに内容を分けることによって、「人間が読む文章」ではなく、「アプリが使うデータ」にするためです。 ↩︎