メインコンテンツまでスキップ

Databricks Vector Search MCP サーバーのメタパラメーター

備考

ベータ版

この機能はベータ版です。ワークスペース管理者は、 プレビュー ページからこの機能へのアクセスを制御できます。「Databricks プレビューの管理」を参照してください。

DatabricksマネージドのVector Search MCP サーバーを使用するAIエージェントを構築する場合は、 _metaパラメーターを使用して、返される結果の最大数や検索フィルターの適用などのツールの動作を決定的に制御します。

主な利点は、エージェントが動的に生成できるように実際の検索クエリを柔軟に保ちながら、検索動作(結果の制限やフィルターなど)を事前に設定できることです。

_metaパラメーターは、公式のMCP 仕様の一部です。

ツール呼び出しの引数と_metaパラメーター

Vector Search MCP サーバーは次の 2 つの方法でパラメーターを処理します。

  • ツール呼び出し引数 : 通常、ユーザー入力に基づいてLLMによって動的に生成されるパラメーター
  • _metaパラメーター : 検索動作を決定的に制御するためにエージェント コードで事前設定できる構成のパラメーター

サポートされている_metaパラメーター

次の_metaパラメーターが、Vector Searchでサポートされています。

パラメーター名

Type

説明

columns

str

検索結果で返される列名のコンマ区切りリスト。

例: "id,text,metadata"

指定しない場合は、すべての列(「__」で始まる内部列を除く)が返されます。

columns_to_rerank

str

コンテンツが再ランキング モデルに送信される列名のカンマ区切りリスト。リランカーはこのコンテンツを使用してすべての検索結果のスコアを再調整し、関連性を向上させます。

例: "text,title,description"

指定しない場合は再ランク付けは実行されません。

filters

str

検索に適用するフィルターを含む JSON 文字列。有効な JSON 文字列形式である必要があります。

例: '{"updated_after": "2024-01-01"}'

指定しない場合はフィルターは適用されません。

include_score

bool

返される結果に類似度スコアを含めるかどうか。

サポートされている値: "true"または "false"

デフォルト: "false"

num_results

int

返される結果の数。

例: "5"

query_type

str

結果を取得するために使用する検索アルゴリズム。

サポートされている値: "ANN" (近似最近傍、デフォルト) または"HYBRID" (ベクトル検索とキーワード検索の組み合わせ)

デフォルト: "ANN"

score_threshold

float

結果をフィルタリングするための類似度スコアの最小しきい値。このしきい値を下回るスコアの結果は除外されます。

例: "0.7"

指定しない場合は、スコア フィルタリングは適用されません。

これらの問題の詳細については、「一斉検索Python SDKドキュメント」を参照してください。

例: ベクトル検索取得の最大結果とフィルターを構成する

この例では、 _metaを使用して、 AIエージェントからの動的クエリを許可しながら、待機検索の動作を制御する方法を示します。 公式のPython MCP SDK を使用します。

このシナリオでは、次のことを行います。

  • 応答時間を一定に保つために、検索結果を常に 3 項目に制限します。
  • 関連性を確保するため、最近のドキュメント(2024年1月1日以降に更新されたもの)のみを検索します。
  • ハイブリッド検索を使用すると、純粋なベクトル検索よりも精度が向上します。
  • 特定の列(ID、テキスト、メタデータ)のみを返す
  • 結果に類似度スコアを含める
  • 類似度スコアが0.5未満の結果を除外する
  • 関連性を高めるためにテキストとタイトルの列の再ランク付けを使用する

この例を実行するには、マネージド MCP 開発用の Python 環境をセットアップします

Python
# Import required libraries for MCP client and Databricks authentication
import asyncio
from databricks.sdk import WorkspaceClient
from databricks_mcp.oauth_provider import DatabricksOAuthClientProvider
from mcp.client.streamable_http import streamablehttp_client
from mcp.client.session import ClientSession
from mcp.types import CallToolRequest, CallToolResult

async def run_vector_search_tool_call_with_meta():
# Initialize Databricks workspace client for authentication
workspace_client = WorkspaceClient()

# Construct the MCP server URL for your specific catalog and schema
# Replace <workspace-hostname>, YOUR_CATALOG, and YOUR_SCHEMA with your values
mcp_server_url = "https://<workspace-hostname>/api/2.0/mcp/vector-search/YOUR_CATALOG/YOUR_SCHEMA"

# Establish connection to the MCP server with OAuth authentication
async with streamablehttp_client(
url=mcp_server_url,
auth=DatabricksOAuthClientProvider(workspace_client),
) as (read_stream, write_stream, _):

# Create an MCP session for making tool calls
async with ClientSession(read_stream, write_stream) as session:
# Initialize the session before making requests
await session.initialize()

# Create the tool call request with both dynamic and preset parameters
request = CallToolRequest(
method="tools/call",
params={
# Tool name follows the pattern: CATALOG__SCHEMA__INDEX_NAME
"name": "YOUR_CATALOG__YOUR_SCHEMA__YOUR_INDEX_NAME",

# Dynamic arguments - typically provided by your AI agent or user input
"arguments": {
"query": "How do I reset my password?" # This comes from your agent
},

# Meta parameters - preset configuration to control search behavior
"_meta": {
"num_results": "3", # Limit to 3 results for consistent performance
"filters": '{"updated_after": "2024-01-01"}', # JSON string for date filtering
"query_type": "HYBRID", # Use hybrid search for better relevance
"columns": "id,text,metadata", # Return only specific columns
"score_threshold": "0.5", # Filter out results with similarity score < 0.5
"include_score": "true", # Include similarity scores in results
"columns_to_rerank": "text,title" # Use reranker on these columns for better quality
}
}
)

# Send the request and get the response
response = await session.send_request(request, CallToolResult)
return response

# Execute the async function and get results
response = asyncio.run(run_vector_search_tool_call_with_meta())

次のステップ