💽
FastMCPサーバとQdrantを使ったRAGシステムについて

エイン
2025/06/25に公開
 背景ベクトル埋め込みと検索のプロトタイプを基にし、様々なベクトルDBのパフォーマンス、導入の容易さ、保守性などの機能を実際に比較するために、まずQdrantベクトルDBを使ったプロトタイプを実装しました。

 対象読者ベクトルの基本概念を理解している方
RAG導入を検討している方

 環境

項目
バージョン


OS
Ubuntu 20.04.6 LTS

ランタイム
Python 3.10.12

主要ライブラリ
fastapi 0.115.12、fastmcp 2.8.0、langchain 0.3.25、numpy 2.2.6、openai 1.86.0、python-dotenv 1.1.0、qdrant_client 1.14.2、requests 2.32.4、tiktoken 0.9.0

モデル
Azure OpenAI text-embedding-3-small

参照：Azure OpenAIテキスト埋め込みモデル

 事前準備「ベクトル埋め込みと検索のプロトタイプ」の事前準備で用意したCopilotに生成してもらった10件のmarkdownファイルを今度はmongoDBに登録しました。

 プロジェクトの全体像現在、MCPサーバ内で、検索とデータ操作の役割に応じて、ツールとリソースとして2つに分けられています。


 プロジェクト構造今度のプロジェクト構造は、「ベクトル埋め込みと検索のプロトタイプ」の構造とほぼ同じですが、追加および変更した分のみを記述します。
モデルの定義
mongoDBからのデータ取得API ※
QdrantベクトルDBの操作
セクションの操作
QdrantベクトルDB操作用のMCPサーバ
./
├── model モデルの定義
    ├── section.py #分割されたセクションのモデルクラス
    └── text_embed.py #テキスト埋め込みのモデルクラス
├── .env #環境変数の設定ファイル
├── mongoDB_api.py #mongoDBからのデータ取得API
├── mcp_qdrant_server.py #QdrantベクトルDB操作用のMCPサーバ
├── qdrant.py #QdrantベクトルDBの操作
├── requirements.txt #必須のパッケージ
├── section_handler.py #セクションの操作
├── split_markdown.py #mdファイルの分割
├── split_text.py #チャンクの分割
└── vector_embed.py #ベクトル埋め込み

※ こちらの記事はQdrantベクトルDBの操作を強調するため、mongoDBからのデータ取得APIの内容を省略します。

※ mongoDBからのデータ取得APIには、以下の2つのメソッドが存在しています。
get_page_list → mongoDBのあるパス以下の全ページを取得するメソッド
get_page_data → 各ページデータを取得するメソッド

 手順
 1. pythonパッケージのインストールスクリプトと同じ階層に、依存パッケージを記載した requirements.txt ファイルを用意しました。
requirements.txtファイルの中は以下のようです。
requirements.txt
fastapi
fastmcp
langchain
numpy
openai
python-dotenv
qdrant_client 
requests
下記のコマンドを実行することで、requirements.txtに記載された依存パッケージをインストールできます。
pip install -r requirements.txt
ローカルでインストールされた全てのpythonパッケージをpip listコマンドで確認出来ます。
C:\Users\...\...> pip list
Package                 Version
...
fastapi                 0.115.12
fastmcp                 2.8.0
langchain               0.3.25
numpy                   2.3.0
openai                  1.85.0
python-dotenv           1.1.0
qdrant_client　　　　　　1.14.2 
requests　　　　　　　   2.32.4
※2025年6月17日現在※

 2. .envの設定AZURE_OPENAI_API_KEY = '' # テキストモデルのAPIキー
AZURE_OPENAI_ENDPOINT = '' # Azure OpenAIのエンドポイント
MONGODB_URL = '' # mongoDBのURL
MONGODB_API_TOKEN = '' # mongoDBのAPIトークン
MONGODB_PATH = "" # このパス以下で全ページを保存している
QDRANT_LOCAL_URL = 'http://localhost:6333' #Qdrant_DBのポートフォワードしたポート番号

 3. QdrantDBの起動以下のコマンドで行います。
docker run -p 6333:6333 -p 6334:6334 \
    -v "$(pwd)/qdrant_storage:/qdrant/storage:z" \
    qdrant/qdrant

 4. FastMCPサーバの起動mcp_qdrant_server.pyを実行し、起動したMCPサーバの URLをコピーします。

