Připojení agentů k nestrukturovaným datům

Agenti umělé inteligence často potřebují dotazovat nestrukturovaná data, jako jsou kolekce dokumentů, znalostní báze nebo textové korpory, aby mohli odpovídat na otázky a poskytovat odpovědi na kontext.

Databricks poskytuje několik přístupů pro připojení agentů k nestrukturovaným datům v indexech AI Search a externích vektorových úložištích. Pomocí předem nakonfigurovaných serverů MCP můžete získat okamžitý přístup k indexům AI Search v Databricks, lokálně vyvíjet nástroje retrieveru s balíčky AI Bridge nebo vytvářet vlastní funkce retrieveru pro specializované pracovní postupy.

Databricks AI Search se dříve označoval jako Databricks Vector Search. Předpona starší verze /api/2.0/mcp/vector-search/ adresy URL nadále funguje kvůli zpětné kompatibilitě.

Dotazování indexu vyhledávání AI Databricks pomocí MCP

Pomocí serveru AI Search MCP spravovaného službou Databricks získáte agentovi přístup k indexu vyhledávání AI Databricks. Nejprve vytvořte index pomocí vkládání spravovaných službou Databricks. Viz Vytvoření koncových bodů a indexů vyhledávání AI.

Spravovaná adresa URL MCP pro vyhledávání AI je https://<workspace-hostname>/api/2.0/mcp/ai-search/{catalog}/{schema}/{index_name}. Připojte se k němu a uveďte nástroje, které zveřejňuje:

from databricks.sdk import WorkspaceClient
from databricks_mcp import DatabricksMCPClient

workspace_client = WorkspaceClient()
host = workspace_client.config.host

mcp_client = DatabricksMCPClient(
    server_url=f"{host}/api/2.0/mcp/ai-search/<catalog>/<schema>/<index-name>",
    workspace_client=workspace_client,
)
tools = mcp_client.list_tools()

Pokud chcete sestavit a nasadit agenta, který používá tento server, přečtěte si téma Použití serverů MCP v agentech. Udělit agentovi SELECT oprávnění k zabezpečitelnému objektu Unity Catalog indexu.

Další přístupy

Dotazování indexu vektorového vyhledávání mimo Databricks

Dotazování indexu vektorového vyhledávání hostovaného mimo Databricks

Pokud je váš vektorový index hostovaný mimo Azure Databricks, můžete vytvořit připojení katalogu Unity pro připojení k externí službě a použít připojení v kódu agenta. Viz Připojení agentů k nástrojům třetích stran pomocí služeb MCP.

Následující příklad vytvoří retriever, který volá vektorový index hostovaný mimo Databricks pro agenta s příchutí PyFunc.

Vytvořte připojení katalogu Unity k externí službě, v tomto případě Azure.

CREATE CONNECTION ${connection_name}
TYPE HTTP
OPTIONS (
  host 'https://example.search.windows.net',
  base_path '/',
  bearer_token secret ('<secret-scope>','<secret-key>')
);

Definujte nástroj retrieveru v kódu agenta pomocí připojení katalogu Unity. Tento příklad používá dekorátory MLflow pro umožnění sledování agenta.

Note

Aby odpovídal schématu MLflow retriever, funkce retriever by měla vrátit List[Document] objekt a použít metadata pole ve třídě Document k přidání dalších atributů do vráceného dokumentu, například doc_uri a similarity_score. Viz dokument MLflow.

import mlflow
import json

from mlflow.entities import Document
from typing import List, Dict, Any
from dataclasses import asdict

