ChromaDB + Ollama で作る高機能ローカルRAGシステム

Posted at 2025-07-14

ChromaDB + Ollama で作る高機能ローカルRAGシステム

はじめに

この記事は、Claudeによって作成されました。
このコードの大部分はGeminiによって生成されたもので、筆者はその動作検証と最終的な調整を担当しました。
この記事では、ChromaDBとOllamaを組み合わせて、完全にローカル環境で動作するRAG（Retrieval-Augmented Generation）システムを構築する方法をご紹介します。
このシステムは、ご自身のPC上でサーバーを起動して利用するものです

特徴

🔍 高性能ベクトル検索: ChromaDBによる高速な類似度検索
🤖 ローカルLLM: Ollamaで複数のモデルをサポート
📄 文書管理: 複数ファイルのアップロードと管理
🌐 Webインターフェース: 直感的なチャットUI
🔧 設定可能: 埋め込みモデル、検索件数、コレクション管理

必要な環境

前提条件

Python 3.8以上
Node.js（開発時のみ）
十分なメモリ（4GB以上推奨）

必要なサービス

ChromaDB サーバー (ポート: 8001)
Ollama サーバー (ポート: 11434)
Pythonバックエンド (ポート: 5000)

インストール手順

1. ChromaDBのセットアップ

# ChromaDBのインストール
pip install chromadb

# ChromaDBサーバーの起動
chroma run --host 0.0.0.0 --port 8001 --path ./chroma_db

2. Ollamaのセットアップ

# Ollamaのインストール（公式サイトから）
# https://ollama.ai/

# モデルのダウンロード
ollama pull llama3.2:latest
ollama pull llama3.2:3b
ollama pull llama3.2:1b

# Ollamaサーバーの起動
ollama serve

3. Pythonバックエンドのセットアップ

# 必要なパッケージのインストール
pip install flask flask-cors sentence-transformers chromadb requests

システム構成

バックエンドサーバー (Python + Flask)

Faiss.py

import os
# [WinError 1114] PyTorchのDLL初期化エラーを回避するための設定
os.environ['KMP_DUPLICATE_LIB_OK']='TRUE'

import json
import requests
import chromadb
import traceback
from flask import Flask, request, jsonify, Response
from flask_cors import CORS
from sentence_transformers import SentenceTransformer

# --------------------------------------------------------------------------
# 1. Flaskアプリケーションの初期設定
# --------------------------------------------------------------------------
app = Flask(__name__)
CORS(app) # CORSをapp定義の直後に適用

# Sentence Transformerモデルのキャッシュディレクトリ設定
os.environ['SENTENCE_TRANSFORMERS_HOME'] = './st_models'

# ChromaDBクライアントの初期化
try:
    db_client = chromadb.PersistentClient(path="./chroma_db")
except Exception as e:
    print(f"Error initializing ChromaDB: {e}")
    db_client = None

# 埋め込みモデルのキャッシュ
embedding_models = {}

def get_embedding_model(model_name="all-MiniLM-L6-v2"):
    """埋め込みモデルをキャッシュから取得またはロードする"""
    if model_name not in embedding_models:
        print(f"Loading embedding model: {model_name}")
        try:
            embedding_models[model_name] = SentenceTransformer(model_name)
        except Exception as e:
            print(f"Error loading embedding model {model_name}: {e}")
            return None
    return embedding_models[model_name]

# --------------------------------------------------------------------------
# 2. ステータス確認用API
# --------------------------------------------------------------------------
@app.route('/health', methods=['GET'])
def health_check():
    return jsonify({"status": "ok"}), 200

@app.route('/chromadb/status', methods=['GET'])
def chromadb_status():
    if not db_client: return jsonify({"status": "error", "detail": "ChromaDB client not initialized"}), 500
    try:
        db_client.heartbeat()
        return jsonify({"status": "ok"}), 200
    except Exception as e:
        return jsonify({"status": "error", "detail": str(e)}), 500

@app.route('/ollama/status', methods=['GET'])
def ollama_status():
    """Ollamaサーバーの接続状況とモデルリストを確認"""
    try:
        response = requests.get('http://127.0.0.1:11434/api/tags', timeout=5)
        response.raise_for_status()
        models_data = response.json().get('models', [])
        return jsonify({"connected": True, "models": models_data}), 200
    except requests.exceptions.RequestException as e:
        return jsonify({"connected": False, "detail": "Failed to connect to Ollama server."}), 500
    except Exception as e:
        return jsonify({"connected": False, "detail": str(e)}), 500

