Important
この機能は ベータ版です。 ワークスペース管理者は、[ プレビュー] ページからこの機能へのアクセスを制御できます。 Azure Databricks プレビューの管理を参照してください。
Databricks マネージド ベクター検索 MCP サーバーを使用する AI エージェントを構築する場合は、 _meta パラメーターを使用して、検索フィルターを返して適用する結果の最大数などのツールの動作を決定論的に制御します。
主な利点は、エージェントが動的に生成するための実際の検索クエリの柔軟性を維持しながら、検索動作 (結果の制限やフィルターなど) を事前設定できることです。
_meta パラメーターは、公式の MCP 仕様の一部です。
ツール呼び出し引数と _meta パラメーター
Vector Search MCP サーバーは、次の 2 つの方法でパラメーターを処理します。
- ツール呼び出し引数: 通常、ユーザー入力に基づいて LLM によって動的に生成されるパラメーター
-
_metaパラメーター: エージェント コードで事前設定して検索動作を確定的に制御できる構成パラメーター
サポートされている _meta パラメーター
ベクター検索では、次の _meta パラメーターがサポートされています。
| パラメーター名 | タイプ | Description |
|---|---|---|
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"指定しない場合、スコア フィルターは適用されません。 |
これらのパラメーターの詳細については、 Vector Search Python SDK のドキュメントを参照してください。
例: ベクター検索を取得するための最大結果とフィルターを構成する
この例では、 _meta パラメーターを使用してベクター検索の動作を制御し、AI エージェントからの動的クエリを許可する方法を示します。 公式の Python MCP SDK を使用します。
このシナリオでは、次の操作を行います。
- 常に検索結果を 3 つの項目に制限して、一貫した応答時間を実現する
- 関連性を確保するために、最近のドキュメント (2024-01-01 以降に更新) のみを検索します
- 純粋ベクター検索よりも精度を高めるハイブリッド検索を使用する
- 特定の列 (ID、テキスト、メタデータ) のみを返します
- 結果に類似性スコアを含める
- 類似性スコアが 0.5 未満の結果を除外する
- テキスト列とタイトル列の再ランク付けを使用して関連性を向上させる
この例を実行するには、 マネージド MCP 開発用に 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())