Connettere agenti a dati non strutturati

Gli agenti spesso devono eseguire query su dati non strutturati come raccolte di documenti, knowledge base o corpora di testo per rispondere alle domande e fornire risposte in grado di essere consapevoli del contesto.

Per i nuovi agenti Python, usa AI Bridge (consigliato) per interrogare direttamente gli indici di AI Search. Per le integrazioni MCP esistenti, vedi l'endpoint legacy workspace. Puoi anche collegarti a vector store esterni o leggere file dai volumi di Unity Catalog.

Sviluppa un retriever con AI Bridge (consigliato)

Per i nuovi agenti Python che interrogano gli indici di Azure Databricks AI Search, Azure Databricks consiglia AI Bridge. Usa VectorSearchRetrieverTool da databricks-langchain o databricks-openai per esporre un indice come strumento agente.

Prima di eseguire questi esempi, accedi al tuo workspace e scegli un indice AI Search che puoi interrogare. Sostituire il nome dell'indice di esempio con il suo nome in tre parti. I primi esempi utilizzano embedding gestiti di Azure Databricks e un endpoint modello che supporta la chiamata di strumenti.

Esegui i comandi di installazione in un terminale. In un notebook, usa %pip invece di pip e riavvia Python dopo l'installazione.

LangChain/LangGraph

Installa l'ultima versione di databricks-langchain e LangGraph:

pip install --upgrade databricks-langchain langgraph

Crea uno strumento di recupero e dagli una descrizione che dica all'agente quando cercare. Questo esempio testa direttamente il recupero, poi esegue un agente che può chiamare lo strumento e utilizzare i suoi risultati.

from databricks_langchain import VectorSearchRetrieverTool, ChatDatabricks
from langgraph.prebuilt import create_react_agent

vs_tool = VectorSearchRetrieverTool(
    index_name="catalog.schema.my_databricks_docs_index",
    tool_name="databricks_docs_retriever",
    tool_description="Search Databricks documentation to answer product questions.",
)
print(vs_tool.invoke("What is Databricks Agent Framework?"))

agent = create_react_agent(
    ChatDatabricks(endpoint="databricks-claude-sonnet-4-5"),
    tools=[vs_tool],
)
result = agent.invoke({"messages": [{
    "role": "user",
    "content": "Search the documentation and explain Databricks Agent Framework.",
}]})
for message in result["messages"]:
    print(message)

Controlla la conversazione stampata per una chiamata a uno strumento, documenti recuperati e una risposta basata su quei documenti.

Per gli scenari che usano indici di accesso diretto o indici di Sincronizzazione Delta tramite incorporamenti autogestiti, è necessario configurare VectorSearchRetrieverTool e specificare un modello di incorporamento personalizzato e una colonna di testo. Consultare le opzioni per fornire incorporazioni.

L'esempio seguente mostra come configurare un VectorSearchRetrieverTool con le chiavi columns e 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="databricks_docs_retriever", # 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.
)

Per altri dettagli, vedere la documentazione dell'API per VectorSearchRetrieverTool.

OpenAI

Installa l'ultima versione di databricks-openai:

pip install --upgrade databricks-openai

Questo esempio utilizza il client Azure Databricks compatibile con OpenAI. Esso esegue le chiamate agli strumenti del modello e restituisce i documenti recuperati al modello.

import json
from databricks_openai import DatabricksOpenAI, VectorSearchRetrieverTool

client = DatabricksOpenAI()
dbvs_tool = VectorSearchRetrieverTool(
    index_name="catalog.schema.my_databricks_docs_index",
    tool_name="databricks_docs_retriever",
    tool_description="Search Databricks documentation to answer product questions.",
)
messages = [{
    "role": "user",
    "content": "Search the documentation and explain Databricks Agent Framework.",
}]

for _ in range(10):
    response = client.chat.completions.create(
        model="databricks-claude-sonnet-4-5",
        messages=messages,
        tools=[dbvs_tool.tool],
    )
    message = response.choices[0].message
    messages.append(message.model_dump(exclude_none=True))
    if not message.tool_calls:
        print(message.content)
        break
    for call in message.tool_calls:
        if call.function.name != dbvs_tool.tool["function"]["name"]:
            raise ValueError(f"Unknown tool: {call.function.name}")
        args = json.loads(call.function.arguments)
        documents = dbvs_tool.execute(**args, openai_client=client)
        print(documents)
        messages.append({
            "role": "tool",
            "tool_call_id": call.id,
            "content": json.dumps(documents),
        })