class VectorSearchRetriever:
  """
  Class using Databricks AI Search to retrieve relevant documents.
  """

  def __init__(self):
    self.azure_search_index = "hotels_vector_index"

  @mlflow.trace(span_type="RETRIEVER", name="vector_search")
  def __call__(self, query_vector: List[Any], score_threshold=None) -> List[Document]:
    """
    Performs vector search to retrieve relevant chunks.
    Args:
      query: Search query.
      score_threshold: Score threshold to use for the query.

    Returns:
      List of retrieved Documents.
    """
    import requests
    from databricks.sdk import WorkspaceClient

    w = WorkspaceClient()
    json = {
      "count": true,
      "select": "HotelId, HotelName, Description, Category",
      "vectorQueries": [
        {
          "vector": query_vector,
          "k": 7,
          "fields": "DescriptionVector",
          "kind": "vector",
          "exhaustive": true,
        }
      ],
    }

    response = requests.post(
      f"{w.config.host}/api/2.0/unity-catalog/connections/{connection_name}/proxy/indexes/{self.azure_search_index}/docs/search?api-version=2023-07-01-Preview",
      headers={
        **w.config.authenticate(),
        "Content-Type": "application/json",
      },
      json=json,
    ).text

    documents = self.convert_vector_search_to_documents(response, score_threshold)
    return [asdict(doc) for doc in documents]

  @mlflow.trace(span_type="PARSER")
  def convert_vector_search_to_documents(
    self, vs_results, score_threshold
  ) -> List[Document]:
    docs = []

    for item in vs_results.get("value", []):
      score = item.get("@search.score", 0)

      if score >= score_threshold:
        metadata = {
          "score": score,
          "HotelName": item.get("HotelName"),
          "Category": item.get("Category"),
        }

        doc = Document(
          page_content=item.get("Description", ""),
          metadata=metadata,
          id=item.get("HotelId"),
        )
        docs.append(doc)

    return docs

Pokud chcete spustit načítač, spusťte následující Python kód.

retriever = VectorSearchRetriever()
query = [0.01944167, 0.0040178085 . . .  TRIMMED FOR BREVITY 010858015, -0.017496133]
results = retriever(query, score_threshold=0.1)

Vývoj místního retrieveru

Vyvinout vyhledávač lokálně pomocí AI Bridge

Chcete-li lokálně vytvořit nástroj retrieveru pro Databricks AI Search, použijte balíčky Databricks AI Bridge, například databricks-langchain a databricks-openai. Tyto balíčky zahrnují pomocné funkce jako from_vector_search a from_uc_function pro vytváření získávačů z existujících prostředků Databricks.

LangChain/LangGraph

Nainstalujte nejnovější verzi databricks-langchain, která obsahuje Databricks AI Bridge.

%pip install --upgrade databricks-langchain

Následující kód vytvoří prototyp nástroje retrieveru, který se dotazuje na hypotetický index vyhledávání vektorů a váže ho místně na LLM, abyste mohli otestovat jeho chování při volání nástrojů.

Zadejte popisný popis, tool_description který agentu pomůže pochopit nástroj a určit, kdy ho má vyvolat.

from databricks_langchain import VectorSearchRetrieverTool, ChatDatabricks

# Initialize the retriever tool.
vs_tool = VectorSearchRetrieverTool(
  index_name="catalog.schema.my_databricks_docs_index",
  tool_name="databricks_docs_retriever",
  tool_description="Retrieves information about Databricks products from official Databricks documentation."
)

# Run a query against the vector search index locally for testing
vs_tool.invoke("Databricks Agent Framework?")

# Bind the retriever tool to your Langchain LLM of choice
llm = ChatDatabricks(endpoint="databricks-claude-sonnet-4-5")
llm_with_tools = llm.bind_tools([vs_tool])

# Chat with your LLM to test the tool calling functionality
llm_with_tools.invoke("Based on the Databricks documentation, what is Databricks Agent Framework?")

Pro scénáře, které využívají buď indexy přímého přístupu, nebo indexy Delta Sync pomocí samostatně řízených vkládání, musíte nakonfigurovat VectorSearchRetrieverTool a zadat vlastní model pro vkládání a textový sloupec. Podívejte se na možnosti pro poskytování embeddingů.

Následující příklad ukazuje, jak nakonfigurovat VectorSearchRetrieverTool pomocí klíčů columns a embedding.

from databricks_langchain import VectorSearchRetrieverTool
from databricks_langchain import DatabricksEmbeddings

embedding_model = DatabricksEmbeddings(
    endpoint="databricks-bge-large-en",
)