# --------------------------------------------------------------------------
# 3. コレクション管理API
# --------------------------------------------------------------------------
@app.route('/collections', methods=['GET', 'OPTIONS'])
def list_collections():
    if request.method == 'OPTIONS': return Response(status=200)
    if not db_client: return jsonify({"detail": "ChromaDB not available"}), 500
    collections = db_client.list_collections()
    return jsonify([{"name": c.name} for c in collections]), 200

@app.route('/collections', methods=['POST'])
def create_collection():
    if not db_client: return jsonify({"detail": "ChromaDB not available"}), 500
    data = request.get_json()
    name = data.get('name')
    if not name: return jsonify({"detail": "Collection name is required"}), 400
    try:
        db_client.get_or_create_collection(name=name)
        return jsonify({"message": f"Collection '{name}' created successfully."}), 201
    except Exception as e:
        return jsonify({"detail": str(e)}), 500

@app.route('/collections/<string:collection_name>', methods=['DELETE', 'OPTIONS'])
def delete_collection(collection_name):
    if request.method == 'OPTIONS': return Response(status=200)
    if not db_client: return jsonify({"detail": "ChromaDB not available"}), 500
    try:
        db_client.delete_collection(name=collection_name)
        return jsonify({"message": f"Collection '{collection_name}' deleted."}), 200
    except Exception as e:
        return jsonify({"detail": str(e)}), 500

@app.route('/collections/<string:collection_name>/documents', methods=['GET', 'OPTIONS'])
def get_documents_in_collection(collection_name):
    if request.method == 'OPTIONS': return Response(status=200)
    if not db_client: return jsonify({"detail": "ChromaDB not available"}), 500
    try:
        collection = db_client.get_collection(name=collection_name)
        data = collection.get(include=['metadatas'])
        
        documents = []
        if data and data['ids']:
            unique_filenames = set()
            for metadata in data['metadatas']:
                filename = metadata.get('filename')
                if filename and filename not in unique_filenames:
                    documents.append({"id": filename})
                    unique_filenames.add(filename)
        
        return jsonify({"documents": documents, "count": collection.count()}), 200
    except ValueError:
        return jsonify({"documents": [], "count": 0, "detail": "Collection not found"}), 200
    except Exception as e:
        return jsonify({"detail": str(e)}), 500

# --------------------------------------------------------------------------
# 4. 文書アップロードとチャットAPI
# --------------------------------------------------------------------------
@app.route('/upload', methods=['POST', 'OPTIONS'])
def upload_file():
    if request.method == 'OPTIONS': return Response(status=200)
    if not db_client: return jsonify({"detail": "ChromaDB not available"}), 500
    if 'file' not in request.files: return jsonify({"detail": "No file part"}), 400
    
    file = request.files['file']
    collection_name = request.form.get('collection_name', 'default')
    
    if file.filename == '': return jsonify({"detail": "No selected file"}), 400

    try:
        text = file.read().decode('utf-8')
        chunk_size = 500
        chunks = [text[i:i+chunk_size] for i in range(0, len(text), chunk_size)]
        
        collection = db_client.get_or_create_collection(name=collection_name)
        model = get_embedding_model()
        if not model: return jsonify({"detail": "Embedding model not available"}), 500

        embeddings = model.encode(chunks).tolist()
        ids = [f"{file.filename}:{i}" for i in range(len(chunks))]
        metadatas = [{"filename": file.filename, "chunk": i} for i in range(len(chunks))]

        collection.add(embeddings=embeddings, documents=chunks, metadatas=metadatas, ids=ids)
        
        return jsonify({"message": "File processed successfully", "chunks": len(chunks)}), 200
    except Exception as e:
        return jsonify({"detail": str(e)}), 500