else:
    raise RuntimeError("The agent reached the tool-calling limit.")

L'esempio seguente mostra come configurare un VectorSearchRetrieverTool con le chiavi columns e 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="databricks_docs_retriever", # 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.
)

Quando si utilizza un endpoint di embedding Azure Databricks, passa openai_client=client a vs_tool.execute(), come nel loop agente sopra. Usa lo stesso modello di embedding che ha creato i vettori indice.

Per altri dettagli, vedere la documentazione dell'API per VectorSearchRetrieverTool.

Dopo aver testato il retriever, aggiungilo al tuo agente personalizzato.

Consulta un indice di ricerca vettoriale al di fuori di Databricks

Se l'indice vettoriale è ospitato all'esterno di Azure Databricks, è possibile creare una connessione di Unity Catalog per connettersi al servizio esterno e usare la connessione nel codice dell'agente. Vedere Connettersi a servizi HTTP esterni.

Il seguente esempio interroga un indice Azure AI Search tramite una connessione HTTP. Serve una connessione con le autorizzazioni necessarie per interrogare quell'indice ed eseguire USE CONNECTION sulla connessione. Installa requests, databricks-sdk, e mlflow prima di eseguire l'esempio in Python.

Crea una connessione Unity Catalog con Azure AI Search usando la configurazione di connessione HTTP. Questo esempio utilizza un token bearer; scegli il metodo di autenticazione richiesto dalla tua distribuzione Azure AI Search.
```
CREATE CONNECTION azure_search_connection
TYPE HTTP
OPTIONS (
  host 'https://example.search.windows.net',
  base_path '/',
  bearer_token secret ('<secret-scope>','<secret-key>')
);
```

Definire lo strumento retriever nel codice dell'agente usando la connessione al catalogo Unity. Questo esempio usa i decoratori MLflow per abilitare il tracciamento dell'agente.

Note

Per essere conforme allo schema del retriever MLflow, la funzione retriever deve restituire un List[Document] oggetto e utilizzare il metadata campo nella classe Document per aggiungere altri attributi al documento restituito, ad esempio doc_uri e similarity_score. Vedere Documento MLflow.

import mlflow
import requests
from databricks.sdk import WorkspaceClient
from mlflow.entities import Document

class VectorSearchRetriever:
    def __init__(self, connection_name, index_name):
        self.workspace = WorkspaceClient()
        self.connection_name = connection_name
        self.index_name = index_name

    @mlflow.trace(span_type="RETRIEVER", name="vector_search")
    def __call__(self, query_vector: list[float], score_threshold: float = 0.0) -> list[Document]:
        response = requests.post(
            f"{self.workspace.config.host}/api/2.0/unity-catalog/connections/"
            f"{self.connection_name}/proxy/indexes/{self.index_name}/docs/search",
            params={"api-version": "2024-07-01"},
            headers=self.workspace.config.authenticate(),
            json={
                "select": "HotelId,HotelName,Description,Category",
                "vectorQueries": [{
                    "vector": query_vector,
                    "k": 7,
                    "fields": "DescriptionVector",
                    "kind": "vector",
                    "exhaustive": True,
                }],
            },
            timeout=30,
        )
        response.raise_for_status()
        return self.parse_results(response.json(), score_threshold)

    @mlflow.trace(span_type="PARSER")
    def parse_results(self, results, score_threshold) -> list[Document]:
        return [
            Document(
                page_content=item.get("Description", ""),
                id=item.get("HotelId"),
                metadata={
                    "similarity_score": item.get("@search.score", 0),
                    "HotelName": item.get("HotelName"),
                    "Category": item.get("Category"),
                },
            )
            for item in results.get("value", [])
            if item.get("@search.score", 0) >= score_threshold
        ]

Genera un embedding di query con lo stesso modello e le stesse dimensioni usate dal tuo indice. Salva l'array numerico come query-vector.json, poi esegui il retriever:

import json
from pathlib import Path

query_vector = json.loads(Path("query-vector.json").read_text())
retriever = VectorSearchRetriever("azure_search_connection", "hotels_vector_index")
results = retriever(query_vector, score_threshold=0.1)
for document in results:
    print(document.page_content)

Aggiungi il tracciamento a uno strumento retriever

Aggiungi il tracciamento MLflow per monitorare e eseguire il debug del retriever. La traccia consente di visualizzare input, output e metadati per ogni passaggio di esecuzione.

