Ahogrammer

Deep Dive Into NLP, ML and Cloud

Text Embeddings InferenceをBlackwell GPUで動かす

Hugging Faceが提供するText Embeddings Inferenceは、テキスト埋め込みモデルで高速な推論を提供するためのツール。久しぶりに使おうとしたところ、執筆時点では手元のBlackwellアーキテクチャGPUには正式対応していないことに気がついた。幸い、プルリクエストで対応が進んでいるのを見つけたため、試してみた。

イメージのビルド

まず、リポジトリをクローンしてプルリクエストのブランチをフェッチし、CUDAのcompute capabilityをBlackwell(12.0)向けに設定してDockerイメージをビルドする。

git clone https://github.com/huggingface/text-embeddings-inference.git  
git fetch origin pull/735/head:pr-735
git checkout pr-735

runtime_compute_cap=120                                                
docker build . -f Dockerfile-cuda --build-arg CUDA_COMPUTE_CAP=$runtime_compute_cap

コンテナの起動

ビルドしたイメージを使用して、Text Embeddings Inferenceサーバーを起動する。ビルド時にイメージ名を付け忘れたので、以下ではイメージIDを直接指定している。

model=Qwen/Qwen3-Embedding-0.6B
volume=$PWD/data

docker run --gpus all -p 8080:80 -v $volume:/data [イメージID] --model-id $model --auto-truncate

動作確認

正常に実行できているか確認するために、curlコマンドでリクエストを送る。

curl 127.0.0.1:8080/embed \
    -X POST \
    -d '{"inputs":"こんにちは"}' \            
    -H 'Content-Type: application/json'

正常に動作していれば、埋め込みが返される。