@app.route('/chat', methods=['POST', 'OPTIONS'])
def chat():
    if request.method == 'OPTIONS': return Response(status=200)
    
    if not db_client: return jsonify({"detail": "ChromaDB not available"}), 500

    data = request.get_json()
    query = data.get('query')
    model_name = data.get('model', 'llama3.2:latest')
    collection_name = data.get('collection', 'default')
    embedding_model_name = data.get('embedding_model', 'all-MiniLM-L6-v2')
    top_k = data.get('top_k', 3)

    if not query: return jsonify({"detail": "Query is required"}), 400

    def generate_response():
        try:
            embedding_model = get_embedding_model(embedding_model_name)
            if not embedding_model:
                yield f"data: {json.dumps({'error': 'Embedding model not available'})}\n\n"
                return

            query_embedding = embedding_model.encode([query]).tolist()
            
            try:
                collection = db_client.get_collection(name=collection_name)
                results = collection.query(query_embeddings=query_embedding, n_results=top_k)
                context = "\n---\n".join(results['documents'][0])
                yield f"data: {json.dumps({'search_results': True})}\n\n"
            except Exception:
                context = "利用可能な参考情報はありません。"

            prompt = f"参考情報に厳密に基づいて、ユーザーの質問に日本語で回答してください。参考情報に答えがない場合は、「情報が見つかりません」と回答してください。\n\n### 参考情報:\n{context}\n\n### ユーザーの質問:\n{query}"
            
            payload = {
                "model": model_name,
                "messages": [{'role': 'user', 'content': prompt}],
                "stream": True
            }
            
            with requests.post('http://127.0.0.1:11434/api/chat', json=payload, stream=True) as r:
                r.raise_for_status()
                for line in r.iter_lines():
                    if line:
                        decoded_line = json.loads(line.decode('utf-8'))
                        if 'content' in decoded_line.get('message', {}):
                            content_data = {"content": decoded_line['message']['content']}
                            yield f"data: {json.dumps(content_data)}\n\n"
        
        except Exception as e:
            traceback.print_exc()
            yield f"data: {json.dumps({'error': str(e)})}\n\n"

    return Response(generate_response(), mimetype='text/event-stream')

if __name__ == '__main__':
    get_embedding_model()
    app.run(host='0.0.0.0', port=5000, debug=True, use_reloader=False)

フロントエンド（HTML + JavaScript）

ソースコードはGitHubリポジトリ内の RAG3.html をご参照ください

フロントエンドは、モダンなWebUIを提供するHTMLファイルで構成されています。主な機能：

リアルタイムチャット: ストリーミング応答対応
文書管理: ファイルアップロード、コレクション管理
設定パネル: モデル選択、検索パラメータ調整
接続状況監視: 各サービスの状態をリアルタイム表示

使用方法

1. サーバー起動

# ターミナル1: ChromaDBサーバー
chroma run --host 0.0.0.0 --port 8001 --path ./chroma_db

# ターミナル2: Ollamaサーバー
ollama serve

# ターミナル3: Pythonバックエンド
python Faiss.py

2. Webインターフェースの利用

ブラウザで http://localhost:5000 にアクセス
接続状況を確認（ChromaDB、Ollama共に緑色になることを確認）
新しいコレクションを作成
文書をアップロード（.txt, .md, .pdf, .docx対応）
チャットで質問を開始

3. 設定のカスタマイズ

LLMモデル: Llama 3.2, Gemma, Mistralなど
埋め込みモデル: all-MiniLM-L6-v2, all-mpnet-base-v2など
検索件数: 3, 5, 10件から選択
コレクション: 複数のコレクションで文書を分類管理

技術的な特徴

RAGパイプライン

文書の前処理: アップロードされた文書を500文字ごとにチャンク分割
ベクトル化: Sentence Transformersで埋め込みベクトルを生成
インデックス保存: ChromaDBに永続化して保存
検索: ユーザーの質問をベクトル化し、類似度検索を実行
生成: 検索結果を文脈としてOllamaのLLMで回答を生成

パフォーマンス最適化

モデルキャッシュ: 埋め込みモデルをメモリ上でキャッシュ
ストリーミング応答: Server-Sent Eventsでリアルタイム応答
永続化: ChromaDBによる高速な永続化ストレージ

トラブルシューティング

よくある問題と解決方法

ChromaDB接続エラー

# ポート8001が使用中でないか確認
netstat -ano | findstr :8001

Ollama接続エラー

# Ollamaサービスの状態確認
ollama list

PyTorchエラー

# 環境変数設定
os.environ['KMP_DUPLICATE_LIB_OK']='TRUE'

拡張可能性

今後の改良点

PDFパーサー: より高度な文書解析
マルチモーダル: 画像や音声への対応
スケーラビリティ: 複数ユーザー対応
認証機能: セキュリティ機能の追加

まとめ

このシステムは、完全にローカル環境で動作するRAGシステムとして、以下の利点を提供します：

プライバシー: データが外部に送信されない
カスタマイズ性: 各コンポーネントを自由に調整可能
コスト効率: APIコストが発生しない
オフライン対応: インターネット接続不要

企業内での文書検索システムや、個人の知識管理システムとして活用できます。
ソースコードはこちら (GitHub):https://github.com/ishikawamasahito/RAG3.html

参考資料

この記事が参考になりましたら、ぜひ⭐をお願いします！

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up