Łączenie agentów z danymi bez struktury

Agenty często muszą przeszukiwać dane nieustrukturyzowane, takie jak kolekcje dokumentów, bazy wiedzy lub korpusy tekstowe, aby odpowiadać na pytania i udzielać odpowiedzi uwzględniających kontekst.

W przypadku nowych agentów Python użyj AI Bridge (zalecane), aby bezpośrednio wysyłać zapytania do indeksów AI Search. W przypadku istniejących integracji MCP zobacz punkt końcowy MCP w przestrzeni roboczej. Możesz też połączyć się z zewnętrznymi magazynami wektorowymi lub czytać pliki z woluminów Unity Catalog.

Rozwijaj retrievera z AI Bridge (zalecane)

Dla nowych agentów Python, którzy zapytują indeksy Azure Databricks AI Search, Azure Databricks zaleca AI Bridge. Użyj VectorSearchRetrieverTool z databricks-langchain lub databricks-openai, aby udostępnić indeks jako narzędzie agenta.

Przed uruchomieniem tych przykładów zaloguj się do swojego obszaru roboczego i wybierz indeks wyszukiwania AI, który możesz przeszukiwać. Zastąp przykładową nazwę indeksu jej trzyczęściową nazwą. Pierwsze przykłady wykorzystują zarządzane embeddingi Azure Databricks oraz endpoint modelu wspierający wywoływanie narzędzi.

Uruchom polecenia instalacyjne w terminalu. W notatniku używaj %pip zamiast pip i uruchom ponownie Python po instalacji.

LangChain/LangGraph

Zainstaluj najnowszą wersję databricks-langchain i LangGraph:

pip install --upgrade databricks-langchain langgraph

Stwórz narzędzie retrievera i daj mu opis, który informuje agenta, kiedy ma szukać. Ten przykład testuje pobieranie bezpośrednio, a następnie uruchamia agenta, który może wywołać narzędzie i wykorzystać jego wyniki.

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)

Sprawdź wydrukowaną rozmowę pod kątem wywołania narzędzia, pobranych dokumentów oraz odpowiedzi opartej na tych dokumentach.

W przypadku scenariuszy korzystających z indeksów bezpośredniego dostępu lub indeksów usługi Delta Sync przy użyciu samodzielnie zarządzanych osadzeń należy skonfigurować VectorSearchRetrieverTool i określić niestandardowy model osadzenia oraz kolumnę tekstową. Zobacz opcje zapewniania embedowań.

W poniższym przykładzie pokazano, jak skonfigurować element VectorSearchRetrieverTool z kluczami columns oraz 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.
)

Aby uzyskać dodatkowe informacje, zobacz dokumentację interfejsu API dla programu VectorSearchRetrieverTool.

OpenAI

Zainstaluj najnowszą wersję databricks-openai:

pip install --upgrade databricks-openai

Ten przykład wykorzystuje klienta Azure Databricks kompatybilnego z OpenAI. Wykonuje wywołania narzędzi modelu i zwraca pobrane dokumenty do modelu.

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

W przypadku scenariuszy korzystających z indeksów bezpośredniego dostępu lub indeksów usługi Delta Sync przy użyciu samodzielnie zarządzanych osadzeń należy skonfigurować VectorSearchRetrieverTool i określić niestandardowy model osadzenia oraz kolumnę tekstową. Zobacz opcje zapewniania embedowań.

W poniższym przykładzie pokazano, jak skonfigurować element VectorSearchRetrieverTool z kluczami columns oraz 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.
)

Używając punktu końcowego osadzenia Azure Databricks, przekaż openai_client=client do vs_tool.execute(), jak w pętli agenta powyżej. Użyj tego samego modelu osadzania, który stworzył wektory indeksowe.

Aby uzyskać dodatkowe informacje, zobacz dokumentację interfejsu API dla programu VectorSearchRetrieverTool.

Po przetestowaniu retrievera dodaj go do swojego agenta customowego.

Wykonywanie zapytań względem indeksu wyszukiwania wektorowego poza Databricks

Jeśli indeks wektorowy jest hostowany poza Azure Databricks, możesz utworzyć połączenie Unity Catalog, aby połączyć się z usługą zewnętrzną i użyć tego połączenia w kodzie agenta. Zobacz Łączenie z zewnętrznymi usługami HTTP.

