MinIOとtool callingで実現する「動的な知識管理」RAGシステム

AIエンジニアを目指しているME_DE_AEです。

今回は「チャット画面からファイルをアップロードできて、削除をお願いしたら自動で忘れるシステムを作りたい」という新たな挑戦をすることになりました。
これが、私とMinIOとtool callingとの出会いでした。

単純なRAGから一歩進んで、「動的に知識を管理し、リアルタイムで追加・削除する」 システムへ。この挑戦を通じて、AIシステムの可能性と奥深さを改めて実感することになりました。

MinIOとは？

MinIOとは、簡単に言うと「ローカルで使えるAWS S3」です。

従来のRAGシステムでは、PDFファイルをローカルに置いて、手動でインデックスを再構築する必要がありました。しかし、MinIOを使うことで以下のような流れで動作するようになります。

1. ユーザーが、チャットからPDFファイルをMinIOへアップロード
2. MinIOがファイルの保存を検知し、Webhookでバックエンドに通知
3. バックエンドが自動でPDFを読み込み、ベクトル化してインデックスに追加
4. AIが新しい知識を即座に活用できるようになる

これにより、ファイルの追加・削除がリアルタイムでAIの知識に反映される 動的なシステムが実現できます。

技術選定

今回の動的RAG実装では、以下の技術スタックを選択しました。

技術	目的・役割
MinIO	ローカルS3互換ストレージ
LangChain Tools	AI用ツール定義
FastAPI Webhook	リアルタイムイベント処理
Docker	環境の統一化
Chroma	ベクトルストア(データベース)

MinIOの魅力

特に今回の開発で重要な役割を果たしたのがMinIOです。

MinIOの主な特徴

• S3互換API
  • AWS S3と同じAPIを使用できるため、既存のS3クライアントライブラリがそのまま使える
  • 本番環境ではAWS S3に、開発環境ではMinIOに切り替えることが容易
• イベント通知機能
  • ファイルの追加・削除・変更をリアルタイムで検知
  • Webhookやメッセージキューなど、様々な通知方法に対応
• 軽量で高速
  • Dockerコンテナとして簡単に起動
  • ローカル開発環境でも本格的なストレージ機能を体験可能

今回のシステムでは、MinIOのイベント通知機能を使って、ストレージ内のファイルの変更を即座にAIの知識に反映させる仕組みを構築しました。

tool callingの活用

今回の開発で最も驚いたのは、tool callingという技術でした。

tool callingとは、AIが自分で判断して外部のツールやAPIを呼び出せる機能です。例えるとAIに「この道具を場面によって使ってもいいよ」と許可するようなものです。従来のRAGでは「検索→回答生成」という固定の流れでしたが、tool callingを使うことで以下のような動的な処理が可能になります。

# ツールの定義例
@tool
def list_files_in_bucket():
    """バケット内にあるファイルの一覧を問い合わせるために使用します。引数は不要です。"""
    try:
        bucket_name = os.environ.get("S3_BUCKET_NAME")
        s3_client = get_client()
        response = s3_client.list_objects_v2(Bucket=bucket_name)
        
        if "Contents" not in response or not response["Contents"]:
            return "現在、ストレージにファイルはありません。"
        
        files = [obj["Key"] for obj in response["Contents"]]
        return "ストレージ上のファイル一覧:\n- " + "\n- ".join(files)
    except Exception as e:
        return f"ファイル一覧の取得中にエラーが発生しました: {e}"

@tool  
def delete_file_from_bucket(file_name: Annotated[str, "削除したいファイルの名前。"]):
    """バケット内にある特定のファイルを削除するために使用します。削除したいファイル名を引数に指定してください。"""
    try:
        bucket_name = os.environ.get("S3_BUCKET_NAME")
        s3_client = get_client()
        
        # ファイルの存在確認
        s3_client.head_object(Bucket=bucket_name, Key=file_name)
        # ファイル削除実行
        s3_client.delete_object(Bucket=bucket_name, Key=file_name)
        
        return f"ファイル '{file_name}' をストレージから正常に削除しました。"
    except ClientError as e:
        if e.response["Error"]["Code"] == "404":
            return f"削除対象のファイル '{file_name}' はストレージに存在しません。"
        else:
            return f"ファイルの削除処理中にエラーが発生しました: {e}"

AIは、ユーザーの質問に応じて「ファイル一覧を確認する」「特定のファイルを削除する」「ドキュメントを検索する」といったツールを自分で選択して実行します。

MinIOのWebhook機能による自動化

