2026年8月29日土曜日

100+ AI Concepts for Developers Explained in 20 Minutes. (Part 01) - - YouTube動画の紹介

YouTube動画「100+ AI Concepts for Developers Explained in 20 Minutes (Part 01)」(Codist チャンネル)で取り上げられている主要なAI概念について、開発者視点での実践的な意味合いや、業界の裏話・雑学を交えて分かりやすく解説します。

1. ニューラルネットワーク(Neural Networks)

人間や動物の脳の神経回路網(ニューロン)を数学的に模したアルゴリズムです。

  • どんな概念? 入力(データ)を受けて重み付け計算を行い、活性化関数(Activation Function)を通して出力を行う層(Layer)の集まりです。

  • 💡 業界話・雑学

    • 冬の時代の克服:1980年代〜90年代にもニューラルネットワークのブームがありましたが、「計算に時間がかかりすぎる」「データが足りない」という理由で下火になり「AIの冬」が訪れました。それを打破したのが2012年の「AlexNet」の登場と、GPU(グラフィックボード)の転用です。ゲーム画面の描写に使われていたGPUが、たまたまニューラルネットの巨大な行列計算に最適だったことが現代のAI革命の引き金になりました。

2. 学習と推論(Training & Inference)

AIモデルの開発と運用の2大プロセスです。

  • どんな概念?

    • 学習(Training):大量のデータと正解を与えて、モデル内のパラメータ(重み)を最適化する工程(AIに勉強させる作業)。

    • 推論(Inference):完成したモデルに新しいデータを与えて、回答や予測を出力する工程(テストを受ける作業)。

  • 💡 業界話・雑学

    • コストのギャップ:巨大なLLMの「学習」には数十億円〜数百億円の計算コスト(電力とGPU利用代)がかかることで有名ですが、実はプロダクト運用における総コストの8割以上は「推論」にかかると言われています。世界中のユーザーが日々ChatGPT等にプロンプトを打ち込むたびにリアルタイムで膨大な計算が発生しているため、各社「いかに推論コストを下げるか(量子化や専用チップの開発)」に命をかけています。

3. トークナイゼーション(Tokenization)

テキストをAIが理解できる数字の最小単位「トークン」に分解する処理です。

  • どんな概念? AIは文章をそのまま読むことができず、内部ではすべて数値(ID)として処理します。「Unbelievable」を「Un」「believ」「able」という部分文字列(サブワード)に切り分けてID化します。

  • 💡 業界話・雑学

    • 日本語の「損」:英語だと「apple」は1トークンで済むことが多いですが、日本語やその他の言語はバイト単位で細かく分割されやすいため、同じ長さの文章でも日本語の方がトークン数が多くなりがちです。ChatGPTなどのAPI料金は「トークン数」で課金されるため、実は日本語でやり取りするとコストが高くつく(かつコンテキストウィンドウを無駄に圧迫する)という地味な悲しい現実があります。

    • 不思議なトークン:トークナイザーの癖で、学習データ内にたまたま多く存在した「ネット掲示板のハンドルネーム(例: SolidGoldMagikarp)」などが単一の特殊トークンに登録され、それを入力するとAIがおかしくなる(Glitch Token)という現象が初期のChatGPTで話題になりました。

4. 埋め込み(Embeddings)

文字や画像などのデータを、意味の近さを計算できる多次元の「ベクトル(数値の配列)」に変換する技術です。

  • どんな概念? 例えば「王様」「女王」「りんご」という言葉を多次元空間に配置すると、「王様」と「女王」の距離は近く、「りんご」は遠い位置にマッピングされます。

  • 💡 業界話・雑学

    • 有名な「ベクトル演算の魔法」:Embeddingの説明で必ず引き合いに出されるのが、以下の有名なベクトル算式です。

      ベクトル("王様") - ベクトル("男性") + ベクトル("女性") ≒ ベクトル("女王") 言葉の意味が空間上の「方向」と「距離」として完全に数値化されているため、このような「引き算と足し算」で意味の検索や類推が可能になります。今流行りのRAG(社内文書検索AI)の根幹を支える技術です。

5. アテンション機構(Attention Mechanism)

文章中の「どの単語とどの単語が強い関連性を持っているか」を動的に見極める仕組みです。

  • どんな概念? 例えば「が公園を走っていた。それはとても大きかった」という文があったとき、「それ」が「公園」ではなく「犬」を指していることに注目(Attention)して文脈を読み解きます。

  • 💡 業界話・雑学

    • 翻訳のイノベーション:元々はRNN(再帰的ニューラルネットワーク)というAIの弱点(長文になると最初の文脈を忘れてしまう)を克服するために、2014〜2015年頃に研究者(BahdanauらやLuongら)が発案したものでした。これが後のTransformerの核となります。

6. トランスフォーマー(Transformers)

現代の生成AI(ChatGPT, Claude, Geminiなど)のすべてのベースとなっている超革命的なニューラルネットワーク建築(アーキテクチャ)です。

  • どんな概念? 「アテンション機構」だけを全面的に使い、データを並列処理(一度にまとめて処理)できるようにしたモデルです。

  • 💡 業界話・雑学

    • 伝説の論文 "Attention Is All You Need":2017年にGoogleの研究チームが発表したこの論文1本が、それまでのAI業界をすべて塗り替えました。論文のタイトルを直訳すると「アテンションさえあれば(他の複雑な仕組みは)何もいらない」という非常にロックなものです。

    • Googleの誤算:この革命的技術を発明したのはGoogleの研究者たち(8人の著者)ですが、当時のGoogleが慎重姿勢をとっている間に、OpenAIがこの技術を全面的に採用して「GPT」を作り、世界を驚かせました。ちなみに論文の著者8人は全員Googleを退職し、それぞれ大成功したAIスタートアップ(Cohere, Character.ai, Essential AIなど)を起業しています。

7. コンテキストウィンドウ(Context Window)

AIが一度に理解・記憶できる「入力+出力の最大トークン量」のことです。

  • どんな概念? いわゆる「AIの短期記憶の限界容量(作業デスクの広さ)」です。

  • 💡 業界話・雑学

    • 4Kから100万、そして数百万へ:初期のGPT-3.5のコンテキストウィンドウはわずか4,000トークン(原稿用紙数枚分)程度でした。そのため、長い会話をするとすぐ最初の方を忘れ「ニワトリ並みの記憶力」と揶揄されました。しかし、2024年以降はGeminiが「200万トークン(本数十冊分、動画数時間分を丸ごと読み込める)」を実現するなど、コンテキストウィンドウの巨大化競争が激化しました。

    • 「針を探す(Needle In A Haystack)」テスト:長大なテキストのどこかにポツンと「秘密の単語」を隠しておき、AIがちゃんとそれを引き出せるかをテストする業界標準のベンチマークが存在します。

8. スケーリング法則(Scaling Laws)

「モデルのパラメータ数」「学習データの量」「計算量(GPUパワー)」の3つを増やせば増やすほど、AIの性能は対数関数的に比例して上がり続けるという経験則です。

  • どんな概念? AI開発における「巨大化こそ力(Big is Better)」の理論的裏付けです。

  • 💡 業界話・雑学

    • OpenAIの基本哲学:2020年にOpenAIのアフレッド・カプランらがこの法則を定式化しました。「知能の質を上げる特別な魔法の工夫をするより、とにかくデータとGPUを100倍投資すれば賢くなる」という力技が証明されたため、GAFAMをはじめとする巨額のデータセンター投資合戦(GPU爆買い狂騒曲)が始まりました。

9. 事前学習と事後学習(Pre-training vs. Post-training)

