freeAiChat【第2回】：FastAPI × LangChain × Weaviate で構築された RAG対応バックエンドを詳細解説

• 📂 目次：【AIチャット（freeAiChat）無料構築】連載の全記事まとめ
• 【第1回】無料で構築できるRAG × マルチLLM対応AIチャットシステムの全体像を紹介
• 【第2回】FastAPI × LangChain × Weaviate で構築された RAG対応バックエンド（閲覧中）
• 【【第3回】React × Next.js × shadcn/ui で構築されたチャットフロントエンドを詳細解説
• 💡 今後も開発効率化・ツール連携に関する記事を随時追加していきます！

🎯 本記事の対象読者

• 第1回を読み、freeAiChat の全体像を把握済みの方
• 「RAGの仕組みをコードレベルで理解したい」という方
• Python / FastAPI の基礎的な知識がある方（変数、関数、クラスの理解程度）
• LangChain のパイプライン構築（Runnable や | 演算子）に興味がある方
• ベクトルDB（Weaviate）のスキーマ定義や検索の仕組みを知りたい方

💡 第1回をまだ読んでいない方は、先にそちらからご覧ください。
全体像の把握なしに本記事を読むと、「どのコードが何の役割を持つのか」が分かりにくくなります。

⚠️ 本リポジトリの位置づけ（第1回同様）

freeAiChat は、AI と RAG の仕組みを「動かしながら学ぶ」ことを目的としたサンプル・教材です。

以下は意図的に簡略化・未対応です。ご理解の上ご利用ください。

• セキュリティ対策（認証・認可、CORS厳格化、入力検証の網羅性など）
• 本番運用を想定した設計（ログ管理、監視、スケーリングなど）
• エラーハンドリングの完全な網羅

「まず動かして仕組みを理解する」→「その後、必要に応じて本番対応を追加する」 という流れを想定しています。

---

🔧 ai-chat-backend の技術スタック

技術	役割	無料運用
FastAPI	APIサーバー・自動ドキュメント生成	✔
LangChain	RAG処理のオーケストレーション	✔
Weaviate OSS版	ベクトルデータベース（Docker）	✔
sentence-transformers	Embeddingモデル（all-MiniLM-L6-v2）	✔
Groq / OpenAI	LLM呼び出し	✔（無料枠）

---

🗂️ ファイル構成

ai-chat-backend/
├─ .env                    ← 環境変数（APIキー等）
├─ app.py                  ← コアサービス（本記事の主役）
├─ init_weaviate.py        ← Weaviateのコレクション初期化
├─ docker-compose.yml      ← Weaviate起動用
├─ requirements.txt        ← 依存パッケージ
└─ README.md               ← 詳細なセットアップ手順

---

🚀 起動から動作確認までの流れ

詳細なコマンドは GitHubリポジトリのREADME を参照ください。ここでは全体の流れを示します。

# 1. WeaviateをDockerで起動
docker-compose up -d

# 2. Python仮想環境を有効化
source venv/bin/activate

# 3. Weaviateのコレクション（DBテーブル相当）を初期化
python init_weaviate.py

# 4. APIサーバーを起動
uvicorn app:app --reload

# 5. Swagger UIでAPIを確認
open http://localhost:8000/docs

---

🧩 RAG処理フロー（内部構造）

/ask エンドポイントが処理するRAGの流れは以下の通りです。

---

📦 Weaviate のスキーマ構成

RAGの基盤となるベクトルDBのスキーマは、意図的に最小構成にしています。

init_weaviate.py

import os
import weaviate
from weaviate.classes.config import Property, DataType
from weaviate.embedded import EmbeddedOptions

# 環境変数から接続先を取得（デフォルト: localhost:8080）
weaviate_url = os.getenv("WEAVIATE_URL", "http://localhost:8080")

# Weaviateクライアントの初期化
try:
    client = weaviate.connect_to_local(
        host="localhost",
        port=8080,
        grpc_port=50051
    )
    print("既存のWeaviateインスタンスに接続しました")
except Exception as e:
    # 接続失敗時は組み込みモードで自動起動（フォールバック）
    client = weaviate.WeaviateClient(
        embedded_options=EmbeddedOptions(
            hostname="localhost",
            port=8090,
            persistence_data_path="./weaviate_data"
        )
    )
    print("Weaviateを組み込みモードで起動しました")

index_name = os.getenv("WEAVIATE_INDEX_NAME", "DefaultCollection")

# コレクション（スキーマ）が存在しなければ作成
if not client.collections.exists(index_name):
    client.collections.create(
        name=index_name,
        properties=[
            Property(name="text", data_type=DataType.TEXT)
        ],
        vectorizer_config=None  # 外部Embeddingモデルを使用するため無効化
    )
    print("コレクションを作成しました")
