Pular para o conteúdo principal
Última atualização em

Conecte agentes a dados não estruturados

Agentes de AI geralmente precisam consultar dados não estruturados, como coleções de documentos, bases de conhecimento ou corpora de texto, para responder a perguntas e fornecer respostas contextualmente relevantes.

Databricks oferece múltiplas abordagens para conectar agentes a dados não estruturados em índices de Pesquisa de AI e armazenamentos de vetores externos. Use servidores MCP pré-configurados para acesso imediato aos índices de Pesquisa de AI do Databricks, desenvolva ferramentas de recuperação localmente com pacotes AI Bridge ou crie funções de recuperação personalizadas para fluxos de trabalho especializados.

O Databricks AI Search era anteriormente conhecido como Databricks Vector Search. O prefixo de URL legado /api/2.0/mcp/vector-search/ continua a funcionar para compatibilidade com versões anteriores.

Consultar um índice de Pesquisa de AI do Databricks usando MCP​

Use o servidor MCP de Pesquisa de AI gerenciado pelo Databricks para dar ao seu agente acesso a um índice de Pesquisa de AI do Databricks. Primeiro, crie um índice usando incorporações gerenciadas pelo Databricks. Consulte Criar endpoints e índices de Pesquisa AI.

O URL MCP gerenciado para Pesquisa de AI é https://<workspace-hostname>/api/2.0/mcp/ai-search/{catalog}/{schema}/{index_name}. Conecte-se a ele e liste as ferramentas que ele expõe:

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

Para construir e implantar um agente que usa este servidor, consulte Usar servidores MCP em agentes. Conceda ao agente SELECT no objeto protegível do Unity Catalog do índice.

Outras abordagens​

Consultar um índice de pesquisa vetorial fora do Databricks

Consultar um índice de pesquisa vetorial hospedado fora do Databricks​

Se o seu índice vetorial estiver hospedado fora do Databricks, você pode criar uma conexão do Unity Catalog para se conectar ao serviço externo e usar a conexão em seu código de agente. Consulte Conectar agentes a ferramentas de terceiros com os Serviços MCP.

O exemplo a seguir cria um recuperador que chama um índice de vetor hospedado fora do Databricks para um agente no estilo PyFunc.

  1. Crie uma conexão do Unity Catalog com o serviço externo, neste caso, o Azure.

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

  2. Defina a ferramenta de recuperação no código do agente usando a conexão do Unity Catalog. Este exemplo usa decoradores do MLflow para habilitar o rastreamento do agente.

nota

Para estar em conformidade com o esquema do recuperador MLflow, a função do recuperador deve retornar um objeto List[Document] e usar o campo metadata na classe Document para adicionar atributos adicionais ao documento retornado, como doc_uri e similarity_score. Consulte o documento do MLflow.

Python
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(),
     &quot;Content-Type&quot;: &quot;application/json&quot;,
   },
      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

  1. Para executar o retriever, execute o seguinte código Python.

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

Desenvolver um mecanismo de recuperação local

Desenvolva um recuperador localmente usando o AI Bridge​

Para construir uma ferramenta de recuperação do Databricks AI Search localmente, use os pacotes do Databricks AI Bridge como databricks-langchain e databricks-openai. Esses pacotes incluem funções auxiliares como from_vector_search e from_uc_function para criar recuperadores a partir de recursos existentes do Databricks.

Instale a versão mais recente de databricks-langchain que inclui o Databricks AI Bridge.

Bash
%pip install --upgrade databricks-langchain

O código a seguir prototipa uma ferramenta de recuperação que consulta um índice hipotético de busca vetorial e o vincula a um LLM localmente para que seja possível testar seu comportamento de chamada de ferramenta.

Forneça um tool_description descritivo para ajudar o agente a entender a ferramenta e determinar quando invocá-la.

Python
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?")

Para cenários que usam índices de acesso direto ou índices de sincronização Delta com incorporações gerenciadas automaticamente, você deve configurar o VectorSearchRetrieverTool e especificar um modelo de incorporação personalizado e uma coluna de texto. Consulte opções para fornecer incorporações.

O exemplo a seguir mostra como configurar um VectorSearchRetrieverTool com a key columns e embedding.

Python
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={&quot;text_column LIKE&quot;: &quot;Databricks&quot;}, # 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.
)

Para detalhes adicionais, consulte a documentação da API para VectorSearchRetrieverTool.

Após sua ferramenta local estar pronta, você pode colocá-la diretamente em produção como parte do código do seu agente, ou migrá-la para uma função do Unity Catalog, que oferece melhor capacidade de descoberta e governança, mas tem certas limitações.

Consulte o Databricks AI Search usando funções do UC (obsoleto)

Consultar Databricks AI Search usando funções UC (obsoleto)​

nota

Databricks recommends MCP servers for most agent tools, but defining tools with Unity Catalog functions remains available for prototyping.