mcp inspectorを以下のコマンドで実行します。
npx @modelcontextprotocol/inspector
※ 参照：MCP Inspectorの実行
下のスクリーンショットでハイライトしたリンクをブラウザで開きます。

ブラウザで立ち上がった MCP inspector のページで、URL にMCPサーバの URL を記入し、Connectボタンを押下します。

そうすると、ブラウザ上でMCPサーバのツールなどが使えるようになります。


 サンプルコード
 モデルの定義
 見出しごとに分割されるセクションのモデルクラスは以下のようです。Sectionクラスに下記のフィールドがあります。


フィールド(データ型)
説明


pg_id (str)
ページのid

h1 (Optional[TextEmbedding])
見出し1のテキスト埋め込み「任意の値」

h2 (Optional[TextEmbedding])
見出し2のテキスト埋め込み「任意の値」

h3 (Optional[TextEmbedding])
見出し3のテキスト埋め込み「任意の値」

p (Optional[TextEmbedding])
本文のテキスト埋め込み「任意の値」

section.py
from typing import Optional

from model.text_embed import TextEmbedding

class Section:
    def __init__(
        self,
        pg_id: str,
        h1: Optional[TextEmbedding] = None,
        h2: Optional[TextEmbedding] = None,
        h3: Optional[TextEmbedding] = None
        p: Optional[TextEmbedding] = None,
    ):
        self.pg_id = pg_id
        self.h1 = h1
        self.h2 = h2
        self.h3 = h3
        self.p = p

 テキスト埋め込みのモデルクラスは以下のようです。TextEmbeddingクラスに下記のフィールドがあります。


フィールド(データ型)
説明


text (str)
見出し、本文のテキストデータ

embeddings (List[List[float]])
ベクトル埋め込み配列

QdrantベクトルDBには、一つのデータまたはペイロードに対して複数のベクトルを持てるマルチベクトルという機能があります。今度のプロジェクトで複数の見出しと本文が属している一つのセクションに対して、その複数の見出しと本文のベクトルをマルチベクトルとしてまとめています。
そのため、text_embed.py ファイルには、見出しや本文などのすべてのベクトルをベクトルDBに保存するときに1つの配列に結合する関数concatenate_embeddingsも用意しています。
参照：Qdrantのマルチベクトル
text_embed.py
from typing import TypedDict, List, Optional

class TextEmbedding(TypedDict):
    text: str 
    embeddings: List[List[float]]


def concatenate_embeddings(*textEmbeds: Optional[TextEmbedding]) -> List[List[float]]:
    """全てのベクトル埋め込みを一つのベクトル埋め込み配列にする
    Args:
        *textEmbeds: いくつかのTextEmbeddingインスタンス
    Returns:
        合わせたベクトル埋め込みの配列
    """
    return [vec for textEmbed in textEmbeds if textEmbed is not None for vec in textEmbed["embeddings"]]

 QdrantベクトルDBの操作以下のサンプルコードは、主に下記のQdrantベクトルDB操作ができます。
検索「文字列型のユーザークエリに対しての」
データ登録
全件取得
全件削除
qdrant.py
import os
from dotenv import load_dotenv
from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance,
    MultiVectorConfig,
    MultiVectorComparator,
    PointStruct,
    VectorParams,
)

import uuid
import time

from model.section import Section
from model import text_embed

import vector_embed

# .envファイルをロードする
qdrant_url = os.getenv("QDRANT_LOCAL_URL")  # qdrantのURL

load_dotenv()

client = QdrantClient(url=qdrant_url)

collection = "md_collection"

# Qdrantコレクションの作成「既存してない場合」
if not client.collection_exists(collection_name=collection):
    client.create_collection(
        collection_name=collection,
        vectors_config=VectorParams(
            size=1536,
            distance=Distance.COSINE,
            multivector_config=MultiVectorConfig(
                comparator=MultiVectorComparator.MAX_SIM
            ),
        ),
    )

