PR

【第2回】バックエンド編：PythonとLangChainで構築するRAG API実践入門

【第2回】バックエンド編：PythonとLangChainで構築するRAG API実践入門

はじめに

LLMアプリケーション開発連載、第2回へようこそ。前回はLLMアプリの全体像と技術スタックを学びました。今回からはいよいよ、具体的なコーディングに入ります。

本記事では、LLMアプリケーションのまさに心臓部となるバックエンドAPIを、Pythonを使ってゼロから構築します。私たちが目指すのは、単なるオウム返しのチャットボットではありません。独自のドキュメント（知識）を読み込み、その内容に基づいてユーザーの質問に答える、賢いQ&Aボットです。

この「外部知識を参照する」仕組みはRAG (Retrieval-Augmented Generation) と呼ばれ、現代のLLMアプリ開発において最も重要な技術の一つです。

この記事を読み終える頃には、あなたは以下の技術を使いこなし、高性能なRAG APIを構築するスキルを身につけているでしょう。

  • FastAPI: モダンで高速なPythonのAPIフレームワーク。
  • LangChain: LLMアプリの複雑な処理フローを驚くほどシンプルに記述できるオーケストレーションフレームワーク。
  • LCEL (LangChain Expression Language): LangChainの処理をパイプラインのように繋ぎ合わせるための、宣言的で強力な記法。

それでは、さっそく開発を始めましょう！

1. RAGパイプラインのアーキテクチャ

まず、RAGがどのような仕組みで動いているのかを理解しましょう。RAGの処理は、大きく分けて「Indexing（索引作成）」と「Retrieval & Generation（検索と生成）」の2つのフェーズから成り立ちます。

RAG Architecture

  • Indexing（事前準備）: あなたが持っているドキュメント（PDF、Markdownなど）を、LLMが検索しやすいように前処理し、データベースに格納しておくフェーズ。

    1. Load: ドキュメントを読み込みます。
    2. Split: 長い文章を、意味のあるまとまり（チャンク）に分割します。
    3. Embed: 各チャンクを、意味を捉えた数値の配列（ベクトル）に変換します。
    4. Store: ベクトルデータを、高速に検索できる「Vector Store（ベクトルデータベース）」に保存します。

  • Retrieval & Generation（実行時）: ユーザーからの質問に答えるフェーズ。

    1. ユーザーの質問も同様にベクトルに変換します。
    2. Vector Storeの中から、質問ベクトルと意味的に類似したチャンク（ベクトル）をいくつか検索・取得します。
    3. 取得したチャンク（＝コンテキスト）と元の質問を組み合わせ、「以下の情報を参考にして、質問に答えてください」という形式のプロンプトを作成します。
    4. LLMが、そのプロンプトに基づいて最終的な回答を生成します。

2. 【実践】開発環境のセットアップ

2.1 仮想環境とライブラリインストール

mkdir llm-rag-api && cd llm-rag-api
python3 -m venv venv
source venv/bin/activate
pip install fastapi uvicorn python-dotenv openai langchain langchain-openai faiss-cpu pypdf
  • faiss-cpu: Facebook (Meta) が開発した、高速なベクトル検索ライブラリ。今回はローカルで手軽に試せるインメモリのVector Storeとして利用します。
  • pypdf: PDFファイルを読み込むためのライブラリです。

2.2 APIキーとドキュメントの準備

プロジェクトルートに.envファイルを作成し、OpenAIのAPIキーを記述します。

OPENAI_API_KEY="sk-..."

また、知識源としたいPDFファイルをdocumentsというディレクトリを作成して、その中に入れておきましょう。（例: documents/my-knowledge.pdf）

3. LangChainでRAGチェーンを構築する

ここからが本番です。rag_chain.pyというファイルを作成し、RAGの処理フローをLangChainを使って実装していきます。

“`python:rag_chain.py
import os
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.prompts import ChatPromptTemplate
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.output_parser import StrOutputParser

1. ドキュメントの読み込みとVector Storeの作成（初回のみ）

def get_vectorstore_from_pdf(pdf_path):
loader = PyPDFLoader(pdf_path)
documents = loader.load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=100)
docs = text_splitter.split_documents(documents)

embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(docs, embeddings)
return vectorstore

2. RAGチェーンの構築

def create_rag_chain(pdf_path):
vectorstore = get_vectorstore_from_pdf(pdf_path)
retriever = vectorstore.as_retriever()

# プロンプトテンプレートの定義
template = """あなたは親切なAIアシスタントです。
以下のコンテキスト情報のみを使って、ユーザーの質問に答えてください。
コンテキスト:
{context}
質問: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
llm = ChatOpenAI(model_name="gpt-4o", temperature=0)
# LCEL (LangChain Expression Language) を使ってチェーンを組み立てる
rag_chain = (
{"context": retriever, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
return rag_chain

**LCELの解説**: `|`（パイプ）演算子で処理を繋いでいる部分がLCELです。SQLのパイプラインのように、左から右へデータが流れていきます。`{"context": retriever, "question": RunnablePassthrough()}` の部分で、ユーザーの質問 (`RunnablePassthrough()`) と、その質問を元に`retriever`が検索した結果 (`context`) を辞書として次の`prompt`に渡しています。これにより、複雑な処理フローを非常に直感的に記述できます。
## 4. FastAPIでAPIエンドポイントを作成する
次に、`main.py`を作成し、今作ったRAGチェーンを呼び出すAPIを定義します。
```python:main.py
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from dotenv import load_dotenv
from rag_chain import create_rag_chain
import asyncio
# .envファイルを読み込む
load_dotenv()
app = FastAPI()
# RAGチェーンをグローバルに保持（本番ではより良い方法を検討）
rag_chain = create_rag_chain("./documents/my-knowledge.pdf")
class Query(BaseModel):
question: str
@app.post("/ask")
async def ask_question(query: Query):
# 非同期ストリーミングで回答を生成
async def stream_generator():
async for chunk in rag_chain.astream(query.question):
yield chunk
return StreamingResponse(stream_generator(), media_type="text/plain")
@app.get("/")
def read_root():
return {"message": "RAG API is running."}

ポイント: rag_chain.astream(query.question) を使っています。これはLCELで構築されたチェーンを非同期で実行し、結果をストリームとして返すメソッドです。FastAPIのStreamingResponseと組み合わせることで、LLMが生成したトークンを一つずつ、リアルタイムでクライアントに送信できます。

5. 動作確認

ターミナルでAPIサーバーを起動します。

uvicorn main:app --reload

別のターミナルからcurlコマンドでAPIを叩いてみましょう。

curl -X POST "http://127.0.0.1:8000/ask" \
-H "Content-Type: application/json" \
-d '{"question": "（あなたのPDFに関する質問）"}' --no-buffer

--no-bufferオプションを付けることで、レスポンスがストリーミングで少しずつ返ってくる様子が確認できます。

まとめ

今回は、FastAPIとLangChainを使い、LLMアプリケーションのバックエンドとなるRAG APIを構築しました。特にLCELを使うことで、複雑なRAGのロジックを驚くほどシンプルに、かつ宣言的に記述できることを体験いただけたかと思います。

しかし、これだけではまだ「API」でしかありません。ユーザーが快適に使えるUIが必要です。

次回、【第3回】フロントエンド編では、今回作成したAPIをNext.jsから呼び出し、ChatGPTのようなリアルタイムストリーミングを実現するモダンなチャットUIを構築していきます。お楽しみに！

“`

コメント

タイトルとURLをコピーしました