現代のLLMを作る2つの主要フェーズです。

  • どんな概念?

    • 事前学習(Pre-training):ネット上のあらゆるテキスト(Wikipedia、Webサイト、書籍など)を丸暗記させ、「次に続く単語を予測する(文補完成能力)」を獲得させる基礎工事。

    • 事後学習(Post-training / RLHF・SFT):事前学習だけだと「質問に答える」のではなく「質問の続きの文章を作ってしまう」ため、人間に役立つ「対話エージェント(アシスタント)」として振る舞うよう教育・調整(RLHF: 人間のフィードバックによる強化学習など)する仕上げ作業。

  • 💡 業界話・雑学

    • 事前学習済みモデルは「狂った天才」:事前学習が終わった直後のモデルは、インターネットの知性をすべて持っていますが、倫理観も会話能力もなく、ただ「文章の続き」を予測するだけの状態です。これを「RLHF(Post-training)」で調教することで、丁寧で安全な「ChatGPT」の性格が完成します。業界では「野性のモンスターを人里で暮らせるように躾ける作業」例えられます。

10. 大規模言語モデル(Large Language Models - LLM)

数百億〜数千億以上のパラメータを持ち、人間のような自然な文章生成や推論能力を獲得した巨大なトランスフォーマーモデルのことです。

  • どんな概念? GPT-4、Claude 3.5、Llama 3、Geminiなどの総称です。単なる「単語予測機」の領域を超えて、コードの記述、論理パズル、創作など多岐にわたる知能を示します。

  • 💡 業界話・雑学

    • 創発的能力(Emergent Abilities):モデルの規模がある閾値(しきい値)を超えると、「設計者が教えてもいない能力(計算、ジョークの解説、プログラミングなど)が突如として開花する」現象が確認されています。なぜスケールを大きくするだけでこんな知能が生まれるのか、完全な数学的メカニズムは未だに解明しきれておらず、現代科学の最もエキサイティングな謎の一つです。

💡 動画タイトルの補足:「API / MCP / A2A」とは?

概要欄で紹介されている「API vs MCP vs A2A」についても簡単に雑学を添えておきます。

  • API:プログラム同士が通信する従来からの規格(Webの基本)。

  • MCP(Model Context Protocol):Anthropicが2024年に提案した、「AIモデルが外部データベースやツールに接続するための新しい標準規格」。AI版のUSBポートのようなものです。

  • A2A(Agent to Agent):単一のAIではなく、「プログラミング専門AI」「デザイン専門AI」「テスト専門AI」といった複数AIが協調して自律的に会話・タスク処理を行う未来のアーキテクチャのことです。

動画のPart 1では基礎的な機械学習・言語モデルの用語がカバーされていますが、Part 2以降ではこれら「AIエージェント時代」の実務開発概念へと踏み込んでいく構成になっています。

この概念を踏まえて、さらに知りたい方向性はありますか?
RAGとMCPの違いを詳しく解説してほしい
AI概念の「Part 02」で出てくる用語を予習したい

「RAGとMCPのどちらを採用すべきか?」という問いは、近年のAI開発で非常に多く見られますが、結論から言えば「比較する次元が異なる技術」です。

Coworker AI
  • RAG (Retrieval-Augmented Generation) は、外部データを検索して回答精度を高めるための「設計パターン(手法)」

    StackOne
  • MCP (Model Context Protocol) は、AIアプリと外部システム(DB・ツール)を接続するための「通信規格(プロトコル)」

    MCP-Builder.ai他 1 件

日常生活に例えると、RAGは「図書館で目当ての本を探して要約する技術」であり、MCPは「USB-Cのように何でも接続できる共通プラグ規格」のような関係にあたります。

MCP-Builder.ai

アーキテクチャと動作仕組みの比較

RAGの仕組み:パイプライン型アーキテクチャ

StackOne
  1. 情報をあらかじめ切り分けてベクトル化し、ベクトルDBにインデックス登録(事前準備)。

    StackOne
  2. ユーザーの質問に対し、コサイン類似度などを用いて関連する文章(チャンク)を検索・抽出。

  3. 抽出した文章をプロンプトに埋め込んでLLMに渡し、回答を生成させる。

    StackOne

MCPの仕組み:Host - Client - Server 型プロトコルアーキテクチャ

MCP-Builder.ai
  1. MCP Server が外部APIやDB、システム(Slack, GitHub, 社内DB等)の「操作機能(Tool)」や「データ源(Resource)」を定義して公開。

    Infragistics
  2. MCP HostClaude DesktopやIDEなど)上の MCP Client がそのサーバーと通信し、利用可能なツール一覧を動的に取得。

    MCP-Builder.ai
  3. LLMが「このツールを実行して最新情報を取ってくる(または書き込む)」と判断すると、MCP経由でリアルタイムに実行・取得する。

    MCP-Builder.ai

機能と使用感の違い

StackOne
比較項目 RAG(検索拡張生成) MCP(Model Context Protocol)
分類 情報検索・知識補完のアルゴリズム クライアントとサーバーをつなぐ通信規格
主な目的 長文ドキュメントからの意味検索、ハルシネーション抑制 外部ツール・APIの実行、リアルタイムな動的データ取得
得意なデータ 静的な大量非構造テキスト(PDF、Wiki、業務マニュアル) 動的な最新データ、構造化データ(SQL DB、CRM、API)
操作権限 参照のみ(Read-only) 参照(Read)+ 書き込み・システム操作(Write / Execute)
データの鮮度 事前にインデックスを作成・更新する必要がある いつでも呼び出し時点のリアルタイム最新データ
再利用性 システムごとに検索パイプラインの個別構築が必要 MCP規格に対応していれば、1つのServerを多様なAIアプリで使い回せる

実務での役割分担と組み合わせ(ハイブリッド構成)

実務のエージェント開発では「どちらか一方を選ぶ」のではなく、「MCPのツールの一つとしてRAGを組み込む」のが標準的な構成(最終形態)となります。

Plaintext
[ ユーザーの要求 ] 
        │
        ▼
   [ AIエージェント (MCP Host) ]
        │
        ├──▶【 MCP Server A 】: RAGサーバー(社内規程・PDFをベクトル検索して読み込む)
        ├──▶【 MCP Server B 】: 基幹DBサーバー(現在の在庫数や注文状況をリアルタイム取得)
        └──▶【 MCP Server C 】: Slack Server(確認した結果をチームチャンネルに通知・書き込み)

【例:カスタマーサポートAIエージェントの処理フロー】

  1. RAGで調べる(Read): 「返金規定はどうなっているか?」→ RAG経由でPDFマニュアルから条件を検索。

  2. MCPで参照する(Live Read): 「このユーザーの直近の購入履歴は?」→ MCP経由で注文管理システムのAPIを叩いて取得。

  3. MCPで実行する(Write): 「返金条件を満たしているので処理を実行する」→ MCP経由で決済システムの返金APIを発行。

    StackOne

のように役割を分担させることで、静的な知見の検索と動的なシステム運用を安全に両立できます。

StackOne

RAGの検索機能をMCP Serverとして実装する場合の具象的なアーキテクチャ例やPythonでのコードイメージを教えてください。

RAGの検索機能をMCP Serverとしてカプセル化(標準化)すると、Claude DesktopやCursor、自作のAIエージェントなど、あらゆるMCP Clientから標準規格で「社内文書検索ツール」として呼び出せるようになります。

全体アーキテクチャ例

以下は、最も標準的な「RAG via MCP Server」のシステム構成です。

Plaintext
┌─────────────────────────────────────────────────────────────┐
│ FastMCP (Python)                                            │
│                                                             │
│   @mcp.tool()                                               │
│   def search_knowledge_base(query: str, top_k: int) -> str  │
│      │                                                      │
│      ├─ 1. OpenAI (Text Embedding 3) でクエリをベクトル化      │
│      │                                                      │
│      └─ 2. ChromaDB / Qdrant から上位 K 件のテキストを取得     │
└──────────────────────────────┬──────────────────────────────┘
                               │ MCP Protocol (STDIO / SSE)
                               ▼