vs_tool = VectorSearchRetrieverTool(
  index_name="catalog.schema.index_name", # Index name in the format 'catalog.schema.index'
  num_results=5, # Max number of documents to return
  columns=["primary_key", "text_column"], # List of columns to include in the search
  filters={"text_column LIKE": "Databricks"}, # Filters to apply to the query
  query_type="ANN", # Query type ("ANN" or "HYBRID").
  tool_name="name of the tool", # Used by the LLM to understand the purpose of the tool
  tool_description="Purpose of the tool", # Used by the LLM to understand the purpose of the tool
  text_column="text_column", # Specify text column for embeddings. Required for direct-access index or delta-sync index with self-managed embeddings.
  embedding=embedding_model # The embedding model. Required for direct-access index or delta-sync index with self-managed embeddings.
)

Další podrobnosti najdete v dokumentaci k rozhraní API pro VectorSearchRetrieverTool.

OpenAI

Nainstalujte nejnovější verzi databricks-openai, která obsahuje Databricks AI Bridge.

%pip install --upgrade databricks-openai

Následující kód vytvoří prototyp retrieveru, který se dotazuje na hypotetický index vektorového vyhledávání a integruje ho s modely GPT OpenAI.

Zadejte popisný popis, tool_description který agentu pomůže pochopit nástroj a určit, kdy ho má vyvolat.

Další informace o doporučeních OpenAI pro nástroje najdete v dokumentaci k volání funkcí OpenAI.

from databricks_openai import VectorSearchRetrieverTool
from openai import OpenAI
import json

# Initialize OpenAI client
client = OpenAI(api_key=<your_API_key>)

# Initialize the retriever tool
dbvs_tool = VectorSearchRetrieverTool(
  index_name="catalog.schema.my_databricks_docs_index",
  tool_name="databricks_docs_retriever",
  tool_description="Retrieves information about Databricks products from official Databricks documentation"
)

messages = [
  {"role": "system", "content": "You are a helpful assistant."},
  {
    "role": "user",
    "content": "Using the Databricks documentation, answer what is Spark?"
  }
]
first_response = client.chat.completions.create(
  model="gpt-4o",
  messages=messages,
  tools=[dbvs_tool.tool]
)

# Execute function code and parse the model's response and handle function calls.
tool_call = first_response.choices[0].message.tool_calls[0]
args = json.loads(tool_call.function.arguments)
result = dbvs_tool.execute(query=args["query"])  # For self-managed embeddings, optionally pass in openai_client=client

# Supply model with results – so it can incorporate them into its final response.
messages.append(first_response.choices[0].message)
messages.append({
  "role": "tool",
  "tool_call_id": tool_call.id,
  "content": json.dumps(result)
})
second_response = client.chat.completions.create(
  model="gpt-4o",
  messages=messages,
  tools=[dbvs_tool.tool]
)

Následující příklad ukazuje, jak nakonfigurovat VectorSearchRetrieverTool pomocí klíčů columns a embedding.

from databricks_openai import VectorSearchRetrieverTool

vs_tool = VectorSearchRetrieverTool(
    index_name="catalog.schema.index_name", # Index name in the format 'catalog.schema.index'
    num_results=5, # Max number of documents to return
    columns=["primary_key", "text_column"], # List of columns to include in the search
    filters={"text_column LIKE": "Databricks"}, # Filters to apply to the query
    query_type="ANN", # Query type ("ANN" or "HYBRID").
    tool_name="name of the tool", # Used by the LLM to understand the purpose of the tool
    tool_description="Purpose of the tool", # Used by the LLM to understand the purpose of the tool
    text_column="text_column", # Specify text column for embeddings. Required for direct-access index or delta-sync index with self-managed embeddings.
    embedding_model_name="databricks-bge-large-en" # The embedding model. Required for direct-access index or delta-sync index with self-managed embeddings.
)

Další podrobnosti najdete v dokumentaci k rozhraní API pro VectorSearchRetrieverTool.

Jakmile je místní nástroj připravený, můžete ho přímo productionizovat jako součást kódu agenta nebo ho migrovat do funkce Katalogu Unity, která poskytuje lepší zjistitelnost a zásady správného řízení, ale má určitá omezení.

Dotazování Databricks AI Search pomocí funkcí UC (zastaralé)

Note

Databricks doporučuje servery MCP pro většinu nástrojů agenta, ale definování nástrojů pomocí funkcí katalogu Unity zůstává k dispozici pro vytváření prototypů.