Poniższy przykład zapytuje indeks Wyszukiwanie AI platformy Azure za pomocą połączenia HTTP. Potrzebujesz połączenia z uprawnieniami do wykonywania zapytań względem tego indeksu i USE CONNECTION w ramach tego połączenia. Zainstaluj requests, databricks-sdk i mlflow przed uruchomieniem przykładu w języku Python.

  1. Stwórz połączenie Unity Catalog z Wyszukiwanie AI platformy Azure za pomocą konfiguracji połączenia HTTP. Ten przykład używa tokenu okaziciela; wybierz metodę uwierzytelniania, której wymaga wdrożenie Wyszukiwanie AI platformy Azure.

    CREATE CONNECTION azure_search_connection
    TYPE HTTP
    OPTIONS (
      host 'https://example.search.windows.net',
      base_path '/',
      bearer_token secret ('<secret-scope>','<secret-key>')
    );
    
  2. Zdefiniuj narzędzie wyszukiwania w kodzie agenta przy użyciu połączenia katalogu Unity. W tym przykładzie użyto dekoratorów MLflow do włączenia śledzenia działań agentów.

    Note

    Aby zachować zgodność ze schematem pobierania MLflow, funkcja pobierająca powinna zwrócić List[Document] obiekt i użyć metadata pola w klasie Document, aby dodać atrybuty do zwróconego dokumentu, na przykład doc_uri i similarity_score. Zobacz dokument 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
            ]
    
  3. Wygeneruj osadzenie zapytania z tym samym modelem i wymiarami używanymi przez Twój indeks. Zapisz tablicę liczbową jako query-vector.json, a następnie uruchom moduł pobierający:

    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)
    

Dodaj śledzenie do narzędzia retrievera

Dodaj śledzenie MLflow, aby monitorować i debugować swojego retrievera. Śledzenie umożliwia wyświetlanie danych wejściowych, wyjściowych i metadanych dla każdego kroku wykonywania.

W poprzednim przykładzie dodano dekorator @mlflow.trace zarówno do __call__, jak i metod analizy. Dekorator tworzy przedział, który rozpoczyna się w momencie wywołania funkcji i kończy się po jej zakończeniu. Narzędzie MLflow automatycznie rejestruje dane wejściowe i wyjściowe funkcji oraz wszelkie zgłoszone wyjątki.

Note

Użytkownicy bibliotek LangChain, LlamaIndex i OpenAI mogą używać automatycznego rejestrowania MLflow, a także ręcznie definiować ślady przy użyciu dekoratora. Zobacz przegląd Śledzenia.

Zwróć listę obiektów Document z page_content i metadata, jak robi to powyższy przykład. Jest to zgodne ze schematem MLflow retriever span, dzięki czemu narzędzia do oceny i przeglądu mogą rozpoznać pobrane dokumenty.

Odczyt plików z woluminu Unity Catalog

Jeśli agent musi odczytywać nieustrukturyzowane pliki (dokumenty tekstowe, raporty, pliki konfiguracyjne itp.) przechowywane w woluminie Unity Catalog, możesz utworzyć narzędzia korzystające z Databricks SDK używającego interfejsu API Files do wyświetlania listy i odczytywania plików bezpośrednio.

Wybierz istniejący tom i nadaj tożsamości agenta USE CATALOG, USE SCHEMA oraz READ VOLUME. Zastąp <catalog>.<schema>.<volume> w kodzie. Do zadania przykładowego użyj pliku tekstowego UTF-8 o nazwie about_databricks.txt, albo zmień prompt na plik w swoim wolumenie.

W poniższych przykładach są tworzone dwa narzędzia, których agent może użyć:

  • list_volume_files: Wyświetla listę plików i katalogów w woluminie.
  • read_volume_file: odczytuje zawartość pliku tekstowego z woluminu.

LangChain/LangGraph

Zainstaluj najnowszą wersję programu obejmującą mostek sztucznej inteligencji databricks-langchain usługi Databricks.

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

Powiąż narzędzia z usługą LLM i uruchom pętlę wywoływania narzędzi:

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

Zainstaluj najnowszą wersję programu obejmującą mostek sztucznej inteligencji databricks-openai usługi Databricks.

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

Zapytanie do indeksu wyszukiwania AI za pomocą MCP