def upload_section(section: Section):
    """区切れたセクションをQdrantベクトルDBにポイントとして登録する"""
    # ランダムなUUID（バージョン4）文字列の生成
    unique_id_str = str(uuid.uuid4())

    # 複数の見出しと本文をまとめたマルチベクトル
    vectors = text_embed.concatenate_embeddings(
        section.h1, section.h2, section.h3, section.p
    )

    payload_data = {"pg_id": section.pg_id, "p": section.p["text"]}

    if (section.h1 is not None) or (section.h1["text"]):
        payload_data["h1"] = section.h1["text"]

    if (section.h2 is not None) or (section.h2["text"]):
        payload_data["h2"] = section.h2["text"]

    if (section.h3 is not None) or (section.h3["text"]):
        payload_data["h3"] = section.h3["text"]

    point = PointStruct(id=unique_id_str, vector=vectors, payload=payload_data)

    client.upsert(collection_name=collection, points=[point])


def get_all_points():
    """全てのポイントを取得する"""
    start_time = time.time()
    all_points = client.scroll(
        collection_name=collection,
        scroll_filter=None,  # No filter = get all
        limit=1000,  # Adjust limit as needed
    )
    end_time = time.time()
    execution_time = end_time - start_time
    print(f"全ポイント取得の実行時間: {execution_time:.4f} seconds")
    return all_points[0]
    # all_points is a tuple: (list_of_points, next_offset)
    # for point in all_points[0]:
    #     print(point)


def get_number_of_points():
    """全てのポイント数を取得する"""
    collection_info = client.get_collection(collection_name=collection)
    print("Number of points: ", collection_info.points_count)


def search_by(query: str, limit: int, with_vector=False):
    """クエリで検索する
    Args:
        query: ユーザークエリ
        limit: 結果件数の制限
    Returns:
        out: 類似度高い辞書型のリスト
    """
    query_embed_start_time = time.time()
    # ユーザークエリのベクトル化
    query_embedding = vector_embed.create_embeddings(query)

    query_embed_end_time = time.time()
    query_embed__time = query_embed_end_time - query_embed_start_time
    
    print(f"クエリ埋め込みの実行時間: {query_embed__time:.4f} 秒")
    
    start_time = time.time()
    scored_points = client.query_points(
        collection_name=collection,
        query=query_embedding,
        with_payload=True,
        limit=limit,
        with_vectors=with_vector
    ).points
    end_time = time.time()
    execution_time = end_time - start_time
    print(f"検索実行時間: {execution_time:.4f} seconds")
    return scored_points


def delete_all_points():
    """全てのポイントを削除する"""
    start_time = time.time()
    if client.collection_exists(collection_name=collection):
        client.delete_collection(collection_name=collection)
        print(f"コレクション'{collection}'の削除が完了しました。")
        # Recreate the collection after deletion
        client.create_collection(
            collection_name=collection,
            vectors_config=VectorParams(
                size=1536,
                distance=Distance.COSINE,
                multivector_config=MultiVectorConfig(
                    comparator=MultiVectorComparator.MAX_SIM
                ),
            ),
        )
        print(f"コレクション'{collection}'の再作成が完了しました。")
    end_time = time.time()
    execution_time = end_time - start_time
    print(f"削除実行時間: {execution_time:.4f} seconds")

 セクションの操作こちらのsection_handler.pyスクリプトは、モデルの定義のSectionとTextEmbeddingモデルを使用し、見出しごとに分割し、分割されたセクションをQdrantベクトルDBに登録する役割を果たします。
section_handler.py
from typing import Optional
from langchain.schema import Document
import time

from model.section import Section
from model.text_embed import TextEmbedding

import mongoDB_api
import qdrant
import split_markdown
import split_text
import vector_embed


def generate_json_data(text: str) -> dict:
    """JSONデータを生成する
    Args:
        text: テキスト
    Returns:
        テキスト、ベクトル埋め込み配列のJSONデータ
    """
    # テキストをチャンクに分ける
    text_chunks = split_text.split(text)

    # テキストとトークン数のJSONデータ
    result = {
        "text": text,
        "embeddings": [vector_embed.create_embeddings(chunk) for chunk in text_chunks],
    }

    return result


