VOOZH about

URL: https://zenn.dev/umibows/articles/02eb4567a8bf0a

⇱ Claude MCPとVertex AIで遊戯王のルールを生成AIに理解させる


BigQuery
Model Context Protocol
Vertex AI
生成 AI
Claude
tech

LLM・LLM活用 Advent Calendar 2024 7日

ヤプリ #1 Advent Calendar 2024 9日

の記事です。

はじめにお断り

タイトルや本文で生成AIが「文脈を理解する」や「文脈を捉える」という表現を使っていますが、厳密には実行環境では行列計算で次の単語の出現確率を計算して予測しているのでAIが理解することはないです。ただ話として伝わりやすいようにそういう表現を使っています。

概要

遊戯王デュエルモンスターズというカードゲーム(以下遊戯王とする)があります。
そしてこのカードゲームのルールは非常に複雑です。
複雑さを増している要素として、日本語の意味、文法だけでは捉えられない遊戯王特有の文脈が存在するからです。
今回はこの遊戯王特有に文脈を生成AIに捉えさせることを試みました。

また、11月に発表されたModel Context Protocol(MCP)が話題になっていますが、
local sqliteファイルへのアクセスやgithubアクセスなどデフォルトで用意されているMCP Serverの設定の情報が中心で、cloud上のリソースへのアクセスを含んだCustom MCP Serverの実装例はあまりみません。
そこで、よくあるbigqueryで管理するデータ(テーブルやベクトル埋め込み)に対してアクセスする構成も試みました。

https://modelcontextprotocol.io/introduction

具体的な課題

日本語の意味、文法だけでは捉えられない遊戯王特有の文脈が存在するという話を冒頭で出しました。
具体例を示します。

『王の棺』『ティアラメンツ・シェイレーン』の2つのカードについてです。

https://www.db.yugioh-card.com/yugiohdb/card_search.action?ope=2&cid=17415&request_locale=ja

https://www.db.yugioh-card.com/yugiohdb/card_search.action?ope=2&cid=19200&request_locale=ja

『王の棺』の以下の効果

②:手札を1枚墓地へ送って発動できる。デッキから「ホルス」モンスター1体を墓地へ送る。

で手札の『ティアラメンツ・シェイレーン』を墓地に送った場合に、『ティアラメンツ・シェイレーン』の以下の効果

②:このカードが効果で墓地へ送られた場合に発動できる。(中略) 融合召喚する。

が発動するかどうかという問いを考えます。

日本語の意味や文法から判断すると、墓地に送られたのだから発動できそうに見えますが、これは遊戯王のルールでは発動しないというのが答えです。

遊戯王のルール上では、コストと効果という概念がありこれが日本語の文脈と異なります。
今回の場面だと、『ティアラメンツ・シェイレーン』はコストとして墓地送りされているので発動条件を満たせないということです。(これが結構初心者にとって難しい)。

当然何も情報を与えないclaude3.5 sonnetに以下のように投げると、発動するという答え(不正解)になります。
日本語の文脈だけを捉えればそれがむしろ自然に見えます。
生成AI(今回はclaude3.5 sonnet)にこの日本語とカードゲームのルールの文脈の違いを捉えさせて、正解を返すようにしていきます。

全体設計

以下の設計にしています。

前提知識としてMCPがどう動くかは以下で確認してください。
解説記事は他にたくさんあるのでここでは解説しません。

平たくいうと、Claudeに与えられたpromptに対してClaudeがどのMCP Serverにアクセスすれば良いのかを勝手に判断して対応するMCP ClientからMCP Serverに情報をとりにいくということを内部的にやってくれます。そのため、Claudeのユーザーは色々なところに情報アクセスをして頑張ってpromptを作成して生成AIに尋ねる、というプロセスが不要になります。
https://modelcontextprotocol.io/docs/concepts/architecture

まず、カード情報については高々13000種類程度しかないため、あらかじめBigQueryに入れてあります。ベクトル検索を可能にするため、ベクトル埋め込みをしています。手順は後述します。

また、BigQueryにアクセスするためのMCP Serverを用意してあります。
これは自前で作る必要があります。手順は後述します。

もう一つのMCP Serverでbrave search APIと接続し、Web検索をして関連するルールの情報を取得します。

全体の処理の流れとしては
ユーザーがClaude Desktop App上にルールに関する質問を入力
-> Claudeがカード情報をMCP Serverに問い合わせる
-> MCP Serverの処理の内部でBigQueryに関連カード情報を問い合わせる
-> BigQueryは内部的にVertex AIと接続して、ベクトル検索をし、関連情報をMCP Serverに返す
-> MCP ServerからClaudeにそのまま返す
-> Claudeは続いて、カード情報をもとにWeb検索するワードを決定する
-> MCP Serverを通じてbraveでWeb検索し、ルールに関係する情報を見つける
-> その情報がMCP ServerからClaudeに返される
-> Claudeが取得された情報をもとにユーザーから渡されたpromptに対する回答を生成
-> 出力

