Notatka
Dostęp do tej strony wymaga autoryzacji. Może spróbować zalogować się lub zmienić katalogi.
Dostęp do tej strony wymaga autoryzacji. Możesz spróbować zmienić katalogi.
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.
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>') );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ćmetadatapola w klasie Document, aby dodać atrybuty do zwróconego dokumentu, na przykładdoc_uriisimilarity_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 ]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_contentimetadata. - Wszelkie dodatkowe kolumny metadanych należy dodać do
metadatakolumny przy użyciu funkcji mapy SQL, a nie jako klucze wyjściowe najwyższego poziomu.
- Dane wyjściowe funkcji należy wyrównać do schematu modułu pobierania MLflow przy użyciu aliasów
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 funkcjavector_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:
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 jakprimary_key,text_columnidoc_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"], )Określ dodatkowe kolumny w schemacie modułu pobierania, podając listę nazw kolumn z polem
other_columns.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.
-
Apka recenzji wyświetla
doc_uri, aby pomóc recenzentom ocenić odpowiedzi i śledzić pochodzenie dokumentów. Zobacz Przegląd interfejsu użytkownika aplikacji. -
Zestawy oceny służą
doc_urido porównywania wyników modułu pobierającego ze wstępnie zdefiniowanymi zestawami danych oceny w celu określenia jego czułości i precyzji. Zobacz informacje o zbiorze danych do oceny.