Databricks 矢量搜索 MCP 服务器的元参数

重要

此功能在 Beta 版中。

生成使用 Databricks 托管矢量搜索 MCP 服务器的 AI 代理时,请使用 _meta 参数确定性地控制工具行为,例如返回和应用搜索筛选器的最大结果数。

关键好处是,你可以预设搜索行为(如结果限制和筛选器),同时使实际搜索查询灵活,使代理能够动态生成。

此参数 _meta 是官方 MCP 规范的一部分。

工具调用参数与 _meta 参数

矢量搜索 MCP 服务器通过两种方式处理参数:

  • 工具调用参数:通常由 LLM 基于用户输入动态生成的参数
  • _meta 参数:可以在代理代码中预设的配置参数,以确定性地控制搜索行为

支持 _meta 的参数

矢量搜索支持以下 _meta 参数:

参数名称 类型 Description
num_results int 要返回的结果数。
默认值:5
filters str 包含要应用于搜索的筛选器的 JSON 字符串。 必须是有效的 JSON 字符串格式。
示例: '{"updated_after": "2024-01-01"}'
query_type str 用于检索结果的搜索算法。
支持的值: "ANN" (近似最近的邻居、默认值)或 "HYBRID" (合并矢量和关键字搜索)
默认值:"ANN"

有关这些参数的详细信息,请参阅 矢量搜索 API 文档

示例:为矢量搜索检索配置最大结果和筛选器

此示例演示如何使用 _meta 参数来控制矢量搜索行为,同时允许来自 AI 代理的动态查询。 它使用官方 Python MCP SDK。

在这种情况下,您希望:

  • 始终将搜索结果限制为仅 3 个项目,以便获得一致的响应时间
  • 仅搜索最近的文档(在 2024-01-01 之后更新)以确保相关性
  • 混合搜索能够实现比纯矢量搜索更准确的结果

若要运行此示例, 请设置用于托管 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
                    }
                }
            )

            # 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())

后续步骤