整合 Unity Catalog 工具與第三方生成式 AI 框架

Unity Catalog 的 AI 代理工具可用於熱門的生成式 AI 函式庫，如 LangChain、LlamaIndex、OpenAI 和 Anthropic。這些整合結合了 Unity 目錄工具控管與第三方代理程式撰寫架構的功能。例如：

在 LangChain 中，Unity 目錄函式可以是代理程式工作流程的一部分，以執行查詢或轉換數據等工作。
在OpenAI或人類整合中，AI 模型會在執行期間直接呼叫函式。

在以下分頁選擇你的框架，建立 Unity 目錄工具並搭配該框架使用。在 Azure Databricks 筆記本或 Python 腳本中執行程式碼。

Requirements

安裝 Python 3.10 或更新版本。

LangChain

使用 Azure Databricks Unity Catalog 將 SQL 和 Python 函式整合為 LangChain 和 LangGraph 工作流程的工具。此整合結合了 Unity 目錄的治理與 LangChain 功能，以建置功能強大的 LLM 型應用程式。

在此範例中，您會建立 Unity 目錄工具、測試其功能，並將其新增至代理程式。

安裝依賴項

使用 Databricks 選擇性安裝 Unity 目錄 AI 套件，並安裝 LangChain 整合套件。

# Install the Unity Catalog AI integration package with the Databricks extra
%pip install unitycatalog-langchain[databricks]

# Install Databricks Langchain integration package
%pip install databricks-langchain
dbutils.library.restartPython()

初始化 Databricks 函式用戶端

初始化 Databricks 函式用戶端。

from unitycatalog.ai.core.base import get_uc_function_client

client = get_uc_function_client()

定義工具的邏輯

建立包含工具邏輯的 Unity 目錄函式。


CATALOG = "my_catalog"
SCHEMA = "my_schema"

def add_numbers(number_1: float, number_2: float) -> float:
  """
  A function that accepts two floating point numbers adds them,
  and returns the resulting sum as a float.

  Args:
    number_1 (float): The first of the two numbers to add.
    number_2 (float): The second of the two numbers to add.

  Returns:
    float: The sum of the two input numbers.
  """
  return number_1 + number_2

function_info = client.create_python_function(
  func=add_numbers,
  catalog=CATALOG,
  schema=SCHEMA,
  replace=True
)

測試函式

測試您的函式，以確保它如預期地運行：

result = client.execute_function(
  function_name=f"{CATALOG}.{SCHEMA}.add_numbers",
  parameters={"number_1": 36939.0, "number_2": 8922.4}
)

result.value # OUTPUT: '45861.4'

使用 UCFunctionToolKit 來包裝函式

使用 UCFunctionToolkit 封裝函式，以便代理程式開發庫存取。此工具組可確保不同庫的一致性，並新增實用的功能，例如檢索器的自動追踪。

from databricks_langchain import UCFunctionToolkit

# Create a toolkit with the Unity Catalog function
func_name = f"{CATALOG}.{SCHEMA}.add_numbers"
toolkit = UCFunctionToolkit(function_names=[func_name])

tools = toolkit.tools

在代理程式中使用工具

使用 tools 中的 UCFunctionToolkit 屬性將工具新增至 LangChain 代理程式。

本範例使用 LangChain 的 AgentExecutor API 撰寫簡單的代理程式，以求簡單。對於生產工作負載，請使用 Author a AI Agent 中看到的代理編寫工作流程，並部署到 Databricks 應用程式上。

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain.prompts import ChatPromptTemplate
from databricks_langchain import (
  ChatDatabricks,
  UCFunctionToolkit,
)
import mlflow

# Initialize the LLM (replace with your LLM of choice, if desired)
LLM_ENDPOINT_NAME = "databricks-meta-llama-3-3-70b-instruct"
llm = ChatDatabricks(endpoint=LLM_ENDPOINT_NAME, temperature=0.1)

