Vol. 1「Ollamaで動かす『フォルダ読みRAG』」では、ローカルPC上でRAGを簡単に体験することができました。
sci-gen.hatenablog.com
しかし、Vol. 1 のスクリプトには「起動するたびに、すべてのドキュメントをゼロから再処理する」という大きな課題がありました。ドキュメントが数十個になると、起動までに数分〜数十分かかってしまい、実用的ではありません。
このチュートリアル(Vol. 2)では、この課題を解決します。ベクトルDBを「永続化」させ、フォルダ内の変更(追加・削除)だけを検知してDBに反映させる「差分更新」が可能な、より高速で実用的なRAGエージェントを構築します。
達成目標:
- ベクトルDBを永続化させ、2回目以降の起動を高速化する。
- Agno の「Upsert(重複スキップ)」機能を利用した差分追加を実装する。
- フォルダから削除されたファイルを検知し、ベクトルDBからも削除するロジックを実装する。
- チームの共有フォルダでの運用にも耐えうる、効率的なDB更新プロセスを学ぶ。
想定読者:
- Vol. 1 を完了し、次のステップに進みたい方。
- RAG の起動時間を短縮したい方。
- チームでナレッジベースを共有・運用する方法を検討している方。
1. 環境準備と Vol. 1 からの変更点
環境は Vol. 1 と同じです。Ollama と必要なライブラリ (agno, lancedb, pypdf, python-pptx) がインストールされていることを確認してください。
Vol. 1 からの主な変更点
- DBを削除しない:
shutil.rmtree(VECTOR_DB_PATH) を削除します。これによりDBが永続化されます。
- 差分同期ロジックの追加: スクリプト起動時に、以下の3つを比較・実行するロジックを追加します。
- 削除検知: フォルダから消されたファイルは、DBからも削除する。
- 追加検知: フォルダに新しく増えたファイルは、DBに追加する。
- 重複スキップ: 既にDBに存在するファイルは、処理をスキップする (Agnoが自動で対応)。
2. 動的RAGエージェントの構築 (Pythonコード)
Vol. 1 で作成した rag_app.py をベースに、差分更新ロジックを追加した rag_app_v2.py を作成します。
from pathlib import Path
from agno.agent import Agent
from agno.models.ollama import Ollama
from agno.knowledge.knowledge import Knowledge
from agno.knowledge.embedder.ollama import OllamaEmbedder
from agno.vectordb.lancedb import LanceDb
DOCS_FOLDER = "./my_docs"
VECTOR_DB_PATH = "./vector_db"
OLLAMA_CHAT_MODEL = "llama3.2"
OLLAMA_EMBED_MODEL = "mxbai-embed-large"
SUPPORTED_EXTENSIONS = [".pdf", ".pptx"]
try:
print(f"ナレッジベース '{VECTOR_DB_PATH}' をセットアップします...")
embedder = OllamaEmbedder(id=OLLAMA_EMBED_MODEL)
vector_db = LanceDb(uri=VECTOR_DB_PATH, embedder=embedder)
knowledge_base = Knowledge(vector_db=vector_db)
print("ナレッジベースのセットアップ完了。")
except Exception as e:
print(f"\nエラーが発生しました: {e}")
print("Ollamaサーバーが正しく起動しているか確認してください。")
exit()
print("フォルダとナレッジベースの同期を開始します...")
try:
db_contents = knowledge_base.get_contents()
db_file_paths = {content.path for content in db_contents}
print(f" [DB] {len(db_file_paths)} 件のドキュメントが登録済みです。")
doc_path = Path(DOCS_FOLDER)
if not doc_path.exists():
print(f"エラー: ドキュメントフォルダ '{DOCS_FOLDER}' が見つかりません。")
exit()
local_file_paths = set()
for ext in SUPPORTED_EXTENSIONS:
for file in doc_path.rglob(f"*{ext}"):
local_file_paths.add(str(file.resolve()))
print(f" [フォルダ] {len(local_file_paths)} 件のドキュメントが見つかりました。")
files_to_delete = db_file_paths - local_file_paths
if files_to_delete:
print(f"\n--- 削除処理({len(files_to_delete)} 件)---")
for file_path in files_to_delete:
if knowledge_base.delete_content(path=file_path):
print(f" [削除] {os.path.basename(file_path)}")
else:
print(f" [削除失敗] {os.path.basename(file_path)}")
files_to_add = local_file_paths - db_file_paths
if files_to_add:
print(f"\n--- 追加処理({len(files_to_add)} 件)---")
for file_path in files_to_add:
print(f" [追加] {os.path.basename(file_path)} を処理中...")
knowledge_base.add_content_sync(
path=file_path,
chunk=True,
chunk_size=1000,
)
if not files_to_delete and not files_to_add:
print(" [同期] 変更はありません。ナレッジベースは最新です。")
print("同期が完了しました。")
except Exception as e:
print(f"\n同期中にエラーが発生しました: {e}")
exit()
llm = Ollama(id=OLLAMA_CHAT_MODEL)
agent = Agent(
model=llm,
knowledge=knowledge_base,
search_knowledge=True,
description="あなたは提供されたドキュメントに関する質問に答える専門家です。",
instructions=[
"回答は、必ずナレッジベースから検索した情報(コンテキスト)に基づいてください。",
"情報が見つからない場合は、その旨を正直に伝えてください。"
]
)
print("\n--- RAGエージェント起動 ---")
print("ナレッジベースに基づいて回答します。")
print("質問を入力してください (終了は 'q' または 'exit'):")
try:
while True:
query = input("\n[ 質問 ]: ")
if query.lower() in ["q", "exit"]:
print("終了します。")
break
print("[ 回答 ]:")
for chunk in agent.run(query, stream=True):
print(chunk, end="", flush=True)
print("\n")
except KeyboardInterrupt:
print("\n終了します。")
3. コードの詳細解説
Vol. 2 の核心である「セクション 3: 差分同期ロジック」を詳しく見ていきます。
3-1. DB内のファイル一覧取得
db_contents = knowledge_base.get_contents()
db_file_paths = {content.path for content in db_contents}
knowledge_base.get_contents() を使うと、Agno のナレッジベースに登録されているコンテンツのメタデータ(パス、チャンク数など)を取得できます。ここでは、比較のためにファイルパス(content.path)だけを set(集合)に格納しています。
3-2. ローカルフォルダ内のファイル一覧取得
local_file_paths = set()
for file in doc_path.rglob(f"*{ext}"):
local_file_paths.add(str(file.resolve()))
Vol. 1 と似ていますが、pathlib で取得したパスを .resolve() で絶対パスに変換し、str() で文字列化しています。
db_file_paths と local_file_paths の形式を確実に一致させるための重要な処理です。(相対パスのままだと、実行場所によってパスが変わり、比較が失敗するため)
3-3 & 3-4. 集合演算による差分検知
files_to_delete = db_file_paths - local_file_paths
files_to_add = local_file_paths - db_file_paths
ここが差分検知のキモです。Python の set(集合)の演算(引き算)を使っています。
db_file_paths - local_file_paths: DBセットに「あって」、ローカルセットに「ない」もの = 削除されたファイル
local_file_paths - db_file_paths: ローカルセットに「あって」、DBセットに「ない」もの = 新しく追加されたファイル
set を使うことで、数百・数千のファイルがあっても一瞬で差分を計算できます。
3-3. 削除処理
knowledge_base.delete_content(path=file_path)
Agno の delete_content() を呼び出し、ファイルパスを指定してDBから該当するベクトルデータを削除します。
3-4. 追加処理 (Upsert)
knowledge_base.add_content_sync(path=file_path)
新しく追加されたファイルは、Vol. 1 と同じ add_content_sync で処理します。
add_content_sync は Upsert (Update + Insert) として機能するため、もし万が一、重複したファイルを追加しようとしても、Agno が自動でスキップしてくれるため安全です。
4. 実行と動作確認
rag_app_v2.py を実行して、差分更新の動作を確認してみましょう。
1回目: 初回実行
./vector_db フォルダがない状態で実行します。
python rag_app_v2.py
実行ログ(例):
ナレッジベース './vector_db' をセットアップします...
Embeddingモデルを初期化中...
ナレッジベースのセットアップ完了。
フォルダとナレッジベースの同期を開始します...
[DB] 0 件のドキュメントが登録済みです。
[フォルダ] 2 件のドキュメントが見つかりました。
--- 追加処理(2 件)---
[追加] 業務マニュアル.pdf を処理中...
[追加] 新製品プレゼン資料.pptx を処理中...
同期が完了しました。
--- RAGエージェント起動 ---
...
初回はDBが空(0件)なので、フォルダ内の全ファイルが「追加処理」されます。この処理は時間がかかります。
2回目: 変更なしで再実行
my_docs フォルダを何も変更せず、もう一度スクリプトを実行します。
実行ログ(例):
ナレッジベース './vector_db' をセットアップします...
Embeddingモデルを初期化中...
ナレッジベースのセットアップ完了。
フォルダとナレッジベースの同期を開始します...
[DB] 2 件のドキュメントが登録済みです。
[フォルダ] 2 件のドキュメントが見つかりました。
[同期] 変更はありません。ナレッジベースは最新です。
同期が完了しました。
--- RAGエージェント起動 ---
...
DB(2件)とフォルダ(2件)に差分がないため、追加・削除処理がスキップされ、一瞬で起動します。
3回目: ファイルを追加して実行
my_docs フォルダに、新しいファイル 企画書.pdf を追加してから実行します。
実行ログ(例):
...
[DB] 2 件のドキュメントが登録済みです。
[フォルダ] 3 件のドキュメントが見つかりました。
--- 追加処理(1 件)---
[追加] 企画書.pdf を処理中...
同期が完了しました。
...
企画書.pdf(1件)だけが「追加処理」され、既存の2件はスキップされます。
4回目: ファイルを削除して実行
my_docs フォルダから 業務マニュアル.pdf を削除してから実行します。
実行ログ(例):
...
[DB] 3 件のドキュメントが登録済みです。
[フォルダ] 2 件のドキュメントが見つかりました。
--- 削除処理(1 件)---
[削除] 業務マニュアル.pdf
同期が完了しました。
...
業務マニュアル.pdf(1件)だけが「削除処理」されます。
5. 次のステップ
これで、起動が高速で、ドキュメントの変更にも自動で追従できる、実用的なRAGエージェントが完成しました。
このスクリプトは、チームの共有フォルダ(ネットワークドライブ)で運用することも可能です。
しかし、チームで使うにはまだ課題があります。
- 「誰かがDB更新処理(スクリプト実行)をしないと、他のメンバーは古い情報を見てしまう」
- 「もし2人が同時に更新処理を実行したらどうなる?」(=同時書き込み問題)
この問題を解決するには、
- 「DB更新申請APIサーバー」を構築し、RAGエージェントを「読み取り専用」と「書き込み管理」に分離する
- 常に、フォルダを巡回して新しいファイルを読み込む
ような実装が求められますが、それはまたの機会ということで。