def split_sections(pg_id: str, docs: list[Document]) -> list:
    """各ドキュメントごとにセクション化する"""
    h1: Optional[TextEmbedding] = {"text": "", "embeddings": []}
    h2: Optional[TextEmbedding] = {"text": "", "embeddings": []}
    h3: Optional[TextEmbedding] = {"text": "", "embeddings": []}
    p: Optional[TextEmbedding] = {"text": "", "embeddings": []}

    section_list = []

    for doc in docs:
        # print(doc.page_content)
        if doc.metadata.get("header1"):
            h1["text"] = doc.metadata["header1"]
        if doc.metadata.get("header2"):
            h2["text"] = doc.metadata["header2"]
        if doc.metadata.get("header3"):
            h3["text"] = doc.metadata["header3"]
        if doc.page_content:
            p["text"] = doc.page_content

        if (p["text"]) or (not p["text"].startswith("---")):
            if h1["text"] is not None:
                h1 = generate_json_data(h1["text"])

            if h2["text"] is not None:
                h2 = generate_json_data(h2["text"])

            if h3["text"] is not None:
                h3 = generate_json_data(h3["text"])

            if p["text"] is not None:
                p = generate_json_data(p["text"])

            section = Section(pg_id=pg_id, h1=h1, h2=h2, h3=h3, p=p)
            # print(f"section.h1: {h1['text']}")
            # print(f"section.p: {p['text']}")
            section_list.append(section)

    return section_list



def upload_to_qdrant(url: str, access_token: str, all_pages: list):
    """分割されたセクションをqdrantに登録する"""
    start_time = time.time()
    for page in all_pages:
        page_data = mongoDB_api.get_page_data(url, access_token, page["_id"])
        body_length = page_data["latestRevisionBodyLength"]
        page_body = page_data["revision"]["body"]
        pg_id = page["_id"]
        print(f"各ページの_id: {pg_id}")

        # マークダウンの分割
        splitted_docs = split_markdown.split(page_body)
        print(f"分割されたドキュメント数: {len(splitted_docs)}")

        # セクションの分割
        sections = split_sections(pg_id, splitted_docs)
        print(f"分割されたセクション数: {len(sections)}")

        print(f"文書の長さ: {body_length}")
        # if (has_diff_to_prev == True) and (body_length > 0): #should be False, but temporarily set True
        if body_length > 0:
            for section in sections:
                section_index = sections.index(section)
                print(f"section index: {section_index}")
                qdrant.upload_section(section)
    end_time = time.time()
    execution_time = end_time - start_time
    print(f"登録実行時間: {execution_time:.4f} seconds")


 QdrantベクトルDB操作用のMCPサーバこちらは、FastMCP経由でQdrantベクトルDBの操作ができるソースコードです。

MCPサーバの二つの機能であるツール(Tool)とリソース(Resource)を使用しました。
ツール（Tool）には、以下の機能があります：
検索
登録 ※
全件削除 ※

また、リソース（Resource）には「全件取得」の機能があります。
※ 念のため、QdrantベクトルDBの「登録」および「全件削除」機能には、一般ユーザーがアクセスできないようtag機能で隠しました。
mcp_qdrant_server.py
from fastapi import HTTPException
from fastmcp import FastMCP
from pydantic import BaseModel
import os
from dotenv import load_dotenv

import growi_api
import qdrant
import section_handler

# .envファイルをロードする
load_dotenv()

# GrowiのURLとAPIトークン
wiki_url = os.getenv("WIKI_URL")  # WikiのURL
wiki_api_token = os.getenv("WIKI_API_TOKEN")  ##WikiのAPIトークン

# mcp
mcp = FastMCP(name="McpQdrantServer", exclude_tags={"upload", "delete_all"})


# クエリリクエストボディのモデル
class QueryRequest(BaseModel):
    query: str
    top_k: int = 5


# Wikiリクエストボディのモデル
class WikiRequest(BaseModel):
    path: str