今回のシステムでもう一つの重要な技術がWebhookです。Webhookとは、特定のイベントが発生した際に、自動的に指定されたURLにHTTPリクエストを送信する仕組みです。

MinIOのWebhook機能を使うことで、以下のような自動化が実現できます。

# FastAPIのWebhookエンドポイント
@app.post("/webhook")
async def minio_webhook(request: Request, background_tasks: BackgroundTasks):
    try:
        data = await request.json()
        logging.info(f"Webhook受信: {json.dumps(data, indent=2, ensure_ascii=False)}")
        
        for record in data.get("Records", []):
            event_name = record.get("eventName")
            object_key = unquote_plus(record.get("s3", {}).get("object", {}).get("key"))
            
            if "ObjectCreated" in event_name:
                # ファイルがアップロードされた時の処理
                background_tasks.add_task(add_document_to_index, object_key)
            elif "ObjectRemoved" in event_name:
                # ファイルが削除された時の処理
                background_tasks.add_task(remove_document_from_index, object_key)
                
        return {"status": "ok"}
    except Exception as e:
        logging.error(f"Webhook処理中にエラー: {e}", exc_info=True)
        return JSONResponse(status_code=500, content={"message": "Webhook error"})

Webhookの動作フロー

1. ファイルアップロード時

   1. ユーザーがフロントエンドからPDFファイルをアップロード
   2. MinIOがファイルを保存し、ObjectCreatedイベントを検知
   3. MinIOがWebhook設定に従い/webhookエンドポイントにPOSTリクエスト送信
   4. バックエンドがadd_document_to_indexをバックグラウンドタスクで実行
   5. PDFが読み込まれ、ベクトル化されてChromaに追加

2. ファイル削除時

   1. AIがdelete_file_from_bucketツールを実行してファイル削除
   2. MinIOがObjectRemovedイベントを検知
   3. MinIOが自動で/webhookエンドポイントにPOSTリクエスト送信
   4. バックエンドがremove_document_from_indexをバックグラウンドタスクで実行
   5. 該当するベクトルデータがChromaから削除

この仕組みにより、ファイルの変更がリアルタイムでAIの知識に反映される動的なシステムが実現できています。

動的RAGシステムの技術構成と処理フロー

実装したシステムは、以下のような流れで動作します。

シーケンス図で表すと、以下のようになります。MinIOのイベント通知とtool callingが連携して、リアルタイムな知識管理を実現しています。

MinIOの設定とWebhook連携

MinIOの最大の特徴は、ファイルの変更をリアルタイムで検知できることです。Docker Composeでの設定例は以下のとおりです。

docker-compose.yml

minio:
  image: minio/minio:RELEASE.2025-05-24T17-08-30Z
  environment:
    MINIO_NOTIFY_WEBHOOK_ENABLE_RAGHOOK: "on"
    MINIO_NOTIFY_WEBHOOK_ENDPOINT_RAGHOOK: "http://backend:8000/webhook"
  command: server /data --console-address ":9001"

webhook関連の環境変数を設定することにより、MinIOはファイルの追加・削除を検知すると、自動でバックエンドの/webhookエンドポイントに通知を送信します。

フロントエンドのファイルアップロード機能

UploadModalによるユーザー体験向上

アップロード処理中は、専用のモーダルでプログレスを表示し、ユーザーに分かりやすいフィードバックを提供します。

以下のような直感的なユーザー体験を実現できました。

1. ファイル選択: クリップアイコンをクリックしてPDFファイルを選択
2. バリデーション: PDF以外のファイルは自動的に拒否
3. アップロード実行: 送信ボタンでアップロード開始
4. プログレス表示: モーダルでアップロード状況をリアルタイム表示
5. 結果通知: 成功・失敗に応じた適切なメッセージ表示

tool callingの実装

LangChainのtool calling機能を使うことで、AIが動的にツールを選択・実行できるようになります。

# ツールの定義
tools = [list_files_in_bucket, delete_file_from_bucket]

# ベクトルストアにドキュメントがある場合、検索ツールも追加
if vectorstore._collection.count() > 0:
    retriever = vectorstore.as_retriever()
    retriever_tool = create_retriever_tool(
        retriever,
        "document_retriever", 
        "ドキュメントの内容に関する質問に答えるために使用します。"
    )
    tools.insert(0, retriever_tool)

# エージェントの構築
agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

AIは、ユーザーの質問を理解し、適切なツールを選択して実行します。例えば