┌─────────────────────────────────────────────────────────────┐
│ MCP Client (Claude Desktop / Cursor / LangChain Agent)      │
└─────────────────────────────────────────────────────────────┘

Pythonコードイメージ(FastMCP + ChromaDB + OpenAI)

Pythonの公式ライブラリ mcp (または便利なラッパー fastmcp)を使うと、わずか数十行でRAGサーバーを作成できます。

1. 依存ライブラリのインストール

Bash
pip install mcp chromadb openai

2. rag_server.py の実装

Python
import os
from chromadb import PersistentClient
from fastmcp import FastMCP
from openai import OpenAI

# 1. 初期化
mcp = FastMCP("RAG-Knowledge-Base")
openai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
chroma_client = PersistentClient(path="./chroma_db")
collection = chroma_client.get_or_create_collection("company_docs")


# 2. 検索用のヘルパー関数
def get_embedding(text: str) -> list[float]:
    response = openai_client.embeddings.create(
        input=text, model="text-embedding-3-small"
    )
    return response.data[0].embedding


# 3. MCP Tool としてRAG検索機能を登録
@mcp.tool()
def search_knowledge_base(query: str, top_k: int = 3) -> str:
    """社内のドキュメント・規程データベースをベクトル検索します。

    Args:
        query: 検索したい質問やキーワード
        top_k: 取得する関連文書の数(デフォルト3)

    Returns:
        検索された関連ドキュメントのテキスト結合
    """
    # ユーザーのクエリをベクトル化
    query_vector = get_embedding(query)

    # ベクトルDBから類似文章を検索
    results = collection.query(query_embeddings=[query_vector], n_results=top_k)

    documents = results.get("documents", [[]])[0]

    if not documents:
        return "関連するドキュメントが見つかりませんでした。"

    # LLMが読みやすい形にフォーマットして返却
    formatted_results = []
    for i, doc in enumerate(documents, 1):
        formatted_results.append(f"--- 関連文書 {i} ---\n{doc}")

    return "\n\n".join(formatted_results)


if __name__ == "__main__":
    # Standard Input/Output 形式でMCPサーバーを起動
    mcp.run(transport="stdio")

ポイント解説

  1. @mcp.tool() デコレータとDocstring Docstring(社内のドキュメント・規程データベースを...)は、そのままLLM(MCP Client)への命令・説明文として送信されます。LLMは「ユーザーが社内規程について質問してきたら、このツールを呼び出そう」と自動判断します。

  2. stdio トランスポート mcp.run(transport="stdio") を指定することで、標準入出力経由でClaude DesktopやCursorなどのローカルクライアントから呼び出せるようになります(リモートサーバー化したい場合は sse トランスポートを使用します)。

  3. Claude Desktop 側の登録設定例 (claude_desktop_config.json)

    JSON
    {
      "mcpServers": {
        "company-rag": {
          "command": "python",
          "args": ["/path/to/rag_server.py"],
          "env": {
            "OPENAI_API_KEY": "sk-proj-..."
          }
        }
      }
    }
    

このように実装しておくことで、将来的にフロントエンドやLLMのモデルが変わったとしても、MCP Server側のコードは一切変更せずにそのまま使い回すことが可能になります。

この構築手順に関して、さらに深掘りしたい部分はありますか?

ChromaDBにPDFやMarkdownファイルを事前登録(インデックス化)するためのPythonインジェクションスクリプトの例を教えてください。

PDFやMarkdownなどのドキュメントを読み込み、テキストを適度なサイズに分割(チャンク化)してベクトル化し、ChromaDBに登録するためのPythonスクリプト例です。

実務でよく使われる pypdf(PDF解析)と langchain-text-splitters(テキスト分割)を組み合わせて実装します。

事前準備

必要なライブラリをインストールします。

Bash
pip install chromadb openai pypdf langchain-text-splitters

インジェクションスクリプト (ingest.py)

指定したディレクトリ内の .pdf.md を自動的にスキャンしてデータベース化します。

Python
import glob
import os
from pathlib import Path
from chromadb import PersistentClient
from langchain_text_splitters import RecursiveCharacterTextSplitter
from openai import OpenAI
from pypdf import PdfReader

# 1. 初期化設定
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")
DATA_DIR = "./documents"  # ドキュメントを置くフォルダ
DB_DIR = "./chroma_db"  # ChromaDBの保存先

openai_client = OpenAI(api_key=OPENAI_API_KEY)
chroma_client = PersistentClient(path=DB_DIR)
collection = chroma_client.get_or_create_collection(name="company_docs")


# 2. ファイル読み込み用関数
def load_text_from_file(file_path: str) -> str:
    ext = Path(file_path).suffix.lower()

    # Markdown / テキストファイル
    if ext in [".md", ".txt"]:
        with open(file_path, "r", encoding="utf-8") as f:
            return f.read()

    # PDFファイル
    elif ext == ".pdf":
        reader = PdfReader(file_path)
        text = ""
        for page in reader.pages:
            extracted = page.extract_text()
            if extracted:
                text += extracted + "\n"
        return text

    return ""


# 3. テキスト分割(チャンク化)関数
def split_text(text: str) -> list[str]:
    # 文脈を壊さないように段落・改行単位で分割
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,  # 1チャンクあたりの文字数目安
        chunk_overlap=50,  # 文脈をつなぐためのオーバーラップ幅
        separators=["\n\n", "\n", "。", " ", ""],
    )
    return text_splitter.split_text(text)


# 4. エンベディング生成関数
def get_embeddings(texts: list[str]) -> list[list[float]]:
    response = openai_client.embeddings.create(
        input=texts, model="text-embedding-3-small"
    )
    return [item.embedding for item in response.data]


# 5. メインインジェクション処理
def main():
    if not os.path.exists(DATA_DIR):
        os.makedirs(DATA_DIR)
        print(f"'{DATA_DIR}' ディレクトリを作成しました。ここにファイルを配置してください。")
        return

    # 対象ファイルを一括検索
    files = glob.glob(f"{DATA_DIR}/**/*", recursive=True)
    supported_files = [
        f for f in files if Path(f).suffix.lower() in [".pdf", ".md", ".txt"]
    ]

    if not supported_files:
        print(f"'{DATA_DIR}' 内に対象ファイル (.pdf, .md, .txt) が見つかりません。")
        return

    print(f"処理対象ファイル数: {len(supported_files)}")

    documents_batch = []
    metadatas_batch = []
    ids_batch = []

    for file_path in supported_files:
        filename = os.path.basename(file_path)
        print(f"読み込み中: {filename}")

        raw_text = load_text_from_file(file_path)
        if not raw_text.strip():
            print(f"  └ 空のファイルのためスキップします: {filename}")
            continue

        # チャンク分割
        chunks = split_text(raw_text)
        print(f"  └ チャンク数: {len(chunks)}")

        for idx, chunk in enumerate(chunks):
            chunk_id = f"{filename}_{idx}"

            documents_batch.append(chunk)
            metadatas_batch.append(
                {"source": filename, "chunk_index": idx, "file_path": file_path}
            )
            ids_batch.append(chunk_id)

    # まとめてEmbedding生成とChromaDB登録
    if documents_batch:
        print(f"\n合計 {len(documents_batch)} チャンクをベクトル化してChromaDBへ登録中...")

        # APIの制限を回避するため100件ずつバッチ処理
        batch_size = 100
        for i in range(0, len(documents_batch), batch_size):
            end_i = i + batch_size
            sub_docs = documents_batch[i:end_i]
            sub_metas = metadatas_batch[i:end_i]
            sub_ids = ids_batch[i:end_i]

            embeddings = get_embeddings(sub_docs)

            collection.add(
                documents=sub_docs,
                embeddings=embeddings,
                metadatas=sub_metas,
                ids=sub_ids,
            )

        print("インジェクション完了!登録されたDBは MCP Server からそのまま利用できます。")