[[-0.015902974,-0.04719861,-0.012466182,-0.05194763,...

Pythonから利用する場合は、huggingface_hubやOpenAIのSDK経由でエンドポイントにアクセスできる。以下はhuggingface_hubを使った例。

from huggingface_hub import InferenceClient

client = InferenceClient()
embedding = client.feature_extraction(
    "こんにちは",
    model="http://localhost:8080/embed"
)
print(len(embedding[0]))
# 1024

バッチ推論をしたい場合は、文字列のリストを渡すことでできる。ただし、feature_extractionのドキュメント上は文字列を渡すことになっているので、将来的に変化する可能性があることに注意が必要。

embedding = client.feature_extraction(
    ["こんにちは", "こんばんは"],
    model="http://localhost:8080/embed"
)
print(len(embedding))
# 2

参考資料

Gemini APIの構成的関数呼び出しでマルチホップQAを試す

Gemini APIのドキュメントを読んでいたところ、「構成的関数呼び出し(compositional function calling)」という機能があることに気がついた。この機能を使うと、LLMが関数を順番に呼び出しながら、段階的に複雑な要求を処理できる。たとえば「現在地の気温」を尋ねられた場合、まず現在地を取得する関数を呼び出し、その結果を用いて天気を取得する関数を実行する、といった流れを自動で組み立ててくれる。

似たようなことはResponses APIでも実現できるが、Geminiの構成的関数呼び出しでは呼び出しの連鎖をより自動的かつシンプルに扱えるように見える。 つまり、アプリ側で明示的なループ制御や状態管理を書く必要がほとんどなく、少ないコードでマルチステップの推論を実現できる。

そこで今回は、この機能を使ってマルチホップQAを解くことを試みる。マルチホップQAとは、複数の情報源や文脈を段階的に推論して答えを導く質問応答のことである。たとえば「ルーヴル美術館が所在する都市の市長の名前」という質問なら、

  1. まず「ルーヴル美術館がある都市=パリ」であることを特定し、
  2. 次に「パリの市長」を調べる、

という二段階の推論が必要になる。ちなみに、この例は日本語のマルチホップ QA データセットであるJEMHopQAの論文から取っている。

この記事では、Geminiと検索API(Tavily)を組み合わせて、このようなマルチホップQAを簡単に実装できるかを試してみる。

準備

まず、必要なパッケージをインストールする。検索にはTavilyを使用する。

pip install -q google-genai tavily-python

実装

次に実装を示す。特別な設定はなく、search関数をツールとして登録しているだけだが、Geminiが自動的に関数呼び出しを行ってくれる。

import os

from google import genai
from google.genai import types
from tavily import TavilyClient

# TavilyClientの用意
tavily_client = TavilyClient(api_key="")


def search(query: str) -> str:
    """Perform a web search and return the relevant context."""
    print(f"クエリ: {query}")
    context = tavily_client.get_search_context(query=query)
    return context


# Geminiのクライアントを用意
client = genai.Client(api_key="")
config = types.GenerateContentConfig(
    tools=[search],
)

# リクエスト
content = "ルーヴル美術館が所在する都市の市長の名前は?"
response = client.models.generate_content(
    model="gemini-2.5-flash",
    contents=content,
    config=config,
)

# 出力の表示
print(f"回答 : {response.text}")

ルーヴル美術館が所在する都市の市長の名前は?」という質問を実行すると、Geminiは次のように関数を順に呼び出した。

クエリ: 都市 ルーヴル美術館
クエリ: パリ市長の名前
回答 : ルーヴル美術館が所在する都市、パリの市長はアンヌ・イダルゴ氏です。

最初に「ルーヴル美術館がある都市」を検索し、その結果「パリ」であると判断。次に「パリ市長の名前」を検索し、最終的に「アンヌ・イダルゴ」と導いている。Mayors of Europeによると、実行時点ではこの回答は正確だった。

まとめ

この記事では、構成的関数呼び出しを使ってマルチホップQAを解けるかを試した。今回の例のようにWeb検索をするだけなら、Responses API組み込みのweb_searchツールを使って実現することもできるが、自前の関数を呼び出すときに簡潔に実装できそうな点が魅力といえる。

参考資料

  1. コンポジション関数呼び出し | Gemini API
  2. JEMHopQA | GitHub
  3. JEMHopQA: 日本語マルチホップ QA データセットの改良
  4. tavily-python | GitHub

Deep Research用MCPサーバーの構築と独自データへの対応

2025年2月のリリース以来、Deep ResearchはChatGPTユーザーの間で広く利用されるようになった。私自身、6月ごろにLangGraphを使って独自に実装もしたが、その後の6月26日、OpenAIからDeep Research専用のモデル(o3-deep-research / o4-mini-deep-research)が公開された。このモデルを使えば、「検索 → 分析 → 執筆」という一連のリサーチフローを自動で実行するエージェントを、わずかなコードで構築できる。

Deep Research用モデルが参照できる検索対象は、次の3種類に分かれている。

  • Web検索:ChatGPTが利用するインターネット検索と同等
  • ファイル検索:自分のファイルを対象にする検索
  • リモートMCPサーバー:独自のデータソースやAPIを検索対象として統合できる仕組み

Web検索を使えばChatGPTのDeep Researchと似た体験が得られるが、MCPサーバーを利用すれば、自社データや独自の知識ベースを対象にしたDeep Researchを構築できる。この記事では、そのようなMCPサーバーの作成手順と動作確認までを実例を交えて紹介する。

MCPサーバーの実装

実装に入る前に、Deep Researchモデルが要求するMCPサーバーの仕様を確認しておく。

このモデルで利用するMCPサーバーは、「search」と「fetch」の2つのインターフェースを備えている必要がある。関連する情報をざっくり調べる場合にはsearch、その中から詳細を知りたい文書がある場合はfetchを使うと考えればよい。

  • search:クエリを受け取り、検索結果のリストを返す。
  • fetch:検索結果のIDを受け取り、対応する文書の内容を返す。

searchの出力仕様

各検索結果は「ID」「タイトル」「URL」の3要素を持ち、results配列の下に格納される。一般的な検索APIと似た構造になっている。

フィールド 必須 説明
results 配列 検索結果の配列。各要素は以下のオブジェクト構造を持つ。
id 文字列 ドキュメントまたは検索結果アイテムの一意なID。fetchで使用される。
title 文字列 人間が読めるタイトル。
url 文字列 引用用のURL。

fetchの出力仕様

fetchは、検索結果の詳細を返す役割を持つ。以下がそのフォーマットで、特に特別な点はない。

フィールド 必須 説明
id 文字列 ドキュメントまたは検索結果アイテムの一意なID。
title 文字列 ドキュメントやアイテムのタイトル。
text 文字列 ドキュメントやアイテムの全文。
url 文字列 ドキュメントのURL(引用・出典に使用)。
metadata オブジェクト 任意 ドキュメントに関する補足情報(キー/値のペア)。

実装

searchとfetchの仕様について理解したところで、MCPサーバーを実装していく。今回は、MCPサーバーの実装にFastMCPを、検索データベースにChroma DBを使用する。埋め込みモデルには軽量で日本語にも対応したRuri v3を採用した。Ruri v3は検索クエリと検索対象文書で埋め込み時のプレフィックスが異なる点に注意が必要だ。

from typing import Any

import chromadb
from chromadb import Documents, EmbeddingFunction, Embeddings
from fastmcp import FastMCP
from sentence_transformers import SentenceTransformer


class Ruriv3EmbeddingFunction(EmbeddingFunction):
    def __init__(
        self, model_name: str = "cl-nagoya/ruri-v3-30m", device: str = "cpu"
    ) -> None:
        self.model = SentenceTransformer(model_name, device=device)

    def __call__(self, input: Documents) -> Embeddings:
        docs = []
        for doc in input:
            if doc.startswith("検索クエリ: "):
                docs.append(doc)
            else:
                docs.append(f"検索文書: {doc}")
        return self.model.encode(docs).tolist()


server_instructions = """
This MCP server provides search and document retrieval capabilities
for chat and deep research connectors. Use the search tool to find relevant documents
based on keywords, then use the fetch tool to retrieve complete
document content with citations.
"""

# Create an MCP server
mcp = FastMCP("Demo", instructions=server_instructions)
client = chromadb.PersistentClient()
collection = client.get_collection(
    name="deep-research",
    embedding_function=Ruriv3EmbeddingFunction(),
)


@mcp.tool()
async def search(query: str) -> dict[str, list[dict[str, Any]]]:
    """Search for documents using vector store search.

    This tool searches through the vector store to find semantically relevant matches.
    Returns a list of search results with basic information. Use the fetch tool to get
    complete document content.

    Args:
        query: Search query string. Natural language queries work best for semantic search.

    Returns:
        Dictionary with 'results' key containing list of matching documents.
        Each result includes id, title, text snippet, and optional URL.
    """
    if not query or not query.strip():
        return {"results": []}

    items = collection.query(
        query_texts=[f"検索クエリ: {query}"],
        n_results=5,
    )
    results = []
    for id, text, metadatas in zip(
        items["ids"][0], items["documents"][0], items["metadatas"][0]
    ):
        result = {
            "id": id,
            "title": metadatas.get("chapter_name", "filename"),
            "text": text,
            "url": "https://www.oreilly.co.jp/books/9784814401154/",
        }
        results.append(result)

    return {"results": results}


@mcp.tool()
async def fetch(id: str) -> dict[str, Any]:
    """Retrieve complete document content by ID for detailed analysis and citation.

    This tool fetches the full document
    content from vector store. Use this after finding
    relevant documents with the search tool to get complete
    information for analysis and proper citation.

    Args:
        id: Local document ID

    Returns:
        Complete document with id, title, full text content,
        optional URL, and metadata
    """
    item = collection.get(ids=[id])
    result = {
        "id": id,
        "title": item["metadatas"][0].get("chapter_name", "filename"),
        "text": item["documents"][0],
        "url": "https://www.oreilly.co.jp/books/9784814401154/",
        "metadata": None,
    }

    return result


if __name__ == "__main__":
    mcp.run(transport="sse", host="0.0.0.0", port=8000)

MCP Inspectorを起動して、ツールを確認してみよう。

fastmcp dev server.py 

UI上からはツールとしてsearchとfetchがあることを確認できる。試しに、searchツールに対して「RAGとは?」というクエリを送ると、関連する検索結果が帰っていることを確認できた。

Deep Researchのテスト

MCPサーバーが動作することを確認したら、Deep Researchモデルから実際に利用してみる。ローカルのMCPサーバーには直接アクセスできないため、ngrokを使って公開URLを取得する。

ngrok http http://localhost:8000

Responses APIからでも動作は確認できるが、今回はプロンプトダッシュボードを利用して、Deep Research用モデルとMCPサーバーの連携を確認する。新規プロンプトを作成し、プロンプト設定で新しいMCPツールを追加する。ツールを追加する際には、承認不要の設定(Never)を選択する必要がある点に注意する。

ツールを追加できたら、開発者メッセージとクエリを入力して実行する。今回は「RAGについて教えて」というクエリを入力している。実行するとsearchとfetchを繰り返し呼び出し、数分で以下のようなレポートが完成する。

ちなみに、Responses APIを使う場合は以下のようにする。

from openai import OpenAI

client = OpenAI()

instructions = "<deep research instructions...>"

resp = client.responses.create(
    model="o4-mini-deep-research",
    background=True,
    reasoning={
        "summary": "auto",
    },
    tools=[
        {
            "type": "mcp",
            "server_label": "deep_research",
            "server_url": "https://mycompany.com/mcp",
            "require_approval": "never",
        },
    ],
    instructions=instructions,
    input="RAGについて教えて",
)

まとめ

本記事では、MCPサーバーを用いることで、Deep Researchを独自データに拡張できることを確認した。OpenAIのWeb検索を使う場合と比べると、MCPサーバーを構築する分だけコード量が増えるが、それでもこの程度のコードで、Deep Researchを実現できることは非常に大きい。実践的には、ユーザー意図の明確化やプロンプトの書き換えといったこともする必要があるが、それらについては参考資料を参照してもらいたい。

参考資料