となります。

具体的な手順

BigQueryとVetex AIの接続

BigQueryのクエリからVertex AIを使ってベクトル埋め込みを利用するために、下準備をします。
以下の記事が非常に参考になったので、詳細は譲ります。
https://zenn.dev/cloud_ace/articles/bigquery-llm-rag

結果的に以下を実施しました。

  • 最初に外部接続を作成、bigqueryからvertexAIへの接続を作る。(これはコンソールから)
    • us.vertexaiというconnectionを作った。
  • datasetに属する以下を作成
    • cardsテーブル
    • card_embedding_model(embeddingするためのモデル)
    • card_embeddingsテーブル(カードデータのベクトル化したもの)

このときembedding model作る前にconnectionにattachされているservice accountにVertex AI userを付与する必要があるという注意がありました。このときembedding model作る前にconnectionにattachされているservice accountにVertex AI userを付与する必要があるという注意がありました。

基本的に貼った記事の通りでいけます。
最終的にこのような状態になりました。
👁 Image

正確なカード情報をBigQueryにためる

ここはあまり本質的でないので、概要だけ示します。
まず、公式のカード情報のページからスクレイピングして、13000種類のカードをjsonファイルとしてローカルに作成します。Denoで書きました。
その後、jsonファイルから以下のようにBigQueryのテーブルに入れていきました。

from google.cloud import bigquery

(中略)
self.client = bigquery.Client(project=project_id)
(中略)
errors = self.client.insert_rows_json(
 self.table_ref,
 batch,
 row_ids=[str(row['card_id']) for row in batch]
)
...

以下に一応動かしたコードは貼ります。
Claude3.5sonnetに生成してもらったもので、一部見せたくない情報は隠しています。
雰囲気だけ見てもらえればと思います。

ベクトル埋め込みする

BigQuery上のcardsテーブルからベクトル埋め込みします。
調べればたくさん情報は出てきます。
これも色々やり方がありますが、以下のClaudeが生成してくれたスクリプトで実施しました。

👁 Image

こんな感じでベクトルにしてくれます。
ちなみに今回embeddinに使ったモデルはtext-multilingual-embedding-preview-0409です。
これには特にこだわりがあるわけではなく、適宜最適なモデルを選ぶ必要があります。

MCP Serverのbaseを用意する

以下の公式documentとReadmeに従って作成しました。
documentの通り

uvx create-mcp-server

で大体の使われるmethodは勝手に作ってくれます。

https://modelcontextprotocol.io/docs/first-server/python
https://github.com/modelcontextprotocol/python-sdk

BigQueryに通信するMCP Serverを作る

ざっくりいうと、ボイラーテンプレートで用意しているhandle_call_tool関数が実際にプロンプトを処理するときに動く処理なので、ここにBigQueryのベクトル埋め込みをしたテーブルからベクトル検索して取得してくる処理を追加しました。

# server.py
(中略)
from .bigquery_client import BigqueryClient

server = Server("card_server")

project_id = "xxx"
dataset_id = "yugioh_dataset"
embedding_table = "card_embeddings"
location = "us-central1"


os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "/path/to/credential.json"
inquirer = BigqueryClient(
 project_id=project_id,
 dataset_id=dataset_id,
 location=location
)

(中略)