else:
    print("コレクションは既に存在します")

client.close()

ポイント：

• スキーマは1フィールドのみ：text（本文）。RAGの最小構成を学ぶには余計なフィールドが不要です。
• vectorizer_config=None：Weaviate標準のベクトル化機能を使わず、sentence-transformers で自前でベクトル化します（無料・オフライン可）。
• フォールバック機構：Docker未起動時でも、組み込みモードでWeaviateが自動起動します（初学者が躓きにくい配慮）。

---

🔌 API仕様とリクエスト/レスポンス例

POST /ask — RAG質問応答

リクエスト：

{
  "question": "LangChainとは何ですか？"
}

レスポンス：

{
  "answer": "LangChainは大規模言語モデルアプリケーションの開発用フレームワークです。"
}

エラー時：

{
  "error": "Weaviateへの接続に失敗しました..."
}

---

POST /ingest — テキスト直接登録

リクエスト：

{
  "text": "LangChainはLLMアプリの開発フレームワークです。"
}

レスポンス：

{
  "status": "success",
  "message": "ドキュメントが知識ベースに保存されました"
}

---

POST /upload — PDF/TXTファイル登録

リクエスト（multipart/form-data）：

curl -X POST "http://localhost:8000/upload/" \
  -F "file=@manual.pdf" \
  -F "chunk_size=2000" \
  -F "preprocess=true"

レスポンス：

{
  "message": "File uploaded successfully, 15/15個のチャンクを保存しました",
  "filename": "manual.pdf",
  "ingest_result": {
    "status": "success",
    "message": "15/15個のチャンクを保存しました",
    "details": {
      "chunk_size": 2000,
      "preprocessing": true,
      "source_directory": "./doc/pdfs"
    }
  }
}

---

POST /ingest-url — URLスクレイピング登録

リクエスト：

{
  "url": "https://example.com/article",
  "chunk_size": 1500,
  "preprocess": true
}

レスポンス：

{
  "status": "success",
  "message": "42/45個のチャンクを保存しました",
  "details": {
    "url": "https://example.com/article",
    "chunk_size": 1500,
    "content_length": 12800
  }
}

💡 42/45個 のように一部失敗する場合があります。これはバッチ保存時のエラーに対し、個別リトライを行った結果です（後述のエラーハンドリング参照）。

---

🧠 LangChain の RAGチェーン構成

/ask の核心部分であるRAGチェーンの実装です。

1. リトリーバー設定（app.py 冒頭）

from langchain_weaviate import WeaviateVectorStore

# ベクトルストアの初期化
vector_store = WeaviateVectorStore(
    client=client,
    index_name=os.getenv("WEAVIATE_INDEX_NAME", "DefaultIndex"),
    text_key="text",
    embedding=embeddings  # HuggingFaceEmbeddings
)

# リトリーバー：類似度上位3件を取得
retriever = vector_store.as_retriever(search_kwargs={"k": 3})

2. 言語検出 & プロンプト切り替え

import re

def detect_language(text: str) -> str:
    """日本語文字（ひらがな・カタカナ・漢字）が含まれれば 'ja'、それ以外は 'en'"""
    if re.search(r'[\u3040-\u309F\u30A0-\u30FF\u4E00-\u9FFF]', text):
        return 'ja'
    return 'en'

def get_prompt_template(language: str) -> str:
    templates = {
        'ja': """以下の文脈に基づいて質問に日本語で答えてください:
{context}

質問: {question}

回答は日本語で、明確かつ簡潔にお願いします。""",
        'en': """Answer the question based on the following context:
{context}

Question: {question}

Please provide a clear and concise answer in English."""
    }
    return templates.get(language, templates['en'])

3. RAGチェーン構築（LangChain式）

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

def get_rag_chain(question: str):
    language = detect_language(question)
    template = get_prompt_template(language)
    prompt = ChatPromptTemplate.from_template(template)

    # LangChainのパイプライン（| で繋ぐ）
    return (
        {"context": retriever, "question": RunnablePassthrough()}
        | prompt
        | get_llm()      # Groq or OpenAI
        | StrOutputParser()
    )

パイプラインの解説：

段階	処理	出力
{"context": retriever, ...}	質問文をWeaviateで検索	関連ドキュメント3件
prompt	文脈+質問をプロンプトに埋め込む	完成したプロンプト文字列
get_llm()	LLMに送信して回答生成	生のモデル出力
StrOutputParser()	文字列に整形	最終回答テキスト