Serwer AI Search MCP korzysta z punktu końcowego MCP w przestrzeni roboczej. Wymaga indeksu z zarządzanymi embeddingami Azure Databricks. Zobacz Tworzenie punktów końcowych i indeksów wyszukiwania sztucznej inteligencji.

Adres URL MCP usługi AI Search dla obszaru roboczego to https://<workspace-hostname>/api/2.0/mcp/ai-search/{catalog}/{schema}/{index_name}. Połącz się z nim i wyświetl listę narzędzi, które uwidacznia:

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

Aby zbudować i wdrożyć agenta korzystającego z tego serwera, zobacz Wywoływanie narzędzi MCP z poziomu agenta języka Python. Przyznaj agentowi SELECT na obiekcie zabezpieczanym indeksu w usłudze Unity Catalog.

Wyszukiwanie sztucznej inteligencji było wcześniej wyszukiwaniem wektorowym. Poprzedni /api/2.0/mcp/vector-search/ prefiks URL nadal działa.

Wykonywanie zapytań w Databricks AI Search przy użyciu funkcji UC (wycofane)

Note

Dla nowych agentów Python użyj AI Bridge. Poniższy przykład jest odniesieniem do istniejących integracji.

Możesz stworzyć funkcję Unity Catalog, która opakuje zapytanie indeksu AI Search . Takie podejście:

  • Wspiera przypadki użycia w środowisku produkcyjnym z zarządzaniem i odkrywalnością.
  • Używa funkcji sql vector_search() pod maską
  • Obsługuje automatyczne śledzenie MLflow
    • Dane wyjściowe funkcji należy wyrównać do schematu modułu pobierania MLflow przy użyciu aliasów page_content i metadata.
    • Wszelkie dodatkowe kolumny metadanych należy dodać do metadata kolumny przy użyciu funkcji mapy SQL, a nie jako klucze wyjściowe najwyższego poziomu.

Uruchom następujący kod w notesie lub edytorze SQL, aby utworzyć funkcję:

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
  )

Postępuj zgodnie z konfiguracją narzędzia function tool, a następnie owiń tego retrievera za pomocą UCFunctionToolkit, aby użyć go w swoim agencie. Umożliwia to automatyczne śledzenie za pośrednictwem platformy MLflow poprzez automatyczne generowanie typów zakresów w dziennikach 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

Narzędzia pobierania wykazu katalogu Unity mają następujące uwagi:

  • Klienci SQL mogą ograniczyć maksymalną liczbę zwracanych wierszy lub bajtów. Aby zapobiec obcinaniu danych, obcinaj wartości kolumn zwracane przez funkcję UDF. Na przykład można użyć substring(chunked_text, 0, 8192), aby zmniejszyć rozmiar dużych kolumn zawartości i uniknąć obcinania wierszy podczas wykonywania.
  • Ponieważ to narzędzie jest otoką dla funkcji vector_search(), podlega tym samym ograniczeniom co funkcja vector_search(). Zobacz Ograniczenia.

Aby uzyskać więcej informacji na temat UCFunctionToolkit, zobacz dokumentację Unity Catalog .

Schematy niestandardowych retriever (starsze)

mlflow.models.set_retriever_schema jest przestarzałe. Dla nowych retrieverów zwróć Document obiekty według standardowego schematu retrievera. Poniższa konfiguracja dotyczy istniejących integracji, które zwracają niestandardowe pola.

Aby ręcznie ustawić schemat programu retriever:

  1. Wywołaj mlflow.models.set_retriever_schema podczas definiowania agenta. Użyj set_retriever_schema, aby zmapować nazwy kolumn w zwróconej tabeli na oczekiwane pola MLflow, takie jak primary_key, text_column i 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"],
    )
    
  2. Określ dodatkowe kolumny w schemacie modułu pobierania, podając listę nazw kolumn z polem other_columns .

  3. Jeśli masz wiele retrieverów, możesz zdefiniować wiele schematów, używając unikatowych nazw dla każdego schematu retrievera.

Schemat pobierania ustawiony podczas tworzenia agenta wpływa na aplikacje zależne i przepływy pracy, takie jak aplikacja przeglądowa i zestawy ocen. W szczególności kolumna doc_uri służy jako podstawowy identyfikator dokumentów zwracanych przez program retriever.