Nell'esempio precedente, il decorator @mlflow.trace viene aggiunto sia ai __call__ metodi che a quelli di analisi. Il decoratore crea un intervallo che inizia quando la funzione viene richiamata e termina al suo ritorno. MLflow registra automaticamente l'input e l'output della funzione ed eventuali eccezioni generate.

Note

Gli utenti della libreria LangChain, LlamaIndex e OpenAI possono usare la registrazione automatica di MLflow oltre a definire manualmente le tracce con l'elemento decorator. Consulta Panoramica del tracciamento.

Restituisci una lista di Document oggetti con page_content e metadata, come fa l'esempio sopra. Questo segue lo schema span del recuperatore MLflow, così gli strumenti di valutazione e revisione possono riconoscere i documenti recuperati.

Leggere i file da un volume del catalogo Unity

Se l'agente deve leggere file non strutturati (documenti di testo, report, file di configurazione e così via) archiviati in un volume di Catalogo Unity, è possibile creare strumenti che usano l'API File di Databricks SDK per elencare e leggere direttamente i file.

Scegli un volume esistente e concedi all'identità del tuo agente USE CATALOG, USE SCHEMA e READ VOLUME. Sostituisci <catalog>.<schema>.<volume> nel codice. Per il compito di esempio, usa un file di testo UTF-8 chiamato about_databricks.txt, oppure modifica il prompt indicando un file nel tuo volume.

Gli esempi seguenti creano due strumenti che l'agente può usare:

list_volume_files: elenca i file e le directory nel volume.
read_volume_file: legge il contenuto di un file di testo dal volume.

LangChain/LangGraph

Installare la versione più recente di databricks-langchain che include Databricks AI Bridge.

pip install --upgrade databricks-langchain langgraph

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

VOLUME = "<catalog>.<schema>.<volume>"
w = WorkspaceClient()


def volume_path(relative_path: str = "") -> str:
    path = PurePosixPath(relative_path)
    if path.is_absolute() or ".." in path.parts:
        raise ValueError("Use a path relative to the configured volume.")
    return str(PurePosixPath("/Volumes", *VOLUME.split(".")) / path)


@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."""
    path = volume_path(directory)
    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'."""
    full_path = volume_path(file_path)
    resp = w.files.download(full_path)
    return resp.contents.read().decode("utf-8")

Associare gli strumenti a un LLM ed eseguire un ciclo di chiamata degli strumenti:

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"]))

else:
    raise RuntimeError("The agent reached the tool-calling limit.")

print(response.content)

OpenAI

Installare la versione più recente di databricks-openai che include Databricks AI Bridge.

pip install --upgrade databricks-openai

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

VOLUME = "<catalog>.<schema>.<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 volume_path(relative_path: str = "") -> str:
    path = PurePosixPath(relative_path)
    if path.is_absolute() or ".." in path.parts:
        raise ValueError("Use a path relative to the configured volume.")
    return str(PurePosixPath("/Volumes", *VOLUME.split(".")) / path)


def execute_tool(name: str, args: dict) -> str:
    if name == "list_volume_files":
        directory = args.get("directory", "")
        path = volume_path(directory)
        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 = volume_path(args["file_path"])
        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."},
]

for _ in range(10):
    response = client.chat.completions.create(
        model="databricks-claude-sonnet-4-5", messages=messages, tools=tools
    )
    message = response.choices[0].message
    messages.append(message.model_dump(exclude_none=True))
    if not message.tool_calls:
        print(message.content)
        break
    for call in message.tool_calls:
        args = json.loads(call.function.arguments)
        result = execute_tool(call.function.name, args)
        messages.append({"role": "tool", "tool_call_id": call.id, "content": result})
else:
    raise RuntimeError("The agent reached the tool-calling limit.")

Consulta un indice di ricerca AI usando MCP (legacy)

Il server MCP di AI Search utilizza un endpoint MCP legacy di workspace. Richiede un indice con embedding gestiti da Azure Databricks. Vedere Creare endpoint e indici di ricerca di intelligenza artificiale.

L'URL MCP del workspace legacy per la ricerca AI è https://<workspace-hostname>/api/2.0/mcp/ai-search/{catalog}/{schema}/{index_name}. Connettiti e elenca gli strumenti che espone:

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

Per costruire e distribuire un agente che utilizza questo server, vedi Usa strumenti MCP in un agente Python. Concedere all'agente SELECT l'entità a protezione diretta del catalogo Unity dell'indice.

