AI検索

AI検索MCPサーバーはAzure Databricksが管理するMCPサーバーで、エージェントがAI検索インデックスに対してセマンティックサーチを実行し、関連する文書を検索できます。結果はUnity Catalogの権限で管理されます。 インデックスのクエリには Azure Databricks 管理の埋め込みが必要です。

AI検索 _meta パラメータ

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

これらのパラメーターの詳細については、AI Search Python SDK のドキュメントを参照してください。

例: AI 検索を取得するための最大結果とフィルターを構成する

この例では、_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/ai-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())