if __name__ == "__main__":
    main()

実装の重要ポイント

  1. RecursiveCharacterTextSplitter の活用 単純に文字数で区切るのではなく、\n\n(段落)や (句点)など意味の切れ目を優先して分割します。これにより検索精度(精度と再現率)が大きく向上します。

  2. chunk_overlap の設定 前後のチャンク同士で文章を50文字程度重複させることで、文章が途中で分断された際の情報欠損(「その結果、〜」のような指示語の孤立など)を防ぎます。

  3. メタデータ (metadatas) の保持 ファイル名やパスをメタデータとして一緒に保持しておくことで、MCP Server側で回答を出力する際「〇〇マニュアル 3ページ目より引用」といった出展の提示が可能になります。

ChromaDBでデータを更新・削除する場合、主に「個別操作(ID/メタデータ指定)」「ファイルごとの一括置き換え(再インデックス化)」の2つのアプローチがあります。

特にRAG運用では、ドキュメントが更新された際に「旧バージョンの古いチャンクが残留して検索結果を汚染する」現象が起きやすいため、削除・更新の設計が重要になります。

1. 個別の更新・削除操作(基本API)

ChromaDBには直感的なメソッドがあらかじめ用意されています。

Python
from chromadb import PersistentClient

chroma_client = PersistentClient(path="./chroma_db")
collection = chroma_client.get_or_create_collection(name="company_docs")

# --------------------------------------------------
#  A. 削除 (Delete)
# --------------------------------------------------
# ① ID を指定して削除
collection.delete(ids=["sample_doc.pdf_0", "sample_doc.pdf_1"])

# ② メタデータで絞り込んで一括削除(特定のファイル由来のデータを全消し)
collection.delete(where={"source": "old_manual.pdf"})


# --------------------------------------------------
#  B. 更新・差し替え (Upsert)
# --------------------------------------------------
# upsert は「IDが存在しなければ追加、存在すれば上書き更新」を行う最も安全な操作です
new_text = "新しく改定された社内規程の本文です。"
new_vector = [0.12, -0.45, 0.88]  # 新テキストのベクトル

collection.upsert(
    ids=["rules.md_0"],
    documents=[new_text],
    embeddings=[new_vector],
    metadatas=[{"source": "rules.md", "chunk_index": 0}],
)

2. ファイル改定時の「完全再インデックス化」ロジック

ドキュメントを更新する際、テキストの長さが変わると「前回のインデックス時は 5 チャンクあったが、更新後は 3 チャンクに減った」というケースが発生します。この時に単純な upsert だけを行うと、余った rules.md_3rules.md_4 が削除されずに残り、古い情報と新しい情報が混在するハルシネーションの原因になります。

そのため、実務では「メタデータ削除 + 新規挿入」の手順を踏むのが定石です。

Python
def reindex_file(file_path: str, new_chunks: list[str], new_embeddings: list[list[float]]):
    filename = os.path.basename(file_path)

    # 1. 既存の同名ファイル由来チャンクをメタデータ指定で完全に全削除
    print(f"古いデータを削除中: {filename}")
    collection.delete(where={"source": filename})

    # 2. 新しく分割し直したチャンクを登録
    if not new_chunks:
        return

    ids = [f"{filename}_{i}" for i in range(len(new_chunks))]
    metadatas = [{"source": filename, "chunk_index": i} for i in range(len(new_chunks))]

    print(f"新しいデータ({len(new_chunks)} チャンク)を登録中: {filename}")
    collection.add(
        ids=ids,
        documents=new_chunks,
        embeddings=new_embeddings,
        metadatas=metadatas
    )

3. 効率的な運用パターン:ハッシュ値(MD5)による差分更新

巨大なドキュメント群を運用する場合、毎回全データを再インデックスするとOpenAIのエミッショントークンコストや計算時間が無駄になります。

実務では、ファイルの「MD5ハッシュ値」をメタデータに持たせ、変更があったファイルのみ処理する差分検出を実装します。

Python
import hashlib

def calculate_md5(file_path: str) -> str:
    """ファイルの変更検知用ハッシュ値を計算"""
    with open(file_path, "rb") as f:
        return hashlib.md5(f.read()).hexdigest()

def sync_document(file_path: str):
    filename = os.path.basename(file_path)
    current_hash = calculate_md5(file_path)

    # DB内の既存ハッシュ値を確認
    existing = collection.get(
        where={"source": filename},
        limit=1
    )

    # ハッシュ値が変わっていない場合は処理をスキップ
    if existing["metadatas"]:
        old_hash = existing["metadatas"][0].get("file_hash")
        if old_hash == current_hash:
            print(f"変更なし(スキップ): {filename}")
            return

    # 変更がある場合のみ削除&再登録
    print(f"更新検知(再インデックス開始): {filename}")
    collection.delete(where={"source": filename})
    
    # ... ここでテキスト抽出・チャンク化・Embedding生成を行って collection.add() ...

この「MD5ハッシュ差分チェック」を定期スクリプト(CronやFile Watcher)で回すことで、MCP Server経由で検索される社内データベースを常に最新・最小コストで維持できます。

ChromaDBのベクトル検索に加えて、キーワード検索(BM25など)を組み合わせたハイブリッド検索の仕組みと実装方針について教えてください。

単なるベクトル検索(Semantic Search)だけでは、「製品型番」や「固有名詞」「正確なコード表記」などのピンポイントなキーワード検索に弱いという弱点があります。これを補うのが、キーワード検索(BM25)とベクトル検索を組み合わせるハイブリッド検索(Hybrid Search)です。

ハイブリッド検索の仕組みとRRF(Reciprocal Rank Fusion)

2つの異なる検索アルゴリズムを組み合わせる際、スコアの尺度(尺度の違い)が異なるため、単純にスコアを足し合わせることはできません。そこで標準的に使われるのがRRF(Reciprocal Rank Fusion:逆順位融合)という順位ベースのリランキングアルゴリズムです。

$$\text{RRF\_Score}(d) = \sum_{m \in M} \frac{1}{k + r_m(d)}$$
  • : 検索手法の集合(ベクトル検索、BM25)

  • : 手法 におけるドキュメント の検索順位(1位なら 1)

  • : パラメータ(一般的に が推奨値)

順位のみを使ってスコア化するため、スコアの絶対値に依存せず安全かつ強力に両者の強みを統合できます。

Pythonによる実装方針例(ChromaDB + rank_bm25)

ChromaDB単体ではフルテキスト検索(BM25)機能が弱いため、Pythonライブラリ rank_bm25 をインメモリで組み合わせる軽量なハイブリッド検索の実装が一般的です。

1. 依存ライブラリのインストール

Bash
pip install chromadb rank-bm25 janome openai

※ 日本語テキストの形態素解析(単語分割)用に janome または mecab を使用します。

2. ハイブリッド検索クラスの実装例

Python
import os
from chromadb import PersistentClient
from janome.tokenizer import Tokenizer
from openai import OpenAI
from rank_bm25 import BM25Okapi

# 日本語の形態素解析器
tokenizer = Tokenizer()


def tokenize_ja(text: str) -> list[str]:
    """日本語テキストを単語(形態素)単位に分割"""
    return [token.surface for token in tokenizer.tokenize(text)]