Můžete vytvořit funkci v katalogu Unity, která zapouzdřuje dotaz na index služby AI Search. Tento přístup:

Podporuje případy použití v produkčním prostředí se zásadami správného řízení a zjistitelností.
Používá funkci SQL vector_search() pod kapotou.
Podporuje automatické sledování MLflow.
- Výstup funkce musíte zarovnat ke schématu načítáče MLflow pomocí page_content aliasů a metadata aliasů.
- Všechny další sloupce metadat musí být přidány do metadata sloupce pomocí funkce mapování SQL, nikoli jako výstupní klíče nejvyšší úrovně.

Spuštěním následujícího kódu v poznámkovém bloku nebo editoru SQL vytvořte funkci:

CREATE OR REPLACE FUNCTION main.default.databricks_docs_vector_search (
  -- The agent uses this comment to determine how to generate the query string parameter.
  query STRING
  COMMENT 'The query string for searching Databricks documentation.'
) RETURNS TABLE
-- The agent uses this comment to determine when to call this tool. It describes the types of documents and information contained within the index.
COMMENT 'Executes a search on Databricks documentation to retrieve text documents most relevant to the input query.' RETURN
SELECT
  chunked_text as page_content,
  map('doc_uri', url, 'chunk_id', chunk_id) as metadata
FROM
  vector_search(
    -- Specify your AI Search index name here
    index => 'catalog.schema.databricks_docs_index',
    query => query,
    num_results => 5
  )

Pokud chcete tento nástroj pro vyhledávání použít v agentu AI, obalte jej UCFunctionToolkit. To umožňuje automatické trasování prostřednictvím MLflow automatickým generováním RETRIEVER typů rozsahů v protokolech MLflow.

from unitycatalog.ai.langchain.toolkit import UCFunctionToolkit

toolkit = UCFunctionToolkit(
    function_names=[
        "main.default.databricks_docs_vector_search"
    ]
)
tools = toolkit.tools

Nástroje pro načítání katalogu Unity mají následující upozornění:

Klienti SQL můžou omezit maximální počet vrácených řádků nebo bajtů. Pokud chcete zabránit zkrácení dat, zkraťte hodnoty sloupců vrácené uživatelsky definovanou funkcí. Pomocí substring(chunked_text, 0, 8192) můžete například zmenšit velikost velkých sloupců obsahu a vyhnout se zkrácení řádků během provádění.
Vzhledem k tomu, že tento nástroj je obálka funkce vector_search() , podléhá stejným omezením jako vector_search() funkce. Viz Omezení.

Další informace o UCFunctionToolkit naleznete v dokumentaci k katalogu Unity.

Přidání trasování do nástroje pro získávání dat

Přidejte trasování MLflow pro monitorování a ladění vyhledávače. Trasování umožňuje zobrazit vstupy, výstupy a metadata pro každý krok spuštění.

Předchozí příklad přidá dekorátor @mlflow.trace jak k metodám __call__, tak pro analýzu. Dekorátor vytvoří časový interval, který začíná, když je funkce vyvolána, a končí, když se vrátí. MLflow automaticky zaznamenává vstup a výstup funkce a všechny výjimky vyvolané.

Note

Uživatelé knihoven LangChain, LlamaIndex a OpenAI mohou vedle ručního definování trasování pomocí dekorátoru také používat automatické protokolování MLflow. Viz Přidání trasování do aplikací: automatické a ruční trasování.

import mlflow
from mlflow.entities import Document

# This code snippet has been truncated for brevity. See the full retriever example above.
class VectorSearchRetriever:
  ...

  # Create a RETRIEVER span. The span name must match the retriever schema name.
  @mlflow.trace(span_type="RETRIEVER", name="vector_search")
  def __call__(...) -> List[Document]:
    ...

  # Create a PARSER span.
  @mlflow.trace(span_type="PARSER")
  def parse_results(...) -> List[Document]:
    ...

Pokud chcete ověřit aplikace, které navazují, jako jsou Agent Evaluation a AI Playground, že správně zobrazují trasu, ujistěte se, že dekorátor splňuje následující požadavky:

Použijte schéma rozsahu načítáče MLflow a ověřte, že funkce vrací objekt List[Document].
Název trasování a název retriever_schema se musí shodovat, aby se trasování správně nakonfigurovalo. V následující části se dozvíte, jak nastavit schéma retrieveru.