# Define the prompt
prompt = ChatPromptTemplate.from_messages(
  [
    (
      "system",
      "You are a helpful assistant. Make sure to use tools for additional functionality.",
    ),
    ("placeholder", "{chat_history}"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
  ]
)

# Enable automatic tracing
mlflow.langchain.autolog()

# Define the agent, specifying the tools from the toolkit above
agent = create_tool_calling_agent(llm, tools, prompt)

# Create the agent executor
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
agent_executor.invoke({"input": "What is 36939.0 + 8922.4?"})

LlamaIndex

使用 Azure Databricks Unity Catalog 將 SQL 與 Python 函式整合為 LlamaIndex 工作流程的工具。此整合將 Unity Catalog 的數據管理功能與 LlamaIndex 的能力結合，目的是為大型語言模型（LLM）編製索引並查詢大型數據集。

安裝適用於 LlamaIndex 的 Databricks Unity 目錄整合套件。

%pip install unitycatalog-llamaindex[databricks]
dbutils.library.restartPython()

建立 Unity 目錄函式用戶端的實例。

from unitycatalog.ai.core.base import get_uc_function_client

client = get_uc_function_client()

建立以 Python 撰寫的 Unity 目錄函式。

CATALOG = "your_catalog"
SCHEMA = "your_schema"

func_name = f"{CATALOG}.{SCHEMA}.code_function"

def code_function(code: str) -> str:
  """
  Runs Python code.

  Args:
    code (str): The Python code to run.
  Returns:
    str: The result of running the Python code.
  """
  import sys
  from io import StringIO
  stdout = StringIO()
  sys.stdout = stdout
  exec(code)
  return stdout.getvalue()

client.create_python_function(
  func=code_function,
  catalog=CATALOG,
  schema=SCHEMA,
  replace=True
)

建立一個 Unity 目錄函式作為工具包實例，並執行以驗證工具是否正常運作。

from unitycatalog.ai.llama_index.toolkit import UCFunctionToolkit
import mlflow

# Enable traces
mlflow.llama_index.autolog()

# Create a UCFunctionToolkit that includes the UC function
toolkit = UCFunctionToolkit(function_names=[func_name])

# Fetch the tools stored in the toolkit
tools = toolkit.tools
python_exec_tool = tools[0]

# Run the tool directly
result = python_exec_tool.call(code="print(1 + 1)")
print(result)  # Outputs: {"format": "SCALAR", "value": "2\n"}

藉由將 Unity 目錄函式定義為 LlamaIndex 工具集合的一部分，以使用 LlamaIndex ReActAgent 中的工具。然後呼叫 LlamaIndex 工具集合，確認代理程式的行為正確。

from llama_index.llms.openai import OpenAI
from llama_index.core.agent import ReActAgent

llm = OpenAI()

agent = ReActAgent.from_tools(tools, llm=llm, verbose=True)

agent.chat("Please run the following python code: `print(1 + 1)`")

OpenAI

使用 Azure Databricks Unity Catalog，將 SQL 和 Python 函式整合為 OpenAI 工作流程的工具。此整合結合了 Unity 目錄與 OpenAI 的治理，以建立功能強大的 Gen AI 應用程式。

安裝適用於 OpenAI 的 Databricks Unity 目錄整合套件。

%pip install unitycatalog-openai[databricks]
%pip install mlflow -U
dbutils.library.restartPython()

建立 Unity 目錄函式用戶端的實例。

from unitycatalog.ai.core.base import get_uc_function_client

client = get_uc_function_client()

建立以 Python 撰寫的 Unity 目錄函式。

CATALOG = "your_catalog"
SCHEMA = "your_schema"

func_name = f"{CATALOG}.{SCHEMA}.code_function"

def code_function(code: str) -> str:
  """
  Runs Python code.

  Args:
    code (str): The python code to run.
  Returns:
    str: The result of running the Python code.
  """
  import sys
  from io import StringIO
  stdout = StringIO()
  sys.stdout = stdout
  exec(code)
  return stdout.getvalue()

client.create_python_function(
  func=code_function,
  catalog=CATALOG,
  schema=SCHEMA,
  replace=True
)

建立 Unity Catalog 函式的實例作為工具組，並藉由執行函式來確認工具運作正常。

from unitycatalog.ai.openai.toolkit import UCFunctionToolkit
import mlflow

# Enable tracing
mlflow.openai.autolog()

# Create a UCFunctionToolkit that includes the UC function
toolkit = UCFunctionToolkit(function_names=[func_name])

# Fetch the tools stored in the toolkit
tools = toolkit.tools
client.execute_function = tools[0]

將要求連同工具一起提交至 OpenAI 模型。

import openai

messages = [
  {
    "role": "system",
    "content": "You are a helpful customer support assistant. Use the supplied tools to assist the user.",
  },
  {"role": "user", "content": "What is the result of 2**10?"},
]
response = openai.chat.completions.create(
  model="gpt-4o-mini",
  messages=messages,
  tools=tools,
)
# check the model response
print(response)

在 OpenAI 傳回回應之後，叫用 Unity Catalog 函式，以產生回應並傳回給 OpenAI。

import json

# OpenAI sends only a single request per tool call
tool_call = response.choices[0].message.tool_calls[0]
# Extract arguments that the Unity Catalog function needs to run
arguments = json.loads(tool_call.function.arguments)

# Run the function based on the arguments
result = client.execute_function(func_name, arguments)
print(result.value)

傳回答案之後，您可以建構後續對OpenAI呼叫的響應承載。

# Create a message containing the result of the function call
function_call_result_message = {
  "role": "tool",
  "content": json.dumps({"content": result.value}),
  "tool_call_id": tool_call.id,
}
assistant_message = response.choices[0].message.to_dict()
completion_payload = {
  "model": "gpt-4o-mini",
  "messages": [*messages, assistant_message, function_call_result_message],
}

# Generate final response
openai.chat.completions.create(
  model=completion_payload["model"], messages=completion_payload["messages"]
)

公用程式

為了簡化製作工具回應的程式， ucai-openai 套件具有公用程式 generate_tool_call_messages，可轉換 OpenAI ChatCompletion 回應訊息，以便用於產生回應。

from unitycatalog.ai.openai.utils import generate_tool_call_messages

messages = generate_tool_call_messages(response=response, client=client)
print(messages)

Note

如果回應包含多個選擇專案，您可以在呼叫 generate_tool_call_messages 時傳遞choice_index自變數，以選擇要使用哪一個選擇專案。目前不支援處理多項選項。

Anthropic

使用 Azure Databricks Unity Catalog 來整合 SQL 和 Python 函式，作為 Anthropic SDK LLM 呼叫的工具。此集成將 Unity Catalog 的治理與 Anthropic 模型相結合，以創建強大的 gen AI 應用程式。

Note

Anthropic 整合功能需使用 Databricks Runtime 15.0 或以上版本。

安裝適用於 Anthropic 的 Databricks Unity Catalog 集成包。

%pip install unitycatalog-anthropic[databricks]
dbutils.library.restartPython()

建立 Unity 目錄函式用戶端的實例。

from unitycatalog.ai.core.base import get_uc_function_client

client = get_uc_function_client()

建立以 Python 撰寫的 Unity 目錄函式。

CATALOG = "your_catalog"
SCHEMA = "your_schema"

func_name = f"{CATALOG}.{SCHEMA}.weather_function"

def weather_function(location: str) -> str:
  """
  Fetches the current weather from a given location in degrees Celsius.

  Args:
    location (str): The location to fetch the current weather from.
  Returns:
    str: The current temperature for the location provided in Celsius.
  """
  return f"The current temperature for {location} is 24.5 celsius"

client.create_python_function(
  func=weather_function,
  catalog=CATALOG,
  schema=SCHEMA,
  replace=True
)

將 Unity Catalog 函數的實體建立為工具包。

from unitycatalog.ai.anthropic.toolkit import UCFunctionToolkit

# Create an instance of the toolkit
toolkit = UCFunctionToolkit(function_names=[func_name], client=client)

在 Anthropic 中使用工具呼叫。

import anthropic

# Initialize the Anthropic client with your API key
anthropic_client = anthropic.Anthropic(api_key="YOUR_ANTHROPIC_API_KEY")

# User's question
question = [{"role": "user", "content": "What's the weather in New York City?"}]

# Make the initial call to Anthropic
response = anthropic_client.messages.create(
  model="claude-3-5-sonnet-20240620",  # Specify the model
  max_tokens=1024,  # Use 'max_tokens' instead of 'max_tokens_to_sample'
  tools=toolkit.tools,
  messages=question  # Provide the conversation history
)

# Print the response content
print(response)

建立工具回應。如果需要調用工具，則來自 Claude 模型的回應包含一個工具請求元數據塊。

from unitycatalog.ai.anthropic.utils import generate_tool_call_messages

# Call the UC function and construct the required formatted response
tool_messages = generate_tool_call_messages(
  response=response,
  client=client,
  conversation_history=question
)

# Continue the conversation with Anthropic
tool_response = anthropic_client.messages.create(
  model="claude-3-5-sonnet-20240620",
  max_tokens=1024,
  tools=toolkit.tools,
  messages=tool_messages,
)

print(tool_response)

該 unitycatalog.ai-anthropic 包包括一個消息處理程序實用程式，用於簡化對 Unity Catalog 函數的調用的解析和處理。該實用程式執行以下作：

檢測工具調用要求。
從查詢中擷取工具呼叫資訊。
執行對 Unity Catalog 函數的調用。
分析來自 Unity Catalog 函數的回應。
製作下一個消息格式以繼續與 Claude 的對話。

Note

必須在 API 的conversation_history參數中提供generate_tool_call_messages整個對話歷史記錄。 Claude 模型需要初始化對話（原始使用者輸入問題）以及所有後續 LLM 生成的回應和多輪次工具調用結果。

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-06-30

整合 Unity Catalog 工具與第三方生成式 AI 框架

Requirements

LangChain

安裝依賴項

初始化 Databricks 函式用戶端

定義工具的邏輯

測試函式

使用 UCFunctionToolKit 來包裝函式

在代理程式中使用工具

LlamaIndex

OpenAI

公用程式

Anthropic

意見反應

其他資源