この手順書では、OpenAIのLLM（ChatGPT API）・Embedding APIとPineconeのベクトルデータベースを組み合わせたRAG（Retrieval-Augmented Generation）チャットボットの具体的な実装方法を詳しく解説します。ファイル構造や保存場所、具体的なコード例まで含めて、ステップバイステップで構築できるよう構成しています。
まずは、プロジェクト用のディレクトリを作成し、以下のような構造にします：
rag-chatbot/ ├── .env # API鍵などの環境変数 ├── requirements.txt # 必要なパッケージリスト ├── data/ # 入力文書を保存するディレクトリ │ ├── documents/ # 元のドキュメントファイル │ └── processed/ # 処理済みファイル（オプション） ├── src/ # ソースコード │ ├── __init__.py │ ├── document_loader.py # 文書読み込み処理 │ ├── vector_store.py # ベクトル保存処理 │ ├── chat_bot.py # チャットボットロジック │ └── utils.py # ユーティリティ関数 └── app.py # メインアプリケーション
mkdir -p rag-chatbot/data/documents rag-chatbot/data/processed rag-chatbot/src cd rag-chatbot touch .env requirements.txt src/__init__.py src/document_loader.py src/vector_store.py src/chat_bot.py src/utils.py app.py
requirements.txtに以下の内容を記述します：
langchain>=0.1.0 langchain-openai>=0.0.5 langchain-community>=0.0.13 langchain-pinecone>=0.0.1 pinecone-client>=3.0.0 openai>=1.5.0 python-dotenv>=1.0.0 tiktoken>=0.5.2 unstructured>=0.10.30 pdf2image>=1.16.3 pytesseract>=0.3.10 pillow>=10.1.0 matplotlib>=3.8.2
仮想環境を作成してパッケージをインストールします：
python -m venv venv source venv/bin/activate # Windowsの場合: venv\Scripts\activate pip install -r requirements.txt
OpenAIとPineconeのAPIキーを取得し、.envファイルに保存します：
# .env OPENAI_API_KEY=your_openai_api_key_here PINECONE_API_KEY=your_pinecone_api_key_here PINECONE_ENVIRONMENT=your_pinecone_environment_here # 例: us-west1-gcp-free
処理したい文書ファイル（PDF、TXT、DOCXなど）を data/documents/ ディレクトリに配置します。
rag-chatbot/data/documents/ ディレクトリがすべての元ファイルの保管場所です。後でベクトル化された文書は、Pineconeのデータベースに保存されます。
src/document_loader.py に以下のコードを実装します：
import os
from typing import List, Optional
from langchain_community.document_loaders import (
PyPDFLoader,
TextLoader,
UnstructuredWordDocumentLoader,
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema.document import Document
class DocumentProcessor:
"""文書を読み込み、チャンクに分割するクラス"""
def __init__(
self,
documents_dir: str = "data/documents/",
chunk_size: int = 1000,
chunk_overlap: int = 200
):
self.documents_dir = documents_dir
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=len,
)
def load_documents(self, file_filter: Optional[str] = None) -> List[Document]:
"""指定ディレクトリから文書を読み込む"""
documents = []
for filename in os.listdir(self.documents_dir):
if file_filter and file_filter not in filename:
continue
file_path = os.path.join(self.documents_dir, filename)
try:
if filename.endswith(".pdf"):
loader = PyPDFLoader(file_path)
documents.extend(loader.load())
elif filename.endswith(".txt"):
loader = TextLoader(file_path)
documents.extend(loader.load())
elif filename.endswith((".docx", ".doc")):
loader = UnstructuredWordDocumentLoader(file_path)
documents.extend(loader.load())
# 他のファイル形式にも対応可能
except Exception as e:
print(f"Error loading {filename}: {e}")
return documents
def split_documents(self, documents: List[Document]) -> List[Document]:
"""文書をチャンクに分割する"""
return self.text_splitter.split_documents(documents)
src/vector_store.py に以下のコードを実装します：
import os
import pinecone
from typing import List, Optional
from dotenv import load_dotenv
from langchain_openai import OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain.schema.document import Document
# 環境変数を読み込む
load_dotenv()
class VectorStoreManager:
"""ベクトル変換とPineconeストアを管理するクラス"""
def __init__(
self,
index_name: str = "rag-knowledge-base",
namespace: Optional[str] = None,
embedding_model: str = "text-embedding-3-small"
):
self.index_name = index_name
self.namespace = namespace
self.embeddings = OpenAIEmbeddings(model=embedding_model)
# Pineconeの初期化
pinecone.init(
api_key=os.getenv("PINECONE_API_KEY"),
environment=os.getenv("PINECONE_ENVIRONMENT")
)
# インデックスがなければ作成
self._create_index_if_not_exists()
def _create_index_if_not_exists(self):
"""インデックスが存在しなければ作成する"""
if self.index_name not in pinecone.list_indexes():
# OpenAIのtext-embedding-3-smallは1536次元
pinecone.create_index(
name=self.index_name,
dimension=1536,
metric="cosine"
)
print(f"Created new Pinecone index: {self.index_name}")
def upsert_documents(self, documents: List[Document]) -> str:
"""文書をベクトル化してPineconeに保存する"""
if not documents:
return "No documents to upsert"
vector_store = PineconeVectorStore.from_documents(
documents=documents,
embedding=self.embeddings,
index_name=self.index_name,
namespace=self.namespace
)
return f"Upserted {len(documents)} documents to Pinecone index {self.index_name}"
def get_vector_store(self) -> PineconeVectorStore:
"""既存のベクトルストアを取得する"""
return PineconeVectorStore(
index_name=self.index_name,
embedding=self.embeddings,
namespace=self.namespace
)
def delete_namespace(self) -> str:
"""指定した名前空間のデータを削除する"""
if not self.namespace:
return "No namespace specified"
index = pinecone.Index(self.index_name)
index.delete(namespace=self.namespace)
return f"Deleted namespace {self.namespace} from index {self.index_name}"
これらのクラスを利用してPineconeにデータをアップロードするスクリプト create_index.py を作成します：
from src.document_loader import DocumentProcessor
from src.vector_store import VectorStoreManager
import argparse
def main():
parser = argparse.ArgumentParser(description='Index documents to Pinecone')
parser.add_argument('--dir', type=str, default='data/documents/',
help='Directory containing documents')
parser.add_argument('--filter', type=str, default=None,
help='Filter files by name')
parser.add_argument('--chunk-size', type=int, default=1000,
help='Chunk size for text splitting')
parser.add_argument('--overlap', type=int, default=200,
help='Overlap between chunks')
parser.add_argument('--index', type=str, default='rag-knowledge-base',
help='Pinecone index name')
parser.add_argument('--namespace', type=str, default=None,
help='Namespace within the index')
args = parser.parse_args()
# 1. 文書の読み込みと分割
print("Loading and splitting documents...")
processor = DocumentProcessor(
documents_dir=args.dir,
chunk_size=args.chunk_size,
chunk_overlap=args.overlap
)
documents = processor.load_documents(file_filter=args.filter)
print(f"Loaded {len(documents)} documents")
chunks = processor.split_documents(documents)
print(f"Split into {len(chunks)} chunks")
# 2. Pineconeへのアップロード
print("Uploading to Pinecone...")
vector_store = VectorStoreManager(
index_name=args.index,
namespace=args.namespace
)
result = vector_store.upsert_documents(chunks)
print(result)
if __name__ == "__main__":
main()
実行コマンド例：
python create_index.py --namespace company-docs --chunk-size 500
src/chat_bot.py にRAGチャットボットのコアロジックを実装します：
import os
from typing import List, Dict, Any
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.document import Document
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.output_parser import StrOutputParser
from langchain_pinecone import PineconeVectorStore
load_dotenv()
class RAGChatBot:
"""検索拡張生成（RAG）チャットボット"""
def __init__(
self,
vector_store: PineconeVectorStore,
model_name: str = "gpt-3.5-turbo",
temperature: float = 0.0,
top_k: int = 3
):
self.vector_store = vector_store
self.retriever = vector_store.as_retriever(
search_type="similarity",
search_kwargs={"k": top_k}
)
self.llm = ChatOpenAI(
model_name=model_name,
temperature=temperature
)
# プロンプトテンプレートの設定
self.prompt = ChatPromptTemplate.from_template("""
以下の情報源に基づいて質問に答えてください。
情報源に関連する情報がない場合は「情報が不足しています」と回答してください。
回答の最後に、関連する情報源を記載してください。
情報源:
{context}
質問: {question}
""")
# RAGチェーンの構築
self.chain = (
{"context": self.retriever, "question": RunnablePassthrough()}
| self.prompt
| self.llm
| StrOutputParser()
)
def _format_docs(self, docs: List[Document]) -> str:
"""取得した文書を整形する"""
return "\n\n".join(f"情報源 {i+1}:\n{doc.page_content}" for i, doc in enumerate(docs))
async def aask(self, question: str) -> str:
"""質問に対して非同期で回答する"""
return await self.chain.ainvoke(question)
def ask(self, question: str) -> str:
"""質問に対して回答する"""
return self.chain.invoke(question)
def get_related_docs(self, question: str) -> List[Document]:
"""質問に関連する文書を取得する"""
return self.retriever.get_relevant_documents(question)
top_k パラメータは、質問に関連する文書をいくつ取得するかを指定します。値が大きいほど多くの情報が得られますが、ノイズも増える可能性があります。
app.py にメインアプリケーションロジックを実装します：
import argparse
from dotenv import load_dotenv
from src.vector_store import VectorStoreManager
from src.chat_bot import RAGChatBot
load_dotenv()
def main():
parser = argparse.ArgumentParser(description='RAG Chatbot')
parser.add_argument('--index', type=str, default='rag-knowledge-base',
help='Pinecone index name')
parser.add_argument('--namespace', type=str, default=None,
help='Namespace within the index')
parser.add_argument('--model', type=str, default='gpt-3.5-turbo',
help='OpenAI model to use')
parser.add_argument('--temperature', type=float, default=0.0,
help='Model temperature')
parser.add_argument('--top-k', type=int, default=3,
help='Number of documents to retrieve')
args = parser.parse_args()
# ベクトルストアの取得
vector_store_manager = VectorStoreManager(
index_name=args.index,
namespace=args.namespace
)
vector_store = vector_store_manager.get_vector_store()
# チャットボットの初期化
chatbot = RAGChatBot(
vector_store=vector_store,
model_name=args.model,
temperature=args.temperature,
top_k=args.top_k
)
print("RAGチャットボットが起動しました。「終了」と入力すると終了します。")
while True:
question = input("\n質問を入力してください: ")
if question.lower() in ['終了', 'quit', 'exit']:
break
# 回答の生成
print("\n回答を生成中...\n")
answer = chatbot.ask(question)
print(answer)
# 関連文書の表示（オプション）
show_docs = input("\n関連文書を表示しますか？ (y/n): ")
if show_docs.lower() == 'y':
docs = chatbot.get_related_docs(question)
print("\n関連文書:")
for i, doc in enumerate(docs):
print(f"\n----- 文書 {i+1} -----")
print(doc.page_content)
if hasattr(doc, 'metadata') and doc.metadata:
print("\nメタデータ:", doc.metadata)
if __name__ == "__main__":
main()
実行コマンド例：
python app.py --namespace company-docs --model gpt-4o --top-k 5
簡単なWebインターフェースを作成するには、Streamlitを使用します：
pip install streamlit
streamlit_app.py を作成します：
import streamlit as st
from dotenv import load_dotenv
from src.vector_store import VectorStoreManager
from src.chat_bot import RAGChatBot
load_dotenv()
st.set_page_config(page_title="RAGチャットボット", layout="wide")
st.title("RAGチャットボット")
# サイドバー設定
with st.sidebar:
st.header("設定")
index_name = st.text_input("インデックス名", value="rag-knowledge-base")
namespace = st.text_input("名前空間", value="")
if namespace == "":
namespace = None
model_name = st.selectbox(
"モデル",
["gpt-3.5-turbo", "gpt-4o", "gpt-4-turbo"]
)
temperature = st.slider("温度", min_value=0.0, max_value=1.0, value=0.0, step=0.1)
top_k = st.slider("取得文書数", min_value=1, max_value=10, value=3)
show_sources = st.checkbox("関連文書を表示", value=True)
# ベクトルストアとチャットボットの初期化
@st.cache_resource
def initialize_chatbot(index_name, namespace, model_name, temperature, top_k):
vector_store_manager = VectorStoreManager(
index_name=index_name,
namespace=namespace
)
vector_store = vector_store_manager.get_vector_store()
return RAGChatBot(
vector_store=vector_store,
model_name=model_name,
temperature=temperature,
top_k=top_k
)
chatbot = initialize_chatbot(index_name, namespace, model_name, temperature, top_k)
# チャット履歴の初期化
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 prompt := st.chat_input("質問を入力してください"):
# ユーザーメッセージを追加
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("user"):
st.markdown(prompt)
# 回答生成
with st.chat_message("assistant"):
with st.spinner("回答を生成中..."):
response = chatbot.ask(prompt)
st.markdown(response)
# 関連文書の表示
if show_sources:
docs = chatbot.get_related_docs(prompt)
with st.expander("関連文書"):
for i, doc in enumerate(docs):
st.markdown(f"##### 文書 {i+1}")
st.text(doc.page_content)
if hasattr(doc, 'metadata') and doc.metadata:
st.text(f"メタデータ: {doc.metadata}")
# アシスタントメッセージを追加
st.session_state.messages.append({"role": "assistant", "content": response})
Streamlitアプリを実行するコマンド：
streamlit run streamlit_app.py
http://localhost:8501 でアプリにアクセスできます。
RESTful APIとして提供するには、FastAPIを使用します：
api_server.py を作成します：
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from src.vector_store import VectorStoreManager
from src.chat_bot import RAGChatBot
from dotenv import load_dotenv
load_dotenv()
app = FastAPI(title="RAG Chatbot API")
# リクエスト/レスポンスモデル
class ChatRequest(BaseModel):
question: str
index_name: str = "rag-knowledge-base"
namespace: Optional[str] = None
model_name: str = "gpt-3.5-turbo"
temperature: float = 0.0
top_k: int = 3
include_sources: bool = False
class Source(BaseModel):
content: str
metadata: Optional[dict] = None
class ChatResponse(BaseModel):
answer: str
sources: Optional[List[Source]] = None
# チャットボットのキャッシュ
chatbot_cache = {}
def get_chatbot(index_name: str, namespace: Optional[str], model_name: str, temperature: float, top_k: int):
"""チャットボットインスタンスを取得またはキャッシュから返す"""
cache_key = f"{index_name}:{namespace}:{model_name}:{temperature}:{top_k}"
if cache_key not in chatbot_cache:
vector_store_manager = VectorStoreManager(
index_name=index_name,
namespace=namespace
)
vector_store = vector_store_manager.get_vector_store()
chatbot_cache[cache_key] = RAGChatBot(
vector_store=vector_store,
model_name=model_name,
temperature=temperature,
top_k=top_k
)
return chatbot_cache[cache_key]
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
try:
chatbot = get_chatbot(
request.index_name,
request.namespace,
request.model_name,
request.temperature,
request.top_k
)
# 回答の生成
answer = chatbot.ask(request.question)
# オプションで関連文書を含める
sources = None
if request.include_sources:
docs = chatbot.get_related_docs(request.question)
sources = [
Source(
content=doc.page_content,
metadata=doc.metadata if hasattr(doc, 'metadata') else None
)
for doc in docs
]
return ChatResponse(
answer=answer,
sources=sources
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/health")
async def health_check():
"""ヘルスチェックエンドポイント"""
return {"status": "healthy"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
FastAPIサーバーを実行するコマンド：
uvicorn api_server:app --reload
http://localhost:8000/docs にアクセスすると、Swagger UIでAPIドキュメントを確認できます。
元の文書ファイルは data/documents/ に保存し、ベクトル化されたデータはPineconeのクラウドサービス上に保管されています。APIキーなどの環境変数は .env ファイルで管理。
Pineconeの名前空間 (namespace) 機能を活用すると、1つのインデックス内で複数のデータセットを分離管理できます。例えば部署ごと、製品ごと、言語ごとにnamespaceを分けるなど。
大量の文書を処理する場合は、バッチ処理を検討してください。例えば100件ずつアップロードし、処理間に適切な待機時間を挟むことで、API制限エラーを回避できます。
無料プランでは最大100kベクトル（次元1536の場合）の制限があります。大量データを扱う場合は、小さい次元の埋め込みモデルを使うか、有料プランへのアップグレードを検討してください。
.env ファイルはGitリポジトリにコミットしないよう注意してください。本番環境では、環境変数を安全に管理するクラウドサービスの仕組みを活用することをお勧めします。
文書が更新される場合は、定期的に再インデックスすることを検討してください。更新頻度に応じてスクリプトを自動実行する仕組みを構築するとよいでしょう。
Pinecone公式ドキュメント - ベクトルデータベースの詳細な使い方
OpenAI Embeddings API - 埋め込みモデルの使い方
LangChain Pinecone統合 - LangChainでのPinecone活用法
Pinecone公式サンプルコード - 様々なユースケースの実装例
LangChain Cookbook - RAGパターンの実装例
日本語文書を扱う場合は、多言語対応の埋め込みモデルを使用することを強くお勧めします。
OpenAIのtext-embedding-3-largeは日本語の理解も良好です。
チャンク分割も文字単位で適切なサイズに調整することが重要です。
OpenAI API: 埋め込み生成とLLM呼び出しに料金が発生します。
Pinecone: 無料プランは容量限定、有料プランはvCPUとデータ量に応じて課金。
コスト削減には、キャッシュ機構の実装や低コストのモデル選択が効果的です。