class HybridRetriever:

    def __init__(self, chroma_collection, openai_client):
        self.collection = chroma_collection
        self.openai_client = openai_client

        # BM25用に全ドキュメントを取得してメモリ上に展開
        all_data = self.collection.get()
        self.doc_ids = all_data["ids"]
        self.documents = all_data["documents"]

        # BM25インデックスの初期化
        tokenized_corpus = [tokenize_ja(doc) for doc in self.documents]
        self.bm25 = BM25Okapi(tokenized_corpus)

    def search(self, query: str, top_k: int = 3, rrf_k: int = 60) -> list[str]:
        # 1. ベクトル検索を実行
        query_vector = (
            self.openai_client.embeddings.create(
                input=query, model="text-embedding-3-small"
            )
            .data[0]
            .embedding
        )

        vector_results = self.collection.query(
            query_embeddings=[query_vector], n_results=len(self.documents)
        )
        vector_doc_ids = vector_results["ids"][0]

        # 2. BM25検索を実行
        tokenized_query = tokenize_ja(query)
        bm25_scores = self.bm25.get_scores(tokenized_query)

        # BM25のスコア順にドキュメントIDを並び替え
        bm25_ranked_pairs = sorted(
            zip(self.doc_ids, bm25_scores), key=lambda x: x[1], reverse=True
        )
        bm25_doc_ids = [doc_id for doc_id, _ in bm25_ranked_pairs]

        # 3. RRF (Reciprocal Rank Fusion) による順位融合
        rrf_scores = {}

        # ベクトル検索の順位を加算
        for rank, doc_id in enumerate(vector_doc_ids, start=1):
            rrf_scores[doc_id] = rrf_scores.get(doc_id, 0) + (1.0 / (rrf_k + rank))

        # BM25検索の順位を加算
        for rank, doc_id in enumerate(bm25_doc_ids, start=1):
            rrf_scores[doc_id] = rrf_scores.get(doc_id, 0) + (1.0 / (rrf_k + rank))

        # RRFスコア順にソートしてTop Kを取得
        sorted_doc_ids = sorted(
            rrf_scores.keys(), key=lambda x: rrf_scores[x], reverse=True
        )[:top_k]

        # IDから対応する本文を取得して返却
        id_to_doc = dict(zip(self.doc_ids, self.documents))
        return [id_to_doc[doc_id] for doc_id in sorted_doc_ids]

実務での選択肢(スケールに応じた選定)

小〜中規模(数万件以下)であれば上記のような ChromaDB + rank_bm25 の構成で十分高速に動作します。しかし、数十万〜数百万件規模のエンタープライズ運用においては、ネイティブでハイブリッド検索とRRFをサポートしているプロダクション向けの検索エンジンを採用するのが一般的です。

  • Qdrant / Milvus: ハイブリッド検索とスパースベクトル(Sparse Vectors)をネイティブサポートするオープンソースのベクトルDB。

  • Elasticsearch / Azure AI Search: 伝統的な全文検索エンジンベースで、近年強力なベクトル検索機能を統合したマネージドサービス。

MCP Serverに組み込む際も、内部でこの HybridRetriever を呼び出すようにするだけで、クライアント側(LLM)からは透明性を保ったまま高精度な検索を実行できます。

検索エンジンの選定や高度なリランキングについて、さらに検討しますか?
Cohere Rerankなどのクロスコープモデルによるリランキングを学びたい
QdrantなどのマネージドベクトルDBでのハイブリッド検索実装を見たい

ChromaDBのベクトル検索に加えて、キーワード検索(BM25など)を組み合わせたハイブリッド検索の仕組みと実装方針について教えてください。

単なるベクトル検索(Semantic Search)だけでは、「製品型番」や「固有名詞」「正確なコード表記」などのピンポイントなキーワード検索に弱いという弱点があります。これを補うのが、キーワード検索(BM25)とベクトル検索を組み合わせるハイブリッド検索(Hybrid Search)です。

ハイブリッド検索の仕組みとRRF(Reciprocal Rank Fusion)

2つの異なる検索アルゴリズムを組み合わせる際、スコアの尺度(尺度の違い)が異なるため、単純にスコアを足し合わせることはできません。そこで標準的に使われるのがRRF(Reciprocal Rank Fusion:逆順位融合)という順位ベースのリランキングアルゴリズムです。

$$\text{RRF\_Score}(d) = \sum_{m \in M} \frac{1}{k + r_m(d)}$$
  • $M$: 検索手法の集合(ベクトル検索、BM25)

  • $r_m(d)$: 手法 $m$ におけるドキュメント $d$ の検索順位(1位なら 1)

  • $k$: パラメータ(一般的に $k = 60$ が推奨値)

順位のみを使ってスコア化するため、スコアの絶対値に依存せず安全かつ強力に両者の強みを統合できます。

Pythonによる実装方針例(ChromaDB + rank_bm25)

ChromaDB単体ではフルテキスト検索(BM25)機能が弱いため、Pythonライブラリ rank_bm25 をインメモリで組み合わせる軽量なハイブリッド検索の実装が一般的です。

1. 依存ライブラリのインストール

Bash
pip install chromadb rank-bm25 janome openai

※ 日本語テキストの形態素解析(単語分割)用に janome または mecab を使用します。

2. ハイブリッド検索クラスの実装例

Python
import os
from chromadb import PersistentClient
from janome.tokenizer import Tokenizer
from openai import OpenAI
from rank_bm25 import BM25Okapi

# 日本語の形態素解析器
tokenizer = Tokenizer()


def tokenize_ja(text: str) -> list[str]:
    """日本語テキストを単語(形態素)単位に分割"""
    return [token.surface for token in tokenizer.tokenize(text)]


class HybridRetriever:

    def __init__(self, chroma_collection, openai_client):
        self.collection = chroma_collection
        self.openai_client = openai_client

        # BM25用に全ドキュメントを取得してメモリ上に展開
        all_data = self.collection.get()
        self.doc_ids = all_data["ids"]
        self.documents = all_data["documents"]

        # BM25インデックスの初期化
        tokenized_corpus = [tokenize_ja(doc) for doc in self.documents]
        self.bm25 = BM25Okapi(tokenized_corpus)

    def search(self, query: str, top_k: int = 3, rrf_k: int = 60) -> list[str]:
        # 1. ベクトル検索を実行
        query_vector = (
            self.openai_client.embeddings.create(
                input=query, model="text-embedding-3-small"
            )
            .data[0]
            .embedding
        )

        vector_results = self.collection.query(
            query_embeddings=[query_vector], n_results=len(self.documents)
        )
        vector_doc_ids = vector_results["ids"][0]

        # 2. BM25検索を実行
        tokenized_query = tokenize_ja(query)
        bm25_scores = self.bm25.get_scores(tokenized_query)

        # BM25のスコア順にドキュメントIDを並び替え
        bm25_ranked_pairs = sorted(
            zip(self.doc_ids, bm25_scores), key=lambda x: x[1], reverse=True
        )
        bm25_doc_ids = [doc_id for doc_id, _ in bm25_ranked_pairs]

        # 3. RRF (Reciprocal Rank Fusion) による順位融合
        rrf_scores = {}

        # ベクトル検索の順位を加算
        for rank, doc_id in enumerate(vector_doc_ids, start=1):
            rrf_scores[doc_id] = rrf_scores.get(doc_id, 0) + (1.0 / (rrf_k + rank))

        # BM25検索の順位を加算
        for rank, doc_id in enumerate(bm25_doc_ids, start=1):
            rrf_scores[doc_id] = rrf_scores.get(doc_id, 0) + (1.0 / (rrf_k + rank))

        # RRFスコア順にソートしてTop Kを取得
        sorted_doc_ids = sorted(
            rrf_scores.keys(), key=lambda x: rrf_scores[x], reverse=True
        )[:top_k]

        # IDから対応する本文を取得して返却
        id_to_doc = dict(zip(self.doc_ids, self.documents))
        return [id_to_doc[doc_id] for doc_id in sorted_doc_ids]