4. /ask エンドポイント実装

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class QueryRequest(BaseModel):
    question: str

@app.post("/ask")
async def ask_question(request: QueryRequest):
    try:
        rag_chain = get_rag_chain(request.question)
        response = rag_chain.invoke(request.question)
        return {"answer": response}
    except Exception as e:
        return {"error": str(e)}

---

✂️ チャンク分割の具体的な戦略

ドキュメント登録時のチャンク分割は、RAGの回答精度に直結します。freeAiChatではファイル形式ごとに異なる戦略を採用しています。

PDF：文単位分割 + オーバーラップ

PDFは**文の区切り（。．.!?）を尊重して分割し、隣接チャンク間にオーバーラップ（重複）**を設けます。

def split_into_chunks_pdf(text: str, chunk_size: int, overlap: int = 100) -> List[str]:
    # 1. 文単位に分割
    sentences = split_into_sentences(text)

    chunks = []
    current_chunk = ""

    for sentence in sentences:
        if len(current_chunk) + len(sentence) <= chunk_size:
            current_chunk += sentence + " "
        else:
            if current_chunk:
                chunks.append(current_chunk.strip())
            current_chunk = sentence + " "

    # 2. オーバーラップチャンクを生成（文脈の連続性を保つ）
    if len(chunks) > 1 and overlap > 0:
        overlapped_chunks = []
        for i in range(len(chunks) - 1):
            overlap_text = chunks[i][max(0, len(chunks[i]) - overlap):]
            next_overlap = chunks[i + 1][:overlap]
            overlapped_chunks.append(f"{overlap_text} {next_overlap}".strip())
        chunks.extend(overlapped_chunks)

    # 3. 重複除去
    seen = set()
    unique_chunks = [c for c in chunks if not (c in seen or seen.add(c))]

    return unique_chunks

なぜPDFは文単位か？

• PDFは論理的な文の区切りが重要（1文が途中で切れると意味が変わる）
• オーバーラップにより、文脈が跨る情報も漏れなく拾える

---

TXT / URL：スライディングウィンドウ

TXTやWebページは固定長のスライディングウィンドウで分割します。

def split_into_chunks_txt(text: str, chunk_size: int, overlap: int = 100) -> List[str]:
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunks.append(text[start:end])
        start = end - overlap  # 次の開始位置をオーバーラップ分戻す
    return chunks

なぜTXT/URLはスライディングウィンドウか？

• 構造化されていないテキストは文の区切りが不鮮明なことが多い
• 固定長で均一に分割し、検索時のカバレッジを確保する

---

前処理の違い

対象	前処理内容
PDF	NFKC正規化（全角半角統一）→ 空白正規化 → 文末スペース追加
TXT	不要な空白・改行を正規化 → 非表示文字除去
URL	TXTと同様（HTMLタグ除去後）

# PDF用前処理
def preprocess_text_pdf(text: str) -> str:
    text = unicodedata.normalize("NFKC", text)  # 全角半角統一
    text = re.sub(r'\s+', ' ', text)
    text = re.sub(r'([。．.!?])([^\s])', r'\1 \2', text)  # 文末スペース追加
    return text.strip()

# TXT用前処理
def preprocess_text_txt(text: str) -> str:
    text = re.sub(r'\s+', ' ', text).strip()
    text = ''.join(c for c in text if c.isprintable() or c.isspace())
    return text

---

🔧 LLM切り替え機構

環境変数 LLM_PROVIDER で、Groq と OpenAI を切り替えます。

import os
from langchain_groq import ChatGroq
from langchain_openai import ChatOpenAI

def get_llm():
    provider = os.getenv("LLM_PROVIDER", "groq")

    if provider == "openai":
        return ChatOpenAI(
            model="gpt-4-turbo",
            temperature=0.5,
            api_key=os.getenv("OPENAI_API_KEY")
        )
    elif provider == "groq":
        return ChatGroq(
            model="openai/gpt-oss-120b",
            temperature=0.5,
            api_key=os.getenv("GROQ_API_KEY")
        )
    else:
        raise ValueError(f"サポートされていないLLMプロバイダー: {provider}")

.env 設定例：

# Groqを使う場合
LLM_PROVIDER=groq
GROQ_API_KEY=gsk_xxxxxxxx

# OpenAIを使う場合（コメントアウト切り替え）
# LLM_PROVIDER=openai
# OPENAI_API_KEY=sk-xxxxxxxx