• 「ファイル一覧を教えて」→ list_files_in_bucketを実行
• 「resume.pdfを削除して」→ delete_file_from_bucketを実行
• 「この人の職歴は？」→ document_retrieverを実行

実装で苦労したポイント

tool callingの動作理解

tool callingの概念を理解するのに時間がかかりました。従来のプログラミングでは「Aの条件ならBを実行」という明確な制御フローを書きますが、tool callingでは「AIが状況を判断してツールを選択」するため、予期しない動作が発生することがありました。

解決策として、以下のアプローチを取りました。

1. ツールの説明を詳細に記述

   @tool
   def delete_file_from_bucket(file_name: Annotated[str, "削除したいファイルの名前。"]):
       """バケット内にある特定のファイルを削除するために使用します。削除したいファイル名を引数に指定してください。"""
   

2. プロンプトでの制御

   prompt = ChatPromptTemplate.from_messages([
       ("system", "あなたは優秀なAIアシスタントです。まず、提供されたコンテキスト情報を確認し、質問に答えられる情報があるか探してください。必要に応じてツールを使用し、ユーザーの質問に丁寧かつ正確に答えてください。"),
       ("user", "{input}"),
       MessagesPlaceholder(variable_name="agent_scratchpad"),
   ])
   

「AIが自分で判断する」という新たな体験

この開発を通じて最も驚いたのは、AIが状況に応じて自分でツールを選択・実行する という体験でした。

従来のシステムでは

• ユーザーが「ファイル一覧」をクリック → システムがファイル一覧を表示
• ユーザーが「削除」ボタンをクリック → システムがファイルを削除

しかし、tool callingを使ったシステムでは

• ユーザーが「ファイル一覧を教えて」と質問 → AIがlist_files_in_bucketを実行
• ユーザーが「resume.pdfを削除して」と依頼 → AIがdelete_file_from_bucketを実行

AIが自然言語を理解し、適切なツールを選択して実行する という、まさに「AIアシスタント」と呼ぶにふさわしい動作を実現できました。

完成の瞬間

数々の試行錯誤を経て、ついに動的RAGシステムが期待通りに動作しました。

実際に「resume.pdfを削除して」という依頼を試してみると、AIは以下の手順で処理を実行しました。

1. ファイル削除ツールを選択
2. MinIOからファイルを削除
3. Webhookでイベントを受信
4. ベクトルストアから該当データを削除
5. 「ファイルを削除し、AIの知識からも削除しました」と報告

この一連の流れが自動で実行された時の感動は、前回のRAGシステムとは全く違うものでした。

「AIがファイル操作を理解し、知識管理まで自動で行っている！」

この体験を通じて、tool callingという技術の可能性を肌で感じることができました。単なる質問応答ではなく、実際の作業を理解し、適切なツールを使って実行する 真のAIアシスタントが実現できたのです。

学んだこと

今回の動的RAG実装を通じて、AIシステムの新たな可能性について学んだことは数多くありますが、特に印象深いのは以下の4点です。

• フロントエンドとバックエンドの統合設計

  • 単純なAPI呼び出しではなく、ファイルアップロード、状態管理、エラーハンドリングまでを含めた総合的なUX設計
  • React + TypeScriptによるタイプセーフな開発の重要性

• リアルタイム性の重要性

  • ファイルの変更が即座にAIの知識に反映されることで、ユーザー体験が劇的に向上
  • Webhookとイベント駆動アーキテクチャの威力を実感

• tool callingの革新的な可能性

  • AIが自然言語を理解し、適切なツールを選択・実行する能力
  • 従来の固定フローから、動的で柔軟な処理への進化

• システム全体の設計思想

  • 単一の技術ではなく、フロントエンド、ストレージ、イベント通知、AI、ツール実行までを含めた総合的な設計
  • 各コンポーネントの連携が全体の価値を生み出す

まとめ

MinIOとtool callingの実装を通じて、AIエンジニアとしての視野がさらに大きく広がりました。前回のRAGシステムから、今度は「動的に知識を管理し、実際の作業を実行するAI」へ。技術的な挑戦はより複雑になりましたが、その分得られる学びと達成感も大きくなりました。

特に、「AIが自分で判断してツールを実行する」 という新しい体験に触れることができたのは、大きな収穫でした。これは単なる技術の進歩ではなく、AIと人間の協働の新たな形を示していると感じています。

異業種からAIエンジニアへの道のりは決して平坦ではありませんが、一歩ずつ確実に前進していることを実感しています。「AIという武器を手に、未経験分野でのモノづくり」 という目標に向かって、引き続き学習を続けていきます。