実務での選択肢(スケールに応じた選定)

小〜中規模(数万件以下)であれば上記のような ChromaDB + rank_bm25 の構成で十分高速に動作します。しかし、数十万〜数百万件規模のエンタープライズ運用においては、ネイティブでハイブリッド検索とRRFをサポートしているプロダクション向けの検索エンジンを採用するのが一般的です。

  • Qdrant / Milvus: ハイブリッド検索とスパースベクトル(Sparse Vectors)をネイティブサポートするオープンソースのベクトルDB。

  • Elasticsearch / Azure AI Search: 伝統的な全文検索エンジンベースで、近年強力なベクトル検索機能を統合したマネージドサービス。

MCP Serverに組み込む際も、内部でこの HybridRetriever を呼び出すようにするだけで、クライアント側(LLM)からは透明性を保ったまま高精度な検索を実行できます。

検索エンジンの選定や高度なリランキングについて、さらに検討しますか?
Cohere Rerankなどのクロスコープモデルによるリランキングを学びたい
QdrantなどのマネージドベクトルDBでのハイブリッド検索実装を見たい

ChromaDBではなく、ネイティブでハイブリッド検索をサポートするQdrantを使った実装例を教えてください。

Qdrant は、密ベクトル(Dense Vector: 意味検索)とスパースベクトル(Sparse Vector: BM25/SPLADEなどのキーワード検索)の双方を同一コレクション内で保持し、データベースエンジン側で直接融合(Fusion)検索できるネイティブハイブリッド検索に対応しています。

Qdrantのハイブリッド検索の仕組み

Qdrantでは1つのポイント(ドキュメント)に対して、名前付きで複数のベクトルを持たせることができます。

  1. dense ベクトル: OpenAI (text-embedding-3-small など) で生成する高次元密ベクトル(意味の検索)

  2. sparse ベクトル: BM25 または SPLADE などで生成する疎ベクトル(キーワードの完全一致検索)

  3. PrefetchQueryFusion: Qdrant内部で両方の検索を並行実行し、RRF(Reciprocal Rank Fusion)やDBSF(Distribution-Based Score Fusion)でスコアを統合