Nastavení schématu retrieveru pro ověření kompatibility MLflow

Pokud trasování vrácené z retrieveru nebo span_type="RETRIEVER" neodpovídá schématu standardního načítáče MLflow, je nutné ručně namapovat vrácené schéma na očekávaná pole MLflow. To ověří, že MLflow dokáže správně trasovat váš vyhledávač a vykreslit trasování v následných aplikacích.

Ruční nastavení schématu retrieveru:

Zavolejte mlflow.models.set_retriever_schema při definování agenta. Slouží set_retriever_schema k mapování názvů sloupců ve vrácené tabulce na očekávaná pole MLflow, například primary_key, text_columna doc_uri.

# Define the retriever's schema by providing your column names
mlflow.models.set_retriever_schema(
  name="vector_search",
  primary_key="chunk_id",
  text_column="text_column",
  doc_uri="doc_uri"
  # other_columns=["column1", "column2"],
)

Zadejte další sloupce do schématu vašeho retrieveru zadáním seznamu názvů sloupců do pole other_columns.
Pokud máte více vyhledávačů, můžete definovat více schémat pomocí jedinečných názvů pro každé schéma vyhledávače.

Sada schématu retrieveru během vytváření agenta má vliv na podřízené aplikace a pracovní postupy, jako je například revizní aplikace a sady vyhodnocení. Konkrétně sloupec doc_uri slouží jako primární identifikátor dokumentů vrácených retrieverem.

Aplikace pro revizi zobrazí doc_uri, které pomáhá recenzentům vyhodnotit odpovědi a sledovat původ dokumentů. Viz Kontrola uživatelského rozhraní aplikace.
Sady vyhodnocení slouží doc_uri k porovnání výsledků retrieveru s předdefinovanými datovými sadami vyhodnocení k určení úplnosti a přesnosti retrieveru. Viz vyhodnocovací sady (MLflow 2).

Čtení souborů ze svazku katalogu Unity

Pokud váš agent potřebuje číst nestrukturované soubory (textové dokumenty, sestavy, konfigurační soubory atd.), které jsou uložené ve svazku katalogu Unity, můžete vytvořit nástroje, které k výpisu a čtení souborů přímo používají rozhraní API pro soubory sady Databricks SDK.

Následující příklady vytvoří dva nástroje, které může váš agent použít:

list_volume_files: Vypíše soubory a adresáře ve svazku.
read_volume_file: Načte obsah textového souboru ze svazku.

LangChain/LangGraph

Nainstalujte nejnovější verzi databricks-langchain, která obsahuje Databricks AI Bridge.

%pip install --upgrade databricks-langchain

from databricks.sdk import WorkspaceClient
from langchain_core.tools import tool

VOLUME = "<catalog>.<schema>.<volume>"  # TODO: Replace with your volume
w = WorkspaceClient()


@tool
def list_volume_files(directory: str = "") -> str:
    """Lists files and directories in the Unity Catalog volume.
    Provide a relative directory path, or leave empty to list the volume root."""
    base = f"/Volumes/{VOLUME.replace('.', '/')}"
    path = f"{base}/{directory.lstrip('/')}" if directory else base
    entries = []
    for f in w.files.list_directory_contents(path):
        kind = "dir" if f.is_directory else "file"
        size = f" ({f.file_size} bytes)" if not f.is_directory else ""
        entries.append(f"  [{kind}] {f.name}{size}")
    return "\n".join(entries) if entries else "No files found."


@tool
def read_volume_file(file_path: str) -> str:
    """Reads a text file from the Unity Catalog volume.
    Provide the path relative to the volume root, for example 'reports/q1_summary.txt'."""
    base = f"/Volumes/{VOLUME.replace('.', '/')}"
    full_path = f"{base}/{file_path.lstrip('/')}"
    resp = w.files.download(full_path)
    return resp.contents.read().decode("utf-8")

Vytvořte vazbu nástrojů na LLM a spusťte smyčku volání nástrojů:

from databricks_langchain import ChatDatabricks
from langchain_core.messages import HumanMessage, ToolMessage

