こんにちは、JS2IIUです。
連載「StreamlitとRAGで作る：実用的なAIチャットボット開発ガイド」の第5回をお届けします。
これまで私たちは、AIアプリを構築するための「部品」を一つずつ丁寧に作り上げてきました。
- UIの基礎: Streamlitの使い方をマスターしました。
- RAGの概念: 外部知識を注入する仕組みを理解しました。
- データ加工: ドキュメントを読み込み、チャンキングしました。
- 記憶の保管庫: ベクトルデータベース（Vector Store）を構築しました。
そして今回、ついにこれら全ての部品を統合します。
今回のゴールは明確です。「独自のドキュメントに基づいて回答する、動くチャットボットアプリ」を完成させることです。
これまではスクリプト単位での実験でしたが、今回はブラウザ上でインタラクティブに動くアプリケーションとして仕上げます。自分のデータについてAIが流暢に答えてくれる感動を、ぜひ手元で体験してください。今回もよろしくお願いします。
1. はじめに：部品を組み立てる時が来た
RAGアプリケーションの構造は、料理に例えると「オーダーを受けてから冷蔵庫（Vector Store）の食材（データ）を探し、シェフ（LLM）が調理して、ウェイター（Streamlit）が提供する」という流れになります。
これまでの連載で、食材の下準備と冷蔵庫への保存は完了しました。今回は、ウェイターがオーダーを取り、裏側のキッチンと連携するシステムを構築します。
アプリケーションの処理フロー:
- User: チャット欄に質問を入力。
- Streamlit: 入力を受け取り、RAGパイプラインを起動。
- Retriever: 質問に関連するドキュメントを検索。
- LLM: 検索結果を参考に回答を生成。
- Streamlit: 回答を画面に表示。
この一連の流れを、たった一つのPythonファイル app.py にまとめていきます。
2. Step 1：検索エンジン（Retriever）の準備とキャッシュ化
まずは、第4回で保存したFAISSインデックスを読み込みます。ここで重要なのが、Streamlitの「再実行」という特性への対策です。
ユーザーが何か入力するたびに重いVector Storeをディスクから読み込み直していては、動作が遅くなりすぎます。
そこで、st.cache_resource を使って、一度読み込んだデータベースをメモリ上にキャッシュします。
import streamlit as st
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
# ページ設定
st.set_page_config(page_title="RAG Chatbot", page_icon="🤖")
st.title("🤖 RAGチャットボット")
# 1. Vector Storeの読み込み関数（キャッシュ化）
@st.cache_resource
def load_vector_store():
# Embeddingモデルの初期化
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# ローカルに保存されたFAISSインデックスを読み込み
# ※ 第4回で作成した "faiss_index" フォルダがある前提です
try:
vectorstore = FAISS.load_local(
"faiss_index",
embeddings,
allow_dangerous_deserialization=True
)
return vectorstore
except Exception as e:
st.error(f"Vector Storeの読み込みに失敗しました: {e}")
return None
# アプリ起動時に一度だけ実行される
vectorstore = load_vector_store()
if vectorstore is None:
st.stop() # 読み込み失敗時はここで処理を止める
# 2. Retriever（検索機）への変換
# k=3 は「上位3件の関連ドキュメントを取得する」という意味
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})コードのポイント
@st.cache_resource: データベース接続やMLモデルなど、変更されない重いリソースを保持するために使います。as_retriever: Vector Storeは単なるデータベースですが、これをLangChainのチェーンに組み込める「検索コンポーネント」に変換するメソッドです。
3. Step 2：RAGチェーンの定義
次に、検索したドキュメントとユーザーの質問をLLMに渡すための「処理パイプライン（チェーン）」を定義します。
ここでは、現代的なLangChainの書き方である LCEL (LangChain Expression Language) を少しだけ先取りして使います。直感的で読みやすいのが特徴です。
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
# プロンプトテンプレートの定義
template = """
あなたは優秀なアシスタントです。
以下の「コンテキスト（検索された情報）」に基づいて、ユーザーの質問に答えてください。
もしコンテキストに情報がない場合は、「その情報は持ち合わせていません」と答えてください。
# コンテキスト
{context}
# 質問
{question}
"""
prompt = ChatPromptTemplate.from_template(template)
# LLMの定義
llm = ChatOpenAI(model_name="gpt-4o-mini", temperature=0)
# ドキュメントを結合して文字列にするヘルパー関数
def format_docs(docs):
return "\n\n".join([d.page_content for d in docs])
# RAGチェーンの構築（LCEL記法）
# 1. retrieve: 質問を受け取り、retrieverで検索を実行
# 2. assign: 検索結果(docs)をformat_docsで文字列化し、context変数に入れる
# 3. prompt: promptテンプレートに値を埋め込む
# 4. llm: LLMに渡す
# 5. parse: 結果を文字列として取り出す
rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)この rag_chain は、rag_chain.invoke("質問") と呼び出すだけで、検索から回答生成までを一気通貫で行ってくれる便利なオブジェクトです。
LCEL（LangChain Expression Language）について詳しく
LCELとは？
LCEL（LangChain Expression Language）は、LangChain v0.1.0以降で導入された、処理パイプラインを宣言的に記述するためのAPIです。従来の LLMChain や SequentialChain といったクラスベースの実装を置き換え、より直感的で柔軟な処理フローを実現します。
概念：Runnableインターフェース
LCELの基盤となるのが Runnable プロトコル（インターフェース）です。LangChainのほぼすべてのコンポーネント（Prompt、LLM、Retriever、Parser等）は Runnable を実装しており、以下の統一メソッドを持ちます：
invoke(input): 単一入力を同期的に実行batch(inputs): 複数入力をバッチ処理stream(input): ストリーミング実行（逐次出力）ainvoke(input): 非同期実行（async/await）
パイプ演算子（|）の仕組み
LCELの最大の特徴は、Pythonのパイプ演算子 | を使った処理の連結です。これは内部的には __or__ メソッドのオーバーロードで実装されています。
# 従来の書き方（LangChain v0.0.x）
chain = LLMChain(llm=llm, prompt=prompt)
output_parser = StrOutputParser()
result = output_parser.parse(chain.run(input))
# LCEL記法（v0.1.x以降）
chain = prompt | llm | StrOutputParser()
result = chain.invoke(input)パイプで連結すると、内部的には RunnableSequence というオブジェクトが生成され、前のステップの出力が次のステップの入力に自動的にマッピングされます。
辞書によるパラレル実行
LCELでは辞書リテラルを使うことで、複数の処理を並列実行できます。
{
"context": retriever | format_docs, # パス1：検索→整形
"question": RunnablePassthrough() # パス2：入力をそのまま通過
}この記法は内部的に RunnableParallel を生成し、context と question の2つの処理を並列に実行します。結果は辞書として次のステップに渡されます。
RunnablePassthrough の役割
RunnablePassthrough() は、入力をそのまま出力に渡す「透過的な」Runnableです。上記の例では、ユーザーの質問文を加工せずに question キーに割り当てています。
従来手法との比較
| 観点 | 従来（LLMChain等） | LCEL |
|---|---|---|
| 記述方法 | クラスのインスタンス化と実行 | パイプ演算子での連結 |
| 可読性 | ネストが深くなりがち | 処理フローが左から右へ直線的 |
| ストリーミング | 個別実装が必要 | .stream() で統一対応 |
| 並列実行 | 手動で制御 | 辞書記法で自動並列化 |
| 型安全性 | 弱い（実行時エラー） | 強い（入出力型が明確） |
実行時のデータフロー
上記のRAGチェーンに "質問文" を渡した場合の内部動作：
- 入力:
"質問文"（文字列） - RunnableParallel実行:
retriever.invoke("質問文")→ 関連ドキュメントのリストformat_docs(ドキュメントリスト)→ 結合されたテキスト（context）RunnablePassthrough().invoke("質問文")→"質問文"（question）
- 辞書出力:
{"context": "...", "question": "質問文"} - prompt.invoke(辞書) → テンプレートに値を埋め込んだPromptValue
- llm.invoke(PromptValue) → AIMessage
- StrOutputParser().invoke(AIMessage) → 文字列の回答
LCELの利点
- 宣言的: 「何をするか」に集中でき、実装の詳細を隠蔽
- 再利用性: 小さなRunnableを組み合わせて複雑なフローを構築
- デバッグ性: 各ステップを個別にテスト可能
- パフォーマンス: 内部で最適化（並列実行、バッチ処理）
LCELは現代のLangChain開発における標準記法となっており、これをマスターすることで、複雑なAIアプリケーションを簡潔に記述できるようになります。
4. Step 3：チャットUIの実装
バックエンドの準備が整いました。ここからはStreamlitを使ったフロントエンド（UI）の実装です。
第1回で学んだ st.session_state を使って、画面上のメッセージ表示を管理します。
# チャット履歴の初期化
if "messages" not in st.session_state:
st.session_state.messages = []
# 過去のメッセージを表示（再実行のたびに描画）
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.markdown(message["content"])そして、ユーザー入力を受け付けるメインループを記述します。
# ユーザー入力の受け付け
if query := st.chat_input("質問を入力してください..."):
# 1. ユーザーの入力を表示
st.session_state.messages.append({"role": "user", "content": query})
with st.chat_message("user"):
st.markdown(query)
# 2. AIの応答生成
with st.chat_message("assistant"):
with st.spinner("検索中..."):
try:
# ここでRAGチェーンを実行！
response = rag_chain.invoke(query)
st.markdown(response)
# 履歴に追加
st.session_state.messages.append({"role": "assistant", "content": response})
except Exception as e:
st.error(f"エラーが発生しました: {e}")これだけで、基本的なチャットボットとして機能します。しかし、RAGアプリとしては「本当にドキュメントを見て答えたのか？」を確認したいですよね。
5. Step 4：回答の根拠（ソース）を表示する
RAGの最大の利点は、回答の根拠を提示できることです。LangChainのチェーンを少し修正して、検索されたドキュメント自体も取得できるようにし、Streamlitの st.expander（折りたたみ表示）を使って表示してみましょう。
LCELでソースも同時に返すように書き換えるのは少し複雑になるため、ここでは「検索」と「生成」をあえて分けて記述する、より平易な実装パターンを紹介します。
# 修正版：メイン処理ブロック
if query := st.chat_input("質問を入力してください..."):
# ...（ユーザー入力表示部分は同じ）...
with st.chat_message("assistant"):
with st.spinner("ドキュメントを検索して回答を生成中..."):
# Step A: 関連ドキュメントの検索
relevant_docs = retriever.invoke(query)
context_text = format_docs(relevant_docs)
# Step B: LLMによる回答生成
# チェーンの一部（Prompt -> LLM -> Parser）だけを実行
generation_chain = prompt | llm | StrOutputParser()
response = generation_chain.invoke({
"context": context_text,
"question": query
})
# 回答の表示
st.markdown(response)
# Step C: 参照ドキュメントの表示
with st.expander("参照したドキュメントを確認する"):
for i, doc in enumerate(relevant_docs):
st.markdown(f"**出典 {i+1}:**")
st.text(doc.page_content) # 引用文を表示
# メタデータがあれば表示（ファイル名やページ数など）
if "source" in doc.metadata:
st.caption(f"ソース: {doc.metadata['source']}")
st.divider()
# 履歴に追加
st.session_state.messages.append({"role": "assistant", "content": response})この実装により、ユーザーはAIの回答の下にある「参照したドキュメントを確認する」をクリックすることで、AIがどの情報を元に判断したのかを自分の目で確かめることができます。これが実務アプリケーションにおける信頼性の鍵となります。
6. 完成コード全体像
これまでの内容をまとめた app.py の全コードです。
（前提：プロジェクトフォルダに .env と、第4回で作成した faiss_index フォルダが存在すること）
import streamlit as st
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv
# 環境変数の読み込み
load_dotenv()
# --- 設定 ---
st.set_page_config(page_title="RAG Chatbot", page_icon="🤖")
st.title("🤖 RAGチャットボット")
# --- 1. リソースの読み込み (Cache) ---
@st.cache_resource
def load_retriever():
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
try:
vectorstore = FAISS.load_local(
"faiss_index",
embeddings,
allow_dangerous_deserialization=True
)
# 検索機（Retriever）として返す
return vectorstore.as_retriever(search_kwargs={"k": 3})
except Exception as e:
return None
retriever = load_retriever()
if not retriever:
st.error("Vector Storeが見つかりません。先にインデックスを作成してください。")
st.stop()
# --- 2. プロンプトとLLMの準備 ---
template = """
あなたは優秀なアシスタントです。
以下の「コンテキスト」に基づいて、ユーザーの質問に答えてください。
# コンテキスト
{context}
# 質問
{question}
"""
prompt = ChatPromptTemplate.from_template(template)
llm = ChatOpenAI(model_name="gpt-4o-mini", temperature=0)
# --- 3. チャット履歴の管理 ---
if "messages" not in st.session_state:
st.session_state.messages = []
# 履歴の表示
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.markdown(message["content"])
# --- 4. メインチャットループ ---
if query := st.chat_input("質問を入力してください..."):
# ユーザー入力の表示と保存
st.session_state.messages.append({"role": "user", "content": query})
with st.chat_message("user"):
st.markdown(query)
# AI応答の処理
with st.chat_message("assistant"):
with st.spinner("検索中..."):
# ドキュメント検索
relevant_docs = retriever.invoke(query)
context_text = "\n\n".join([d.page_content for d in relevant_docs])
# 回答生成
chain = prompt | llm | StrOutputParser()
response = chain.invoke({"context": context_text, "question": query})
st.markdown(response)
# ソースの表示
with st.expander("参照元の情報を表示"):
for doc in relevant_docs:
st.markdown(f"- {doc.page_content[:100]}...") # 冒頭100文字
st.caption(f"Source: {doc.metadata.get('source', 'Unknown')}")
# 履歴に追加
st.session_state.messages.append({"role": "assistant", "content": response})ターミナルで streamlit run app.py を実行してみてください。あなたの独自のデータについて答えてくれるボットが立ち上がります！
7. まとめと次回予告
おめでとうございます！ ついにRAGアプリケーションの基本形が完成しました。
- 統合: StreamlitのUIとLangChainのロジックが繋がり、アプリとして機能しました。
- 効率化:
st.cache_resourceにより、快適な動作速度を実現しました。 - 透明性: 参照ドキュメントを表示することで、信頼できる回答提示が可能になりました。
しかし、実際に使ってみると「少し物足りない点」に気づくかもしれません。
「回答が表示されるまで、じっと待っていなければならない（タイピング中のように表示されない）」
「前の会話の内容を覚えていない（文脈を理解しない）」
これらはユーザー体験（UX）において非常に重要です。
次回、第6回「ユーザー体験を向上：ストリーミング応答とチャット履歴の管理」では、回答をリアルタイムでパラパラと表示する「ストリーミング機能」と、会話の流れを記憶させて「さっきの件だけど…」といった指示に対応させる方法を実装します。
アプリのクオリティを一段階引き上げるテクニックにご期待ください。最後まで読んでいただきありがとうございます。
コメント