Pythonによる実装例(fastembedqdrant-client

Qdrant公式が提供する軽量ライブラリ fastembed を使用すると、BM25(疎ベクトル)とOpenAI/SentenceTransformers(密ベクトル)の埋め込み生成およびRRF統合が非常にシンプルに記述できます。

1. 依存ライブラリのインストール

Bash
pip install qdrant-client fastembed openai

2. ハイブリッド検索実装コード (qdrant_hybrid.py)

Python
import os
from fastembed import SparseTextEmbedding
from openai import OpenAI
from qdrant_client import QdrantClient, models

# 1. クライアントの初期化
# ローカルのインメモリモードでテスト(本番環境では URL や Qdrant Cloud を指定)
qdrant = QdrantClient(":memory:")
openai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

# BM25同等の疎ベクトル生成モデル(日本語にも対応したBM25モデル)
sparse_model = SparseTextEmbedding(model_name="Qdrant/bm25")

COLLECTION_NAME = "hybrid_knowledge_base"

# 2. コレクションの作成(DenseとSparseの2種類のベクトルフィールドを定義)
qdrant.create_collection(
    collection_name=COLLECTION_NAME,
    vectors_config={
        # Denseベクトル(OpenAI: 1536次元)
        "dense": models.VectorParams(
            size=1536, distance=models.Distance.COSINE
        )
    },
    sparse_vectors_config={
        # Sparseベクトル(BM25キーワードインデックス)
        "sparse": models.SparseVectorParams(
            index=models.SparseIndexParams(on_disk=False)
        )
    },
)


# 3. ヘルパー関数: テキストからDense/Sparse双方のベクトルを生成
def generate_embeddings(texts: list[str]):
    # A. OpenAIでDenseベクトル(意味)を取得
    dense_resp = openai_client.embeddings.create(
        input=texts, model="text-embedding-3-small"
    )
    dense_vectors = [item.embedding for item in dense_resp.data]

    # B. FastEmbedでSparseベクトル(BM25キーワード)を取得
    sparse_vectors = list(sparse_model.embed(texts))

    return dense_vectors, sparse_vectors


# 4. ドキュメントの登録(インジェクション)
sample_docs = [
    "エラーコード ERR-9021 はデータベースのタイムアウトエラーを示します。",
    "システムのパフォーマンス低下時は、サーバーのCPU使用率とメモリ空き容量を確認してください。",
    "ユーザーパスワードの再設定手順:設定画面から「セキュリティ」を選択し、再発行ボタンを押します。",
]

dense_vecs, sparse_vecs = generate_embeddings(sample_docs)

points = []
for idx, (text, dense_vec, sparse_vec) in enumerate(
    zip(sample_docs, dense_vecs, sparse_vecs)
):
    points.append(
        models.PointStruct(
            id=idx,
            vector={
                "dense": dense_vec,
                "sparse": models.SparseVector(
                    indices=sparse_vec.indices.tolist(),
                    values=sparse_vec.values.tolist(),
                ),
            },
            payload={"text": text},
        )
    )

qdrant.upsert(collection_name=COLLECTION_NAME, points=points)
print(f"{len(sample_docs)} 件のドキュメントをインデックス化しました。\n")


# 5. ハイブリッド検索の実行(Qdrant内部でRRFで統合)
def hybrid_search(query_text: str, top_k: int = 2):
    # クエリのベクトル化
    query_dense_vec, query_sparse_vec = generate_embeddings([query_text])

    # Prefetch機能を使って Dense と Sparse の両方から上位候補を取得し、RRFでスコア統合
    results = qdrant.query_points(
        collection_name=COLLECTION_NAME,
        prefetch=[
            # 1. Denseベクトル検索の候補(上位10件)
            models.Prefetch(
                query=query_dense_vec[0],
                using="dense",
                limit=10,
            ),
            # 2. Sparse (BM25) ベクトル検索の候補(上位10件)
            models.Prefetch(
                query=models.SparseVector(
                    indices=query_sparse_vec[0].indices.tolist(),
                    values=query_sparse_vec[0].values.tolist(),
                ),
                using="sparse",
                limit=10,
            ),
        ],
        # RRF (Reciprocal Rank Fusion) アルゴリズムで融合
        query=models.FusionQuery(fusion=models.Fusion.RRF),
        limit=top_k,
    )

    return results.points


# --- 検索テスト ---
query = "ERR-9021"  # 固有名詞・型番などのキーワード
print(f"検索クエリ: '{query}'")
search_results = hybrid_search(query)

for point in search_results:
    print(
        f"Score: {point.score:.4f} | Text: {point.payload['text']}"
    )

Qdrantを採用する主なメリット

  1. Python側でのメモリ管理が不要 rank_bm25 のようにPython側で全文書をインメモリ保持する必要がなく、億単位のデータでもQdrant単体でスケーラブルに動作します。

  2. 高速な PrefetchFusion パイプライン データベースエンジンのRustコア層でDense/Sparseの並行取得とRRF演算が行われるため、ネットワーク往復のオーバーヘッドがなく超高速です。

  3. フィルタリングとの強力な併用 where={"tenant_id": "abc"} のようなメタデータフィルタを適用しながら、ハイブリッド検索を同時に実行できます。

Qdrantの構成やデプロイについて、さらに確認しますか?
Dockerを用いたローカル/本番環境でのQdrant構築手順を知りたい
QdrantをMCP Serverとして公開する実装例が見たい

Docker Composeを使ってQdrantサーバーを立ち上げ、永続化ストレージや管理画面(Dashboard)をセットアップする方法を教えてください。

Docker Composeを使うと、Qdrantサーバー本体とWeb UI管理画面(Dashboard)を1つのコマンドで簡単にローカルや本番環境へ立ち上げることができます。

Docker Compose構成のセットアップ

プロジェクトのルートディレクトリに docker-compose.yml を作成し、以下の構成を記述します。

1. docker-compose.yml

YAML
version: '3.8'

services:
  # Qdrant データベース本体
  qdrant:
    image: qdrant/qdrant:v1.10.0
    container_name: qdrant_server
    restart: always
    ports:
      - "6333:6333" # HTTP API & Web Dashboard
      - "6334:6334" # gRPC API (高速通信用)
    volumes:
      - qdrant_storage:/qdrant/storage # ベクトルデータ・インデックスの永続化
    environment:
      - QDRANT__SERVICE__ENABLE_STATIC_CONTENT=true # Qdrant組み込みのWeb Dashboardを有効化

  # Qdrant Web UI (より詳細なサードパーティ製オープンソースGUI管理画面)
  qdrant-ui:
    image: tooljet/qdrant-web-ui:latest # または docker.qdrant.tech/qdrant-web-ui
    container_name: qdrant_ui
    restart: always
    ports:
      - "3000:80" # ブラウザ用ポート
    environment:
      - QDRANT_URL=http://qdrant:6333
    depends_on:
      - qdrant

volumes:
  qdrant_storage:
    driver: local

コンテナの起動と確認手順

1. コンテナの起動

端末で docker-compose.yml が存在するディレクトリに移動し、バックグラウンドで起動します。

Bash
docker compose up -d

2. 管理画面(Web Dashboard)へのアクセス

Qdrantには「公式組み込みダッシュボード」「拡張UIコンテナ」の2種類の画面が利用できます。

  • 公式Web Dashboard (Qdrant内蔵)

    • URL: http://localhost:6333/dashboard

    • 特徴: コレクションの一覧確認、簡単なREST APIクエリの発行(Console機能)、クラスタ状態の監視が可能です。

  • 拡張Web UI (Qdrant Web UIコンテナ)

    • URL: http://localhost:3000

    • 特徴: ペイロードデータやベクトルの可視化・検索・編集が直感的なGUIで行えます。初回接続時に接続先URLを聞かれた場合は http://localhost:6333 を入力します。

Pythonコードからの接続テスト

コンテナ起動後、localhost:6333 を指定して既存の qdrant-client からそのまま接続できます。

Python
from qdrant_client import QdrantClient

# メモリモード (:memory:) から、起動したDockerコンテナへの接続に変更
client = QdrantClient(url="http://localhost:6333")

# コレクション一覧の取得
collections = client.get_collections()
print(collections)

本番運用に向けたセキュリティ設定(APIキー保護)

本番サーバーにデプロイする場合は、外部から勝手にデータを操作されないよう API KEY による認証を追加します。

docker-compose.ymlenvironment セクションに以下を追加するだけで保護されます。

YAML
    environment:
      - QDRANT__SERVICE__API_KEY=your-super-secret-api-key-here

この場合、Pythonクライアント側でも API KEY の指定が必要になります。

Python
client = QdrantClient(
    url="http://localhost:6333", api_key="your-super-secret-api-key-here"
)

このDockerで起動したQdrantのハイブリッド検索をFastMCPと組み合わせて、ClaudeやCursorから呼び出せるMCP Serverにするコード例を教えてください。

Dockerで起動したQdrant(http://localhost:6333)のハイブリッド検索を FastMCP でカプセル化し、Claude DesktopやCursorから直接呼び出せるMCP Serverの実装例です。

ディレクトリ構成

Plaintext
my-qdrant-mcp/
├── docker-compose.yml       # Qdrantサーバー(起動中)
├── mcp_qdrant_server.py     # FastMCP + Qdrant ハイブリッド検索サーバー
└── requirements.txt         # 依存ライブラリ

1. 依存ライブラリの準備 (requirements.txt)

Plaintext
mcp
fastmcp
qdrant-client
fastembed
openai
Bash
pip install -r requirements.txt

2. MCP Server の実装 (mcp_qdrant_server.py)

FastMCP を使い、LLMが「社内文書やFAQの検索が必要だ」と判断したときに自動呼び出しされる @mcp.tool() を定義します。

Python
import os
from fastembed import SparseTextEmbedding
from fastmcp import FastMCP
from openai import OpenAI
from qdrant_client import QdrantClient, models

# --------------------------------------------------
# 1. 初期化と接続設定
# --------------------------------------------------
mcp = FastMCP("Qdrant-Hybrid-RAG")

# Dockerコンテナで起動中の Qdrant に接続 (デフォルト: localhost:6333)
QDRANT_URL = os.environ.get("QDRANT_URL", "http://localhost:6333")
QDRANT_API_KEY = os.environ.get("QDRANT_API_KEY", None)

qdrant = QdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY)
openai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

# BM25同等の疎ベクトル生成モデル
sparse_model = SparseTextEmbedding(model_name="Qdrant/bm25")

COLLECTION_NAME = "hybrid_knowledge_base"


# --------------------------------------------------
# 2. ヘルパー関数:クエリのベクトル化
# --------------------------------------------------
def embed_query(query: str):
    # Dense (OpenAI)
    dense_resp = openai_client.embeddings.create(
        input=[query], model="text-embedding-3-small"
    )
    dense_vec = dense_resp.data[0].embedding

    # Sparse (BM25)
    sparse_vec = list(sparse_model.embed([query]))[0]

    return dense_vec, sparse_vec


# --------------------------------------------------
# 3. MCP ツール定義(LLMから直接呼び出される関数)
# --------------------------------------------------
@mcp.tool()
def search_knowledge_base(query: str, top_k: int = 3) -> str:
    """社内のドキュメント・システム仕様・FAQ・手順書をハイブリッド検索(意味検索+型番/エラーコード等のキーワード検索)します。

    Args:
        query: ユーザーからの質問や検索キーワード(例: "ERR-9021", "パスワード再設定の手順")
        top_k: 取得する関連文書の数(デフォルト3件)

    Returns:
        検索結果のテキスト情報
    """
    try:
        # クエリのベクトル化
        dense_vec, sparse_vec = embed_query(query)

        # Qdrant Prefetch + RRF ハイブリッド検索を実行
        search_results = qdrant.query_points(
            collection_name=COLLECTION_NAME,
            prefetch=[
                models.Prefetch(
                    query=dense_vec,
                    using="dense",
                    limit=10,
                ),
                models.Prefetch(
                    query=models.SparseVector(
                        indices=sparse_vec.indices.tolist(),
                        values=sparse_vec.values.tolist(),
                    ),
                    using="sparse",
                    limit=10,
                ),
            ],
            query=models.FusionQuery(fusion=models.Fusion.RRF),
            limit=top_k,
        )

        points = search_results.points

        if not points:
            return "該当するドキュメントが見つかりませんでした。"

        # LLMが解釈しやすいフォーマットに整形
        formatted_outputs = []
        for i, point in enumerate(points, start=1):
            text = point.payload.get("text", "")
            source = point.payload.get("source", "不明")
            score = point.score

            formatted_outputs.append(
                f"【検索結果 {i}】 (スコア: {score:.4f} / 出典: {source})\n{text}"
            )

        return "\n\n".join(formatted_outputs)

    except Exception as e:
        return f"検索中にエラーが発生しました: {str(e)}"


if __name__ == "__main__":
    # Standard I/O トランスポートでMCPサーバーを起動
    mcp.run(transport="stdio")

3. クライアント側の設定

Claude Desktop の設定

claude_desktop_config.json(Mac: ~/Library/Application Support/Claude/claude_desktop_config.json, Windows: %APPDATA%\Claude\claude_desktop_config.json)に以下を追加します。

JSON
{
  "mcpServers": {
    "qdrant-rag": {
      "command": "python",
      "args": ["/absolute/path/to/mcp_qdrant_server.py"],
      "env": {
        "OPENAI_API_KEY": "sk-proj-your-openai-api-key",
        "QDRANT_URL": "http://localhost:6333"
      }
    }
  }
}

Cursor の設定

  1. Cursor Settings > Features > MCP を開きます。

  2. + Add New MCP Server をクリックします。

    • Name: qdrant-rag

    • Type: command

    • Command: python /absolute/path/to/mcp_qdrant_server.py

動作フロー

  1. ユーザー: Claude や Cursor で「ERR-9021のエラー原因と対応策を教えて」と入力。

  2. LLM: ツールの説明文(Docstring)を見て、「search_knowledge_base ツールを使えば解決できる」と判断。

  3. MCP: ツールパラメータ query="ERR-9021" を渡して mcp_qdrant_server.py を実行。

  4. Qdrant: Dockerコンテナ内のQdrantが Dense(意味) + Sparse(キーワード) をPrefetch&RRF融合して最高精度の文書を取得。

  5. LLM: 得られたテキストをもとに、確実で正確な回答をユーザーに返答。

 

Neural Networks(ニューラルネットワーク)からLLMまでの核心概念を、雑学・業界話を交えて解説します。

この動画はCodistによる「100+ AI Concepts」シリーズのPart 01で、開発者向けに基礎を約20分でサクッとまとめたものです。以下、タイムスタンプ順に各トピックを深掘りしつつ、現場でよく出る話やトリビアを加えます。

Neural Networks(ニューラルネットワーク)

生物の神経細胞を簡略化した計算グラフ。入力層→隠れ層→出力層で、各接続に重み(weight)とバイアスがあり、活性化関数(ReLU、Sigmoidなど)で非線形性を入れます。

雑学・業界話

  • 1980〜90年代の「AIの冬」を抜けたきっかけの一つが、バックプロパゲーションの実用化とGPUの登場。
  • 現代の大規模モデルは「深い」だけでなく「幅も広い」ネットワークで、パラメータ数が数億〜数兆。
  • 実務では「ネットワークが深すぎると勾配消失」が昔からの定番問題で、ResNetのスキップ接続がその解決策として有名。

Training & Inference(学習と推論)

  • Training:データで重みを更新するプロセス(前向き計算+損失計算+逆伝播)。
  • Inference:学習済みモデルで予測だけするフェーズ。勾配計算を切るので高速・省メモリ。

雑学・業界話

  • 推論コストが学習コストを上回るケースが増えてきた(特にChatGPTのようなサービス)。そのため「推論最適化」が今のホットトピック。
  • 量子化(INT8/INT4)、スペキュラティブ・デコーディング、vLLMなどの高速化技術が業界標準になりつつある。
  • 「学習は一度、推論は何度も」なので、コスト構造がクラウド事業者の収益モデルを大きく変えている。

Tokenization(トークン化)

テキストを「トークン」という単位に分割する前処理。BPE(Byte Pair Encoding)やSentencePieceが主流。

雑学・業界話

  • GPT系は「単語」ではなく「サブワード」なので、「ChatGPT」が複数トークンに割れる。日本語は特にトークン効率が悪く、英語の1.5〜2倍くらいトークンを食うことが多い。
  • Tiktoken(OpenAI公式)やHugging Faceのtokenizerがデファクト。
  • 「トークン数=課金単位」なので、プロンプトエンジニアリングでは「同じ意味でトークンを減らす」が重要スキルになっている。

Embeddings(埋め込み)

トークンや文を高次元ベクトルに変換したもの。似た意味のものは近い位置に配置される。

雑学・業界話

  • Word2Vec(2013)が「王様 - 男 + 女 ≈ 女王」みたいなアナロジーで一躍有名に。
  • 今は「文埋め込み」や「マルチモーダル埋め込み」(CLIPなど)が主流。RAG(Retrieval-Augmented Generation)の基盤技術。
  • 次元数は512〜4096がよく使われるが、「次元の呪い」があるので、ただ大きくすれば良いわけではない。

Attention Mechanism(注意機構)

「どのトークンがどのトークンに注目すべきか」を学習する仕組み。Query・Key・Valueの3つで計算。

雑学・業界話

  • 2017年の「Attention is All You Need」論文がTransformerの起源。
  • Softmaxの計算量はシーケンス長の二乗なので、長いコンテキストで爆発する。これがFlashAttentionや線形注意などの改良を生んだ。
  • Multi-Head Attentionは「複数の視点で同時に見る」イメージ。

Transformers(トランスフォーマー)

Attentionを積み重ねたアーキテクチャ。Encoder-DecoderやDecoder-only(GPT系)がある。

雑学・業界話

  • 元論文の著者の多くがGoogle出身で、その後OpenAIやAnthropicなどに散っていった。
  • 「Transformerは本当に必要なのか?」という議論も時々出るが、現状ほぼすべてのLLMの基盤。
  • 位置情報は元々Positional Encodingだったが、今はRoPE(Rotary Position Embedding)が主流。

Context Window(コンテキストウィンドウ)

モデルが一度に処理できるトークン数の上限。

雑学・業界話

  • GPT-3は2048、GPT-4は8k→128k、Geminiは100万トークン超えと急速に拡大中。
  • 長いコンテキストは「針を干し草の中から探す」能力(Needle-in-a-Haystack)で評価されるが、実際は「真ん中の情報を忘れやすい」問題がある。
  • 実務では「全部突っ込まずに要約やRAGで絞る」方がコスト効率が良いケースが多い。

Scaling Laws(スケーリング則)

モデルサイズ・データ量・計算量が性能にどう影響するかを示した経験則。

雑学・業界話

  • OpenAIのKaplanら、DeepMindのChinchilla論文が有名。「パラメータを増やすよりデータを増やした方が効率的」というChinchilla則は大きな影響を与えた。
  • 今は「データ品質」や「合成データ」が次のボトルネックになっている。
  • 「ただ大きくすればいい」時代から「効率よくスケールさせる」時代へ移行中。

Pre-training vs. Post-training(事前学習 vs. 事後学習)

  • Pre-training:大量の生テキストで次トークン予測を学習(自己教師あり)。
  • Post-training:指示追従や安全性を高めるための追加学習(SFT、RLHF、DPOなど)。

雑学・業界話

  • RLHF(人間のフィードバックによる強化学習)はChatGPTの成功の鍵だったが、コストが高い。今はDPO(Direct Preference Optimization)などよりシンプルな手法が人気。
  • 「基盤モデルを作る会社」と「後段のアライメントを専門にする会社」に分かれる傾向がある。
  • オープンソース勢は「事前学習済みモデルを公開して、コミュニティが後段をやる」スタイルが強い。

Large Language Models(大規模言語モデル)

上記すべてを組み合わせた、数十億〜数兆パラメータのモデル群。

雑学・業界話

  • 「LLMはただの次トークン予測器」という言い方が正確だが、創発的能力(Emergent Abilities)が現れるのが面白いところ。
  • クローズド(GPT、Claude、Gemini)とオープン(Llama、Mistral、Qwenなど)の二極化が進んでいる。
  • 2024〜2025年頃から「エージェント化」「ツール使用」「長文推論」が次の焦点に。

これらの概念は相互に密接につながっています。特にAttention → Transformer → Scaling → LLMの流れが現代AIの骨格です。動画が「100+」と謳っているので、Part 02以降ではRAG、エージェント、評価指標、量子化、マルチモーダルなどが続く可能性が高いですね。

特定のトピックをさらに深掘りしたい、実装例が欲しい、業界の最新動向を知りたいなどがあれば教えてください。