llm = ChatDatabricks(endpoint="databricks-claude-sonnet-4-5")
llm_with_tools = llm.bind_tools([list_volume_files, read_volume_file])

messages = [HumanMessage(content="What files are in the volume? Can you read about_databricks.txt and summarize it in 2 sentences?")]
tool_map = {"list_volume_files": list_volume_files, "read_volume_file": read_volume_file}

for _ in range(5):  # max iterations
    response = llm_with_tools.invoke(messages)
    messages.append(response)
    if not response.tool_calls:
        break
    for tc in response.tool_calls:
        result = tool_map[tc["name"]].invoke(tc["args"])
        messages.append(ToolMessage(content=result, tool_call_id=tc["id"]))

print(response.content)

OpenAI

Nainstalujte nejnovější verzi databricks-openai, která obsahuje Databricks AI Bridge.

%pip install --upgrade databricks-openai

from databricks.sdk import WorkspaceClient
from databricks_openai import DatabricksOpenAI
import json

VOLUME = "<catalog>.<schema>.<volume>"  # TODO: Replace with your volume
w = WorkspaceClient()
client = DatabricksOpenAI()

# Define the tool specifications
tools = [
    {
        "type": "function",
        "function": {
            "name": "list_volume_files",
            "description": "Lists files and directories in the Unity Catalog volume. Provide a relative directory path, or leave empty to list the volume root.",
            "parameters": {
                "type": "object",
                "properties": {
                    "directory": {
                        "type": "string",
                        "description": "Relative directory path within the volume. Leave empty for root.",
                    }
                },
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "read_volume_file",
            "description": "Reads a text file from the Unity Catalog volume. Provide the path relative to the volume root, for example 'reports/q4_summary.txt'.",
            "parameters": {
                "type": "object",
                "properties": {
                    "file_path": {
                        "type": "string",
                        "description": "Path to the file relative to the volume root.",
                    }
                },
                "required": ["file_path"],
            },
        },
    },
]


def execute_tool(name: str, args: dict) -> str:
    base = f"/Volumes/{VOLUME.replace('.', '/')}"
    if name == "list_volume_files":
        directory = args.get("directory", "")
        path = f"{base}/{directory.lstrip('/')}" if directory else base
        entries = []
        for f in w.files.list_directory_contents(path):
            kind = "dir" if f.is_directory else "file"
            size = f" ({f.file_size} bytes)" if not f.is_directory else ""
            entries.append(f"[{kind}] {f.name}{size}")
        return "\n".join(entries) if entries else "No files found."
    elif name == "read_volume_file":
        full_path = f"{base}/{args['file_path'].lstrip('/')}"
        resp = w.files.download(full_path)
        return resp.contents.read().decode("utf-8")
    return f"Unknown tool: {name}"


# Call the model with tools
messages = [
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": "List the files in the volume, then read about_databricks.txt and summarize it."},
]

response = client.chat.completions.create(
    model="databricks-claude-sonnet-4-5", messages=messages, tools=tools
)

# Execute tool calls and send results back
while response.choices[0].finish_reason == "tool_calls":
    messages.append(response.choices[0].message)
    for tool_call in response.choices[0].message.tool_calls:
        args = json.loads(tool_call.function.arguments)
        result = execute_tool(tool_call.function.name, args)
        messages.append(
            {"role": "tool", "tool_call_id": tool_call.id, "content": result}
        )
    response = client.chat.completions.create(
        model="databricks-claude-sonnet-4-5", messages=messages, tools=tools
    )

print(response.choices[0].message.content)

Váš názor

Byla tato stránka užitečná?

Last updated on 2026-06-30

Připojení agentů k nestrukturovaným datům

Dotazování indexu vyhledávání AI Databricks pomocí MCP

Další přístupy

Dotazování indexu vektorového vyhledávání hostovaného mimo Databricks

Vyvinout vyhledávač lokálně pomocí AI Bridge

LangChain/LangGraph

OpenAI

Dotazování Databricks AI Search pomocí funkcí UC (zastaralé)

Přidání trasování do nástroje pro získávání dat

Nastavení schématu retrieveru pro ověření kompatibility MLflow

Čtení souborů ze svazku katalogu Unity

LangChain/LangGraph

OpenAI

Váš názor

Další materiály