@mcp.tool(
    tags={"find", "public"}
)
async def search_similar_data(request: QueryRequest):
    """
    指定されたクエリに類似するデータをQdrantで非同期に検索するツール
    Args:
        request (QueryRequest): 検索クエリと返す件数を含むリクエストオブジェクト
        {
            "query": ユーザからの検索クエリ
            "limit": 検索件数（何も指定がなければ5件）
        }
    Returns:
        List[dict]: 検索結果のペイロードのリスト
    Raises:
        HTTPException: 
            - 結果が見つからない場合は404を返します。
            - 検索処理中に予期しないエラーが発生した場合は500を返します。
    """
    
    try:
        results = qdrant.search_by(query=request.query, limit=request.top_k)
        if not results:
            raise HTTPException(status_code=404, detail="No results found")

        print(f"結果の数: {len(results)}")
        print(f"クエリ: {request.query}, 件数制限: {request.top_k}")
        print(f"結果: {results}")
        # 結果のペイロードリストを返す
        return [hit.payload for hit in results]

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@mcp.resource(
    uri="resource://all-points",
    mime_type="application/json",
    tags={"store", "private", "get_all"}
)
async def get_all_points():
    """全てのポイントを取得するツール"""
    try:
        all_points = qdrant.get_all_points()
        if not all_points:
            raise HTTPException(status_code=404, detail="No points found")
        return all_points

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@mcp.tool(
    tags={"store", "private", "upload"}
)
async def insert_all_pages_from(wiki_path: str):
    """
    指定されたパスからWikiページを非同期でQdrantにアップサートするツール
    Args:
        wiki_path (str): Wikiページのパス
    Returns:
        dict: アップサート成功メッセージとアップサートしたページ数を含む辞書
    Raises:
        HTTPException: 
            - 指定パスにページが見つからない場合は404を返します。
            - アップサート処理中にエラーが発生した場合は500を返します。
    """
    
    # Wikiのすべてベージを取得する
    loaded_pages = growi_api.get_page_list(wiki_url, wiki_api_token, wiki_path)
    if not loaded_pages:
        raise HTTPException(
            status_code=404, detail="指定されたパスにページが見つかりませんでした。"
        )
    print(f"取得したページ数: {len(loaded_pages)}")
    try:
        print(f"アップサートするページのパス: {wiki_path}")
        # qdrantにデータ登録
        section_handler.upload_to_qdrant(
            url=wiki_url, access_token=wiki_api_token, all_pages=loaded_pages
        )

        # 成功メッセージを返す
        return {
            "message": "アップサートが成功しました。",
            "page_count": len(loaded_pages),
        }

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@mcp.tool(
    tags={"store", "private", "delete_all"}
)
async def delete_all_points():
    """全てのポイントを削除するツール"""
    try:
        qdrant.delete_all_points()
        return {"message": "全てのポイントが削除されました。"}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


# このスクリプトが直接実行されたときのみアプリを起動する
if __name__ == "__main__":
    mcp.run(transport="streamable-http")


 結果10件のmarkdownファイルの登録実行時間は179.9831 秒かかりました。

よく使うプログラミング言語というクエリに対する結果は、以下のようです。

クエリの埋め込み時間：1.2550 秒

検索実行時間：0.0215 秒


