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

Conectar agentes a dados não estruturados

Os agentes AI frequentemente 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 contextualizadas.

O Databricks oferece diversas abordagens para conectar agentes a dados não estruturados em índices de Busca Vetorial e armazenamentos vetoriais externos. Utilize servidores MCP pré-configurados para acesso imediato aos índices Databricks Vector Search, desenvolva ferramentas de recuperação localmente com o pacote AI Bridge ou crie funções de recuperação personalizadas para fluxos de trabalho especializados.

Consultar um índice de pesquisa vetorial do Databricks usando MCP​

Se o seu agente precisar consultar um índice Databricks Vector Search, use o servidor Databricks-gerenciar MCP. Antes de começar, crie um índice de pesquisa vetorial usando embeddings Databricks-gerenciar. Consulte Criar ponto de extremidade e índices de pesquisa vetorial.

O URL do gerenciador MCP para pesquisa vetorial é: https://<workspace-hostname>/api/2.0/mcp/vector-search/{catalog}/{schema}/{index_name}.

Os exemplos a seguir mostram como conectar seu agente a um índice do Vector Search. Substitua <catalog>, <schema> e <index-name> pelos nomes do seu índice de pesquisa vetorial.

Python
from agents import Agent, Runner
from databricks.sdk import WorkspaceClient
from databricks_openai.agents import McpServer

workspace_client = WorkspaceClient()

async with McpServer.from_vector_search(
    catalog="<catalog>",
    schema="<schema>",
    index_name="<index-name>",
    workspace_client=workspace_client,
    name="vector-search",
) as vs_server:
    agent = Agent(
        name="Research assistant",
        instructions="You are a research assistant. Use the vector search tool to find relevant documents and answer questions.",
        model="databricks-claude-sonnet-4-5",
        mcp_servers=[vs_server],
    )
    result = await Runner.run(agent, "What is the return policy?")
    print(result.final_output)

Conceda ao aplicativo acesso ao índice de pesquisa vetorial em databricks.yml:

YAML
resources:
  apps:
    my_agent_app:
      resources:
        - name: 'my_vector_index'
          uc_securable:
            securable_full_name: '<catalog>.<schema>.<index-name>'
            securable_type: 'TABLE'
            permission: 'SELECT'

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 com o Unity Catalog para se conectar ao serviço externo e usar essa conexão no código do seu agente. Consulte Conectar agentes a serviços externos.

O exemplo a seguir cria um recuperador que chama um índice vetorial hospedado fora do Databricks para um agente com o sabor 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 retriever no código do agente usando a conexão do Unity Catalog. Este exemplo usa decoradores do MLflow para habilitar o rastreamento de agentes.

nota

Para estar em conformidade com o esquema retriever do MLflow, a função retriever 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 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 Vector 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

  1. Para executar o retriever, execute o seguinte código Python. Opcionalmente, você pode incluir filtros de pesquisa vetorial na solicitação para filtrar os resultados.

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

Desenvolver um recuperador local

Desenvolva um recuperador localmente usando o AI Bridge.​

Para construir uma ferramenta de recuperação de pesquisa vetorial Databricks localmente, use o pacoteDatabricks 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 Databricks existentes.

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 cria um protótipo de uma ferramenta de recuperação que consulta um índice de pesquisa vetorial hipotético e o vincula a um LLM localmente para que o senhor possa testar o comportamento de chamada da 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 do Delta Sync usando incorporações autogerenciadas, o senhor deve configurar a opção VectorSearchRetrieverTool e especificar um modelo de incorporação personalizado e uma coluna de texto. Veja as opções para fornecer incorporações.

O exemplo a seguir mostra como configurar um VectorSearchRetrieverTool com as chaves 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={"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.
)

Para obter mais detalhes, consulte os documentos da API para VectorSearchRetrieverTool.

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

Consultar a Pesquisa Vetorial do Databricks usando funções UC (obsoleto)

Consultar a Pesquisa Vetorial do Databricks 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.

O senhor pode criar uma função do Unity Catalog que envolva uma consulta de índice do Mosaic AI Vector Search. Essa abordagem:

  • Oferece suporte a casos de uso de produção com governança e visibilidade
  • Usa a função SQL vector_search() por trás do capô
  • Oferece suporte ao rastreamento automático de MLflow
    • O senhor deve alinhar a saída da função ao esquema de recuperação do MLflow usando os aliases page_content e metadata.
    • Quaisquer colunas de metadados adicionais devem ser adicionadas à coluna metadata usando a função de mapaSQL, e não como chave de saída de nível superior.

Execute o código a seguir em um editor do Notebook ou 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 Vector Search index name here
    index => 'catalog.schema.databricks_docs_index',
    query => query,
    num_results => 5
  )

Para usar essa ferramenta de recuperação em seu agente AI, envolva-a com UCFunctionToolkit. Isso permite o rastreamento automático por meio do site MLflow, gerando automaticamente os tipos de extensão RETRIEVER em MLflow logs.

Python
from unitycatalog.ai.langchain.toolkit import UCFunctionToolkit

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

As ferramentas de recuperação do Unity Catalog têm as seguintes ressalvas:

  • Os 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, você pode 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 que a função vector_search() . Consulte as limitações.

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

Adicionar rastreamento a uma ferramenta de recuperação​

Adicione o rastreamento do MLflow para monitorar e depurar seu recuperador. O rastreamento permite que o senhor view entradas, saídas e metadados para cada etapa da execução.

O exemplo anterior adiciona o decorador @mlflow .trace aos métodos __call__ e de análise. O decorador cria um intervalo que começa quando a função é chamada e termina quando ela retorna. O MLflow registra automaticamente a entrada e a saída da função e todas as exceções levantadas.

nota

Usuários das bibliotecas LangChain, LlamaIndex e OpenAI podem usar o registro automático MLflow , além de definir manualmente os rastreamentos com o decorador. Consulte Adicionar rastreamentos a aplicativos: rastreamento 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 os aplicativos subsequentes, como o Agent Evaluation e o AI Playground, renderizam o rastreamento do recuperador corretamente, certifique-se de que o decorador atenda aos seguintes requisitos:

  • Use o esquema 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 coincidir para configurar o rastreamento corretamente. Consulte a seção a seguir para saber como definir o esquema do retriever.

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

Se o rastreamento retornado pelo recuperador ou span_type="RETRIEVER" não estiver em conformidade com o esquema de recuperação padrão do MLflow, você deverá mapear manualmente o esquema retornado para os campos esperados pelo MLflow. Isso verifica se o MLflow consegue rastrear corretamente seu recuperador e renderizar os rastreamentos em aplicativos subsequentes.

Para definir o esquema do recuperador manualmente:

  1. Chame mlflow.models.set_retriever_schema quando você define 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 colunas com o campo other_columns.

  3. Se você tiver vários recuperadores, poderá definir vários esquemas usando nomes exclusivos para cada esquema de recuperação.

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

  • O aplicativo de avaliação exibe o doc_uri para ajudar os revisores a avaliar as respostas e rastrear as origens dos documentos. Consulte Revisar a interface do usuário do aplicativo.
  • Os conjuntos de avaliação usam o site doc_uri para comparar os resultados do recuperador com o conjunto de dados de avaliação predefinido para determinar a recuperação e a precisão do recuperador. Consulte Conjuntos de avaliação (MLflow 2).

Ler arquivos de um volume 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 Unity Catalog , você pode criar ferramentas que utilizam a APIde Arquivos SDK Databricks para listar e ler arquivos diretamente.

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

  • list_volume_files : Lista os 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 execute um loop de chamada de ferramentas:

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)