Você pode criar uma função do Unity Catalog que envolve uma consulta de Índice de Pesquisa de AI. Esta abordagem:

  • Suporta casos de uso de produção com governança e capacidade de descoberta.
  • Usa a função SQL vector_search() internamente
  • Suporta rastreamento automático do MLflow
    • Você deve alinhar a saída da função ao esquema de recuperador do MLflow usando os aliases page_content e metadata.
    • Quaisquer colunas de metadados adicionais devem ser adicionadas à coluna metadata usando a função de mapa SQL, em vez de como key de saída de nível superior.

Execute o seguinte código em um Notebook ou editor SQL para criar a função:

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

Para usar esta ferramenta de retriever em seu agente de AI, envolva-a com UCFunctionToolkit. Isso permite o rastreamento automático por meio do MLflow, gerando automaticamente RETRIEVER tipos de span nos log do MLflow.

Python
from unitycatalog.ai.langchain.toolkit import UCFunctionToolkit

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

As ferramentas de acesso do Unity Catalog têm as seguintes ressalvas:

  • Clientes SQL podem limitar o número máximo de linhas ou bytes retornados. Para evitar o truncamento de dados, trunque os valores das colunas retornados pela UDF. Por exemplo, é possível usar substring(chunked_text, 0, 8192) para reduzir o tamanho de colunas de conteúdo grandes e evitar o truncamento de linhas durante a execução.
  • Como esta ferramenta é um wrapper para a função vector_search(), ela está sujeita às mesmas limitações da função vector_search(). Consulte Limitações.

Para obter mais informações sobre UCFunctionToolkit, consulte a documentação do Unity Catalog.

Adicionar rastreamento a uma ferramenta de recuperador​

Adicione o rastreamento MLflow para monitorar e depurar seu recuperador. O rastreamento permite que você visualize entradas, saídas e metadados para cada etapa de execução.

O exemplo anterior adiciona o @mlflow.trace decorator tanto para o __call__ quanto para os métodos de análise. O decorador cria uma abrangência que começa quando a função é invocada e termina quando retorna. O MLflow registra automaticamente as entradas e saídas da função e quaisquer exceções geradas.

nota

Usuários de bibliotecas LangChain, LlamaIndex e OpenAI podem usar o registro automático do MLflow, além de definir rastreios manualmente com o decorador. Consulte Adicionar rastreios a aplicativos: rastreio automático e manual.

Python
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]:
    ...

Para verificar se aplicações downstream, como Agent Evaluation e AI Playground, renderizam o rastreamento do recuperador corretamente, certifique-se de que o decorador atenda aos seguintes requisitos:

  • Use o esquema de span do recuperador MLflow e verifique se a função retorna um objeto List[Document].
  • O nome do rastreamento e o nome retriever_schema devem corresponder para configurar o rastreamento corretamente. Consulte a seção a seguir para saber como definir o esquema do recuperador.

Defina o esquema do recuperador para verificar a compatibilidade com MLflow​

Se o rastreamento retornado do recuperador ou span_type="RETRIEVER" não estiver em conformidade com o esquema de recuperador padrão do MLflow, você deve mapear manualmente o esquema retornado para os campos esperados do MLflow. Isso verifica que o MLflow pode rastrear adequadamente seu recuperador e renderizar rastreamentos em aplicativos downstream.

Para definir o esquema do recuperador manualmente:

  1. Chame mlflow.models.set_retriever_schema ao definir seu agente. Use set_retriever_schema para mapear os nomes das colunas na tabela retornada para os campos esperados do MLflow, como primary_key, text_column e doc_uri.

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

  2. Especifique colunas adicionais no esquema do seu retriever, fornecendo uma lista de nomes de coluna com o campo other_columns.

  3. Se tiver vários recuperadores, é possível definir múltiplos esquemas usando nomes únicos para cada esquema de recuperador.

O esquema do retriever definido durante a criação do agente afeta os aplicativos e fluxos de trabalho downstream, como o aplicativo de revisão e os conjuntos de avaliação. Especificamente, a coluna doc_uri serve como o identificador principal para os documentos retornados pelo retriever.

  • O aplicativo de revisão exibe o doc_uri para ajudar os revisores a avaliar as respostas e rastrear as origens dos documentos. Consulte IU do aplicativo de revisão.
  • Conjuntos de avaliação usam doc_uri para comparar os resultados do recuperador com dataset de avaliação predefinidos para determinar o recall e a precisão do recuperador. Consulte Conjuntos de avaliação (MLflow 2).

Ler arquivos de um volume do Unity Catalog​

Se o seu agente precisar ler arquivos não estruturados (documentos de texto, relatórios, arquivos de configuração, etc.) armazenados em um volume do Unity Catalog, você poderá criar ferramentas que usam a API de Arquivos do SDK do Databricks para listar e ler arquivos diretamente.

Os exemplos a seguir criam duas ferramentas que seu agente pode usar:

  • list_volume_files : Lista arquivos e diretórios no volume.
  • read_volume_file ** **: Lê o conteúdo de um arquivo de texto do volume.

Instale a versão mais recente de databricks-langchain que inclui o Databricks AI Bridge.

Bash
%pip install --upgrade databricks-langchain
Python
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")

Vincule as ferramentas a um LLM e realize a execução de um loop de chamada de ferramenta:

Python
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)
Nesta página