Ricerca di intelligenza artificiale era in precedenza Ricerca vettoriale. Il prefisso URL precedente /api/2.0/mcp/vector-search/ funziona ancora.

Interrogare Databricks AI Search tramite le funzioni UC (deprecate)

Note

Per i nuovi agenti Python, usa AI Bridge. Il seguente esempio è un riferimento per le integrazioni esistenti.

Puoi creare una funzione Unity Catalog che racchiuda una query indice AI Search. Questo approccio:

Supporta i casi d'uso di produzione con governance e facilità di individuazione
Usa la funzione SQL vector_search() sotto le quinte
Supporta il tracciamento automatico di MLflow
- È necessario allineare l'output della funzione allo schema del retriever MLflow usando gli page_content alias e metadata .
- Tutte le colonne di metadati aggiuntive devono essere aggiunte alla metadata colonna usando la funzione di mapping SQL, anziché come chiavi di output di primo livello.

Eseguire il codice seguente in un notebook o in un editor SQL per creare la funzione:

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
  )

Segui la configurazione dello strumento function tool, poi incapsula questo retriever con UCFunctionToolkit per usarlo nel tuo agente. Questo consente la tracciatura automatica tramite MLflow, generando automaticamente tipi di span nei log di MLflow.

from databricks_langchain import UCFunctionToolkit
from unitycatalog.ai.core.databricks import DatabricksFunctionClient

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

Gli strumenti di recupero del catalogo Unity presentano le avvertenze seguenti:

I client SQL potrebbero limitare il numero massimo di righe o byte restituiti. Per evitare il troncamento dei dati, troncate i valori delle colonne restituiti dalla UDF. Ad esempio, è possibile usare substring(chunked_text, 0, 8192) per ridurre le dimensioni delle colonne di contenuto di grandi dimensioni ed evitare il troncamento delle righe durante l'esecuzione.
Poiché questo strumento è un wrapper per la vector_search() funzione, è soggetto alle stesse limitazioni della vector_search() funzione. Vedere Limitazioni.

Per altre informazioni sui UCFunctionToolkit, vedere documentazione di Unity Catalog.

Schemi retriever personalizzati (legacy)

mlflow.models.set_retriever_schema è obsoleto. Per i nuovi retriever, restituire Document gli oggetti utilizzando lo schema standard dei retriever. La configurazione seguente riguarda integrazioni esistenti che restituiscono campi personalizzati.

Per impostare manualmente lo schema del retriever:

Chiamare mlflow.models.set_retriever_schema quando si definisce l'agente. Usare set_retriever_schema per mappare i nomi delle colonne nella tabella restituita ai campi previsti di MLflow, ad esempio primary_key, text_column e doc_uri.

import mlflow

# 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"],
)

Specificare colonne aggiuntive nello schema del retriever fornendo un elenco di nomi di colonna con il other_columns campo .
Se si dispone di più retriever, è possibile definire più schemi usando nomi univoci per ogni schema di recupero.

Lo schema del retriever impostato durante la creazione dell'agente influisce su applicazioni e flussi di lavoro downstream, ad esempio l'app di revisione e i set di valutazione. In particolare, la colonna doc_uri funge da identificatore primario per i documenti restituiti dal retriever.

L'app di revisione visualizza il doc_uri per aiutare i revisori a valutare le risposte e tracciare le origini dei documenti. Vedi Revisione dell'interfaccia utente dell'app.
I set di valutazione utilizzano doc_uri per confrontare i risultati del sistema di recupero con i set di dati di valutazione predefiniti, al fine di determinare il recupero e la precisione del sistema. Vedi riferimento del set di dati di valutazione.

Commenti e suggerimenti

Questa pagina è stata utile?

Ultimo aggiornamento: 2026-10-08

Connettere agenti a dati non strutturati

Sviluppa un retriever con AI Bridge (consigliato)

LangChain/LangGraph

OpenAI

Consulta un indice di ricerca vettoriale al di fuori di Databricks

Aggiungi il tracciamento a uno strumento retriever

Leggere i file da un volume del catalogo Unity

LangChain/LangGraph

OpenAI

Consulta un indice di ricerca AI usando MCP (legacy)

Interrogare Databricks AI Search tramite le funzioni UC (deprecate)

Schemi retriever personalizzati (legacy)

Commenti e suggerimenti

Risorse aggiuntive