⚠️ モデル名に関する注意
Groqのモデルは頻繁に入れ替わります。現状のコードでは openai/gpt-oss-120b がデフォルトですが、予告なく廃止される場合があります。
利用開始前に必ず Groqのモデル一覧 で最新状況を確認し、app.py の get_llm() 内の model パラメータを更新してください。
モデル名が無効な場合、アプリは 「回答を取得できませんでした」 と表示され続けます。

---

🛡️ エラーハンドリング（サンプル範囲内）

freeAiChat は学習用サンプルですが、主要な障害パターンについてはエラーハンドリングを入れています。

1. Weaviate接続失敗時のフォールバック

# app.py 冒頭
try:
    client = weaviate.connect_to_local(host="localhost", port=8080)
    print("既存のWeaviateインスタンスに接続しました")
except Exception as e:
    print(f"既存インスタンスへの接続に失敗しました: {e}")
    try:
        # フォールバック：組み込みモードで自動起動
        client = weaviate.WeaviateClient(
            embedded_options=EmbeddedOptions(
                hostname="localhost",
                port=8090,
                persistence_data_path="./weaviate_data"
            )
        )
        print("Weaviateを組み込みモードで起動しました")
    except Exception as e:
        raise RuntimeError("Weaviateの初期化に完全に失敗しました")

初学者が躓きやすいポイント（Docker未起動）をカバーしています。

---

2. ファイルアップロード時の検証

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...), ...):
    ext = os.path.splitext(file.filename)[1].lower()

    if ext not in [".pdf", ".txt"]:
        raise HTTPException(
            status_code=400,
            detail=f"Unsupported file type: {ext}. Only PDF and TXT are allowed."
        )

---

3. バッチ保存失敗時の個別リトライ

Weaviateへのベクトル保存はバッチ処理で効率化していますが、バッチ全体が失敗した場合は個別にリトライします。

batch_size = min(50, max(10, len(chunks) // 10))
successful_chunks = 0

for i in range(0, len(chunks), batch_size):
    batch = chunks[i:i + batch_size]
    try:
        vector_store.add_texts(batch)
        successful_chunks += len(batch)
    except Exception as e:
        print(f"バッチ {i // batch_size + 1} の保存中にエラー: {e}")
        # 個別リトライ
        for chunk in batch:
            try:
                vector_store.add_texts([chunk])
                successful_chunks += 1
            except Exception as e:
                print(f"チャンクの保存に失敗: {e}")

結果としてレスポンスに反映：

{
  "status": "success",
  "message": "42/45個のチャンクを保存しました"
}

---

4. 未対応のエラーパターン（本番化時に検討）

以下は現状未対応で、本番運用や実務利用時に追加が必要です：

パターン	現状	本番化時の対応例
LLMのレート制限超過	例外がそのまま返る	指数バックオフでリトライ
Weaviateの永続化失敗	エラーログのみ	健全性チェック＋アラート
巨大ファイルアップロード	サイズ制限なし	UploadFile にサイズリミット設定
同時アクセス	考慮外	非同期処理の最適化、DB接続プール

---

💡 無料運用のための設定ポイント（再掲）

項目	設定	注意点
Weaviate	DockerでOSS版起動	シングルノード。大規模時は有料版検討
Embedding	all-MiniLM-L6-v2	初回起動時にモデルダウンロード（約80MB）
Groq	無料枠利用	レート制限あり。超過時は一時利用停止
OpenAI	無料クレジット	有効期限・上限あり。超過時は課金必須
FastAPI	uvicorn 起動	本番時は gunicorn + uvicorn 推奨

---

🗂️ シリーズ構成

回	内容
第1回	freeAiChat の全体像・コンセプト・動作イメージ
第2回（本記事）	ai-chat-backend の詳細（RAG処理、API仕様、Weaviate/LangChain構成、チャンク分割、エラーハンドリング）
第3回（予定）	ai-chat-frontend の詳細（UI構成、API連携、導入手順）

---

📚 ソースコード

今回ご紹介したコードの全文は、GitHubで公開しています。

👉 GitHub: https://github.com/8alfalfa8/freeAiChat

本記事がお役に立ちましたら、いいね❤️ や GitHub Star⭐ をいただけると励みになります！

---

• 📂 目次：【AIチャット（freeAiChat）無料構築】連載の全記事まとめ
• 【第1回】無料で構築できるRAG × マルチLLM対応AIチャットシステムの全体像を紹介
• 【第2回】FastAPI × LangChain × Weaviate で構築された RAG対応バックエンド（閲覧中）
• 【【第3回】React × Next.js × shadcn/ui で構築されたチャットフロントエンドを詳細解説
• 💡 今後も開発効率化・ツール連携に関する記事を随時追加していきます！

著者をフォローしていただくと、次回掲載の通知を受け取れます！

---