よく使うプログラミング言語にたいする上位5件の取得したペイロードまたはデータのみのJSON結果は以下のようです。
[
  {
    "pg_id": "683d3ea1a9cb231ee7e80a1d",
    "p": "プログラミング言語は以下のような目的で使用されます：  \n- **ソフトウェア開発**：アプリケーションやゲーム、Web サービスなどを作成。\n- **データ処理**：大量のデータを分析・処理。\n- **自動化**：繰り返し作業の効率化。\n- **教育・研究**：アルゴリズムや論理的思考の学習。  \n---",
    "h1": "プログラミング言語とは",
    "h2": "💡 プログラミング言語の目的",
    "h3": ""
  },
  {
    "pg_id": "683d3ea1a9cb231ee7e80a1d",
    "p": "プログラミング言語は、いくつかの観点で分類されます：  \n- **コンパイル型 vs インタプリタ型**  \n- コンパイル型：C, C++（事前に機械語に変換）\n- インタプリタ型：Python, JavaScript（逐次実行）  \n- **手続き型 vs オブジェクト指向型**  \n- 手続き型：C（命令の順序で処理）\n- オブジェクト指向型：Java, Python（データと処理を一体化）  \n- **静的型付け vs 動的型付け**\n- 静的型付け：Java, C++（変数の型を明示）\n- 動的型付け：Python, Ruby（実行時に型が決定）  \n---",
    "h1": "プログラミング言語とは",
    "h2": "🛠️ プログラミング言語の分類",
    "h3": ""
  },
  {
    "pg_id": "683d3ea1a9cb231ee7e80a1d",
    "p": "AI や IoT、量子コンピューティングの発展により、新しいプログラミング言語やパラダイムが登場しています。今後も、より効率的で安全な開発を支える技術として進化し続けるでしょう。  \n---",
    "h1": "プログラミング言語とは",
    "h2": "📈 今後の展望",
    "h3": ""
  },
  {
    "pg_id": "683d3ea1a9cb231ee7e80a1d",
    "p": "AI や IoT、量子コンピューティングの発展により、新しいプログラミング言語やパラダイムが登場しています。今後も、より効率的で安全な開発を支える技術として進化し続けるでしょう。  \n---",
    "h1": "プログラミング言語とは",
    "h2": "📈 今後の展望",
    "h3": ""
  },
  {
    "pg_id": "683d3ea1a9cb231ee7e80a1d",
    "p": "目的や用途に応じて、適切な言語を選ぶことが重要です：  \n- **初心者向け**：Python, Scratch\n- **Web 開発**：HTML/CSS + JavaScript + フレームワーク（React, Vue など）\n- **モバイルアプリ**：Swift（iOS）、Kotlin（Android）\n- **ゲーム開発**：C++, C#, Unity  \n---",
    "h1": "プログラミング言語とは",
    "h2": "🚀 プログラミング言語の選び方",
    "h3": ""
  }
]
その上位5件のスコアポイントは以下のようです。
[
    ScoredPoint(id='14e8633f-b5a8-467b-984e-892a05ec4026', version=46, score=0.71252066, payload={'pg_id': '683d3ea1a9cb231ee7e80a1d', 'p': 'プログラミング言語は以下のような目的で使用されます：  \n- **ソフトウェア開発**：アプリケーションやゲーム、Web サービスなどを作成。\n- **データ処理**：大量のデータを分析・処理。\n- **自動化**：繰り返し作業の効率化。\n- **教育・研究**：アルゴリズムや論理的思考の学習。  \n---', 'h1': 'プログラミング言語とは', 'h2': '💡 プログラミング言語の目的', 'h3': ''}, vector=None, shard_key=None, order_value=None), 
    ScoredPoint(id='d06aefe6-c3fd-48e8-9de7-40bfd85962cf', version=48, score=0.71252066, payload={'pg_id': '683d3ea1a9cb231ee7e80a1d', 'p': 'プログラミング言語は、いくつかの観点で分類されます：  \n- **コンパイル型 vs インタプリタ型**  \n- コンパイル型：C, C++（事前に機械語に変換）\n- インタプリタ型：Python, JavaScript（逐次実行）  \n- **手続き型 vs オブジェクト指向型**  \n- 手続き型：C（命令の順序で処理）\n- オブジェクト指向型：Java, Python（データと処理を一体化）  \n- **静的型付け vs 動的型付け**\n- 静的型付け：Java, C++（変数の型を明示）\n- 動的型付け：Python, Ruby（実行時に型が決定）  \n---', 'h1': 'プログラミング言語とは', 'h2': '🛠️ プログラミング言語の分類', 'h3': ''}, vector=None, shard_key=None, order_value=None), 
    ScoredPoint(id='422288cd-ccf5-4226-b39e-f6986652c123', version=50, score=0.71252066, payload={'pg_id': '683d3ea1a9cb231ee7e80a1d', 'p': 'AI や IoT、量子コンピューティングの発展により、新しいプログラミング言語やパラダイムが登場しています。今後も、より効率的で安全な開発を支える技術として進化し続けるでしょう。  \n---', 'h1': 'プログラミング言語とは', 'h2': '📈 今後の展望', 'h3': ''}, vector=None, shard_key=None, order_value=None), 
    ScoredPoint(id='750a4891-4522-472b-8c2a-77e1c800af02', version=51, score=0.71252066, payload={'pg_id': '683d3ea1a9cb231ee7e80a1d', 'p': 'AI や IoT、量子コンピューティングの発展により、新しいプログラミング言語やパラダイムが登場しています。今後も、より効率的で安全な開発を支える技術として進化し続けるでしょう。  \n---', 'h1': 'プログラミング言語とは', 'h2': '📈 今後の展望', 'h3': ''}, vector=None, shard_key=None, order_value=None), 
    ScoredPoint(id='91de99fc-f0e9-4c51-8c9b-fee06e245bf8', version=49, score=0.71252066, payload={'pg_id': '683d3ea1a9cb231ee7e80a1d', 'p': '目的や用途に応じて、適切な言語を選ぶことが重要です：  \n- **初心者向け**：Python, Scratch\n- **Web 開発**：HTML/CSS + JavaScript + フレームワーク（React, Vue など）\n- **モバイルアプリ**：Swift（iOS）、Kotlin（Android）\n- **ゲーム開発**：C++, C#, Unity  \n---', 'h1': 'プログラミング言語とは', 'h2': '🚀 プログラミング言語の選び方', 'h3': ''}, vector=None, shard_key=None, order_value=None)
]

 考察導入のしやすさという観点から見ると、QdrantのベクトルDBにはマルチベクトル機能があるため、今回のプロジェクトでは、複数の見出しと本文が属する一つのセクションに対して、それぞれのベクトルを1つの配列にまとめることができました。その結果、ベクトル検索の際には、ユーザークエリに対してすべてのベクトルを考慮し、類似度を照らし合わせることができるため、非常に有用です。