# ↓はボイラーテンプレートで作成される部分をname:get-cards, word:検索文字列で受けられるようにしている
@server.call_tool()
async def handle_call_tool(
 name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
 if name != "get-cards":
 raise ValueError(f"Unknown tool: {name}")

 if not arguments:
 raise ValueError("Missing arguments")

 word = arguments.get("word")
 if not word:
 raise ValueError("Missing word")

 data = await fetch_relative_cards(word)

 return [
 types.TextContent(
 type="text",
 text=json.dumps(data, indent=2),
 )
 ]

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
 """
 List available tools.
 Each tool specifies its arguments using JSON Schema validation.
 """
 return [
 types.Tool(
 name="get-cards",
 description="Get relative cards",
 inputSchema={
 "type": "object",
 "properties": {
 "word": {"type": "string"},
 },
 "required": ["word"],
 },
 )
 ]

# 新規追加
async def fetch_relative_cards(word: str) -> dict[str, any]:
 # inquirerを使ってカード情報を取得
 similar_cards = inquirer.get_similar_cards(word, embedding_table)

 # レスポンス形式に変換
 res = {"cards": []}
 for card in similar_cards:
 res["cards"].append({
 "name": f"card_{card.card_id}",
 "content": card.card_info,
 "similarity": card.similarity
 })

 return res

この設定でget-cardsというtoolをClaudeの方で認識してくれ、呼べる選択肢になります。
この場合、wordというベクトル検索文字列を受け取るようにしています。

実際にBigQueryにクエリを投げる部分は以下です。
実質的には質問文をそのままベクトル検索にかけ、コサイン距離が近いもの10個のカード情報を取得してきます。この処理自体は今までもよくやる手法でお馴染みだと思います。
最終的にはその10個のカードの情報をMCP接続を通してClaudeに渡します。

from dataclasses import dataclass
from typing import List
import os
from pathlib import Path
from langchain_google_vertexai import ChatVertexAI
from google.cloud import bigquery

@dataclass
class CardInfo:
 card_id: int
 card_info: str
 similarity: float

class BigqueryClient:
 def __init__(self, project_id: str, dataset_id: str, location: str = "us-central1"):
 """
 GeminiRAGInquirer初期化
 Args:
 project_id: GCPプロジェクトID
 dataset_id: BigQueryデータセットID
 location: VertexAI APIのロケーション
 """
 self.project_id = project_id
 self.dataset_id = dataset_id
 self.client = bigquery.Client(project=project_id)
 self.embedding_model = "embedding_model"

 def escape_string_for_sql(self, s: str) -> str:
 """SQLクエリ用に文字列をエスケープする"""
 # 文字列内の改行を空白に置換
 s = s.replace('\n', ' ')
 # バックスラッシュをエスケープ
 s = s.replace('\\', '\\\\')
 # シングルクォートをエスケープ
 s = s.replace("'", "\\'")
 # ダブルクォートをエスケープ
 s = s.replace('"', '\\"')
 return s

 def get_similar_cards(self, question: str, embedding_table: str, top_k: int = 10) -> List[CardInfo]:
 escaped_question = self.escape_string_for_sql(question)

 job_config = bigquery.QueryJobConfig(
 query_parameters=[
 bigquery.ScalarQueryParameter("question", "STRING", escaped_question),
 ]
 )

 query = f"""
 WITH 
 -- 質問文をベクトル化
 question_embedding AS (
 SELECT *
 FROM ML.GENERATE_TEXT_EMBEDDING(
 MODEL `{self.project_id}.{self.dataset_id}.{self.embedding_model}`,
 (SELECT @question AS content),
 STRUCT(TRUE AS flatten_json_output)
 )
 ),
 -- カード情報との類似度計算
 similarity AS (
 SELECT 
 card_id,
 card_info,
 ML.DISTANCE(q.text_embedding, c.embedding, 'COSINE') as vector_distance
 FROM question_embedding q
 CROSS JOIN `{self.project_id}.{self.dataset_id}.{embedding_table}` c
 )
 SELECT 
 card_id,
 card_info,
 ROUND(1 - vector_distance, 4) as similarity_score
 FROM similarity
 WHERE vector_distance < 1
 ORDER BY vector_distance
 LIMIT {top_k}
 """

 similar_cards = [
 CardInfo(
 card_id=row.card_id,
 card_info=row.card_info,
 similarity=row.similarity_score
 )
 for row in self.client.query(query, job_config=job_config)
 ]

 return similar_cards

(省略可能)inspectorからMCP Serverの動きを確認

MCP Serverの挙動を確認するためのinspectorが用意されています。
https://modelcontextprotocol.io/docs/tools/inspector

これを使わないと、いちいちコード変更時にClaude Desktop Appを再起動する必要があり面倒です。
また、仮にruntime error等をコンソール上で簡単に確認できるので挙動確認にはほぼ必須です。
http通信のPostmanみたいなものだと思います。

もし、本当に業務上で使うとなると、しっかり関数に対してテストコードを書いて担保するようになると思います。ここは今までと変わらないです。

👁 Image

実際こんな感じで、定義したget-cardsというtoolを確認でき、シェイレーンという文字列から目当ての『ティアラメンツ・シェイレーン』が一番コサイン距離の近い(内積最大)のレコードとして取得できました。(17415がKONAMIのサイトのcidに対応)。

これで、MCP ClientからMCP Serverへの疎通が正しくできるようになったことが確認できます。

Claude Desktop Appから呼び出せるように設定する

最後に、ClaudeからどのMCP Serverを使うかを設定して完了です。

https://modelcontextprotocol.io/docs/first-server/python#connect-to-claude-desktop

{
 "mcpServers": {
 "brave-search": {
 "command": "/Users/username/.anyenv/envs/nodenv/shims/npx",
 "args": [
 "-y",
 "@modelcontextprotocol/server-brave-search"
 ],
 "env": {
 "BRAVE_API_KEY": "xxx"
 }
 },
 "card_server": {
 "command": "uv",
 "args": [
 "--directory",
 "/Users/username/projects/mcp-playground/mcp-server-python",
 "run",
 "card-server"
 ]
 }
 }
}

のような設定をしています。
brave検索用のMCP Serverとcard情報取得用のMCP Serverへの接続を定義しています。
braveはAPI_KEYを取得する必要があります。

結果

肝心の結果です。


遊戯王デュエルモンスターズというカードゲームのルールに関する以下の質問に回答してください。
質問の中にはカード名が含まれる可能性が高いです。
カード名だと思われる言葉があれば、関連度の高いカード情報を取得してください。
その情報はカードの詳細情報が含まれているので参考にしてください。

また、日本語の文脈と異なる遊戯王独自の文脈があることに注意してください。
braveで遊戯王wikiの間違えやすいルールを検索して、ルールを詳細に把握してください。

また、回答する情報については根拠と理由を示してください。

質問: 『王の棺』の以下の効果
②:手札を1枚墓地へ送って発動できる。デッキから「ホルス」モンスター1体を墓地へ送る。
で手札の『ティアラメンツ・シェイレーン』を墓地に送った場合に、『ティアラメンツ・シェイレーン』の以下の効果
②:このカードが効果で墓地へ送られた場合に発動できる。(中略) 融合召喚する。
が発動するか?

というpromptを投げてみます。

すると、
👁 Image

のように、get-cards MCP Serverに繋いでベクトル検索した結果を取得できていることを確認できます。

👁 Image

そして、続いて結果をもとにbrave APIに繋いでいるMCP ServerにリクエストしてWeb検索をしていることを確認できます。

最終的に以下の回答を返します。

カードテキストとルール検索の結果から、この質問に対する回答を整理します。
重要な点は以下の通りです:

『王の棺』の効果


手札を1枚墓地へ送るのはコストとして発動する効果です
デッキから「ホルス」モンスターを墓地へ送るのが効果の処理です


『ティアラメンツ・シェイレーン』の効果


「このカードが効果で墓地へ送られた場合に発動できる」というのが発動条件です


遊戯王における「コスト」と「効果」の区別


コストは効果の発動のために必要な代価であり、効果処理の一部ではありません
コストで墓地へ送られた場合は「効果で墓地へ送られた」とはみなされません

結論:
『王の棺』の②の効果で手札の『ティアラメンツ・シェイレーン』を墓地へ送った場合、これはコストとして送られているため、『シェイレーン』の②の効果は発動できません。
理由:

『王の棺』の効果文で「手札を1枚墓地へ送って発動できる」という部分はコストを示しています
コストで墓地へ送られたカードは「効果で墓地へ送られた」とはみなされないというのが遊戯王のルールです
『シェイレーン』の発動条件である「効果で墓地へ送られた場合」を満たしていないため、効果を発動することはできません

このような判断は、遊戯王OCGにおける「コスト」と「効果」の区別に関する基本的なルールに基づいています。カードを墓地へ送る処理が効果なのかコストなのかを正確に判断することは、適切なカードプレイのために重要です。

おおー!ちゃんとコストと効果の区別を理解し、遊戯王のルールの文脈を把握できていることが確認できます。やりたいことを達成できました。

他にもいくつかルールに関係するpromptを試してみると筋の良い回答もありつつ、カード名の正確さや、考える対象の効果のテキストを明確にpromptで示すことなど、やはりpromptの書き方によって回答の質は大きく変わりそうでした。この点は今までと同じです。

まとめ

今回は日本語の文脈理解だけでは把握できないカードゲームのルールを題材にClaudeで正しい回答が買えるような仕組みをModel Context Protocol (MCP) + Vertex AI ベクトル検索 + BigQueryで実現しました。

この構成は実は社内の問い合わせのようなユースケースに有効です。
publicな情報はWeb検索しながら、社内でしか扱われないデータや社内独自の言葉、文脈を生成AIにinputすることができるからです。

これを実現するためには今まではベクトル検索した結果から手動or何かしらのプログラム処理等でpromptを自前で作成する必要がありました。
それを勝手にやってくれるClaudeのMCPは可能性を感じます。

現状主なHostはClaude Desktop Appと一部のツールですが、この先広がっていくとMCP Serverを実装できることがソフトウェアエンジニアのスキルの一部になる可能性も感じるのでこの辺りは追っていこうと思います。

Discussion

👁 Image