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 recuperador, 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 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.

Você pode criar uma função do Unity Catalog que encapsula uma consulta de índice do Vector Search . Esta 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)
Nesta página