パフォーマンス面では、同じ10件のMarkdownファイルをクラウド上のMongoDBからローカルホストのQdrantベクトルDBに登録する処理時間は、ローカルフォルダーからJSONファイルとして保存する場合より約30秒遅くなりました。ただし、これはネットワークの影響も考えられるため、特に問題はないと考えています。一方、同じ質問に対する検索処理時間は、QdrantベクトルDBからの検索（0.0215秒）の方が、ローカルJSONファイルからの検索（0.1075秒）より約0.08秒速く、パフォーマンス面でも優れていることが確認できました。

 まとめ & 次のステップ
今度のプロジェクトでは、QdrantベクトルDBを使って、登録と検索を行いました。登録と検索の実行時間はそれぞれ179.9831 秒と0.0215 秒でした。
今回は10件のみのMarkdownファイルで検証した結果ですが、実際には今後データ量が何倍にも増えていくことが想定されるため、その際に結果がどう変化するかは、実際に試してみないと分かりません。
今後は、QdrantベクトルDBの更新機能も追加し、ファイル保存型プロトタイプとQdrantDB用のプロトタイプの登録および検索の結果を比較検証していく予定です。

 参考リンクベクトル埋め込みと検索のプロトタイプ
Azure OpenAIテキスト埋込モデルの比較の記事
MCP Inspectorの実行
Qdrantのマルチベクトル
項目	バージョン
OS	Ubuntu 20.04.6 LTS
ランタイム	Python 3.10.12
主要ライブラリ	fastapi 0.115.12、fastmcp 2.8.0、langchain 0.3.25、numpy 2.2.6、openai 1.86.0、python-dotenv 1.1.0、qdrant_client 1.14.2、requests 2.32.4、tiktoken 0.9.0
モデル	Azure OpenAI text-embedding-3-small
フィールド(データ型)	説明
pg_id (str)	ページのid
h1 (Optional[TextEmbedding])	見出し1のテキスト埋め込み「任意の値」
h2 (Optional[TextEmbedding])	見出し2のテキスト埋め込み「任意の値」
h3 (Optional[TextEmbedding])	見出し3のテキスト埋め込み「任意の値」
p (Optional[TextEmbedding])	本文のテキスト埋め込み「任意の値」
フィールド(データ型)	説明
text (str)	見出し、本文のテキストデータ
embeddings (List[List[float]])	ベクトル埋め込み配列
セリオ株式会社テックブログPublication
未来の技術に挑戦し続ける岡山のソフトウェア企業です。製造業をはじめ、物流・介護などさまざまな業界のお客様と共創しDXを実現します。
背景

対象読者

環境

事前準備

プロジェクトの全体像

プロジェクト構造

手順

1. pythonパッケージのインストール

2. .envの設定

3. QdrantDBの起動

4. FastMCPサーバの起動

サンプルコード

モデルの定義

見出しごとに分割されるセクションのモデルクラスは以下のようです。

テキスト埋め込みのモデルクラスは以下のようです。

QdrantベクトルDBの操作

セクションの操作

QdrantベクトルDB操作用のMCPサーバ

結果

考察

まとめ & 次のステップ

参考リンク

Discussion
