メインコンテンツまでスキップ
に最終更新

Unity Catalog 関数を使用して AI エージェント ツールを作成

Unity Catalog関数を使用して、カスタムロジックを実行し、LLMの言語生成能力を超えて機能を拡張する特定のタスクを実行するAIエージェントツールを作成します。

Unity Catalog 関数とMCPサーバーの使い分け​

Databricksは、クエリーが事前にわかっており、エージェントがパラメーターを提供する際に、特に構造化データ取得ツールとしてUnity Catalog関数をエージェントツールに使用することをお勧めします。エージェントを構造化データに接続するを参照してください。

他のほとんどのユースケースでは、Databricks は、実行速度の向上、ユーザーごとの認証サポート、および柔軟性の向上のために、MCP サーバーを使用するか、ロジックをエージェントコードに直接定義することをお勧めします。

要件​

AIエージェントツールとしてUnity Catalog関数を作成して使用するには、以下が必要です：

  • Databricks Runtime : Databricks Runtime 15.0 以降を使用してください。
  • Python バージョン : Python 3.10 以降をインストールします

Unity Catalog 関数を実行するには:

  • 本番運用でAIエージェントツールとしてUnity Catalog関数を実行するには、ワークスペースで**Serverlessコンピュート**を有効にする必要があります。 Serverlessコンピュート要件を参照してください。
    • Python 関数のローカルモード実行は、Serverless ジェネリック コンピュートを必要としませんが、ローカルモードは開発およびテスト目的でのみ使用されます。

Unity Catalog 関数を作成するには：

  • Databricks Workspace Client または SQL ボディ ステートメントを使用して関数を作成するには、ワークスペースで Serverless generic コンピュート を有効にする必要があります。
    • Python関数は、Serverless コンピュートなしで作成できます。

Unity Catalog 関数ツールを作成する​

以下のステップでは、Unity Catalog 関数を作成およびテストする方法について説明します。Databricks ノートブックで次のコードを実行します。

プロンプト

Genie Code（エージェントモード）にこれを実行してもらいます：

Create a Unity Catalog Python function that an AI agent can use as a tool. It should take two floating point numbers and return their sum, with type hints and a Google-style docstring. Register it using the Databricks Function Client, then test calling it.

依存関係をインストール​

[databricks] の追加機能を使用して Unity Catalog AI パッケージをインストールします。

Python
# Install Unity Catalog AI integration packages with the Databricks extra
%pip install unitycatalog-ai[databricks]

dbutils.library.restartPython()

Databricks Function Client を初期化します​

DatabricksでUnity Catalog関数を作成、管理、実行するための特殊なインターフェイスであるDatabricks Function Clientを初期化します。

Python
from unitycatalog.ai.core.databricks import DatabricksFunctionClient

client = DatabricksFunctionClient()

ツールのロジックを定義します​

Unity Catalogツールは、実際にはUnity Catalogのユーザー定義関数 (UDF) です。Unity Catalogツールを定義すると、Unity Catalogに関数を登録します。Unity Catalog UDF の詳細については、「Unity Catalog のSQLおよびPythonユーザー定義関数 (UDF)」を参照してください。

警告

エージェントツールで任意のコードを実行すると、エージェントがアクセスできる機密情報や個人情報が漏洩する可能性があります。顧客は、信頼できるコードのみを実行し、ガードレールと適切な権限を構成して、データへの意図しないアクセスを防ぐ責任があります。

Unity Catalog関数は、次の 2 つのAPIのいずれかを使用して作成できます。

  • create_python_function Pythonの呼び出し可能オブジェクトを受け入れます。
  • create_function SQL 本体の create function ステートメントを受け入れます。Python 関数の作成を参照してください。

create_python_function API を使用して関数を作成します。

Unity Catalog関数データモデルがPython callableを認識できるようにするには、お使いの関数は次の要件を満たす必要があります。

  • 型ヒント : 関数シグネチャは有効なPython型ヒントを定義する必要があります。名前付き引数と戻り値のどちらも、型が定義されている必要があります。

  • 可変引数を使用しないでください ：*args や **kwargs などの可変引数はサポートされていません。すべての引数を明示的に定義する必要があります。

  • 型の互換性 : すべての Python 型が SQL でサポートされているわけではありません。Spark でサポートされるデータ型を参照してください。

  • **説明的なdocstring**：Unity Catalog関数ツールキットは、docstringから重要な情報を読み取り、解析し、抽出します。

    • docstring は、Google docstring 構文に従ってフォーマットする必要があります。
    • LLMがその関数をいつどのように使用するかを理解できるように、関数とその引数について明確な記述を作成してください。

  • 依存関係のインポート : ライブラリは関数本体内でインポートする必要があります。関数外部のインポートは、ツールを実行する際に解決されません。

以下のコードスニペットは、create_python_function を使用して Python の呼び出し可能ファイル add_numbers を登録します:

Python

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
)

関数をテストします​

関数をテストして、期待どおりに動作するか確認します。execute_function API で完全修飾関数名を指定して関数を実行します。

Python
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'

Unity Catalog 関数をエージェントに追加する​

Unity Catalog 関数を作成してテストしたら、次のいずれかの方法を選択してエージェントに追加します。

Mcpアイコン。 MCP を使用 (推奨)

MCPの使用（推奨）​

Databricksは、Unity Catalog関数をエージェントに追加するためにMCPサーバーを使用することをお勧めします。MCPアプローチは、自動ツール検出と組み込みの認証サポートにより、よりシンプルな統合を提供します。

Unity Catalog 関数の管理対象 MCP URL は、https://<workspace-hostname>/api/2.0/mcp/functions/{catalog}/{schema} です。特定の関数は、オプションで /{function_name} を付加して指定できます。

次の例は、MCPを介してエージェントをUnity Catalog関数に接続する方法を示しています。<catalog>と<schema>を関数の場所に置き換えてください。

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

workspace_client = WorkspaceClient()

async with McpServer.from_uc_function(
    catalog="<catalog>",
    schema="<schema>",
    workspace_client=workspace_client,
    name="uc-functions",
) as uc_server:
    agent = Agent(
        name="Tool-using agent",
        instructions="You are a helpful assistant. Use the available tools to answer questions.",
        model="databricks-claude-sonnet-4-5",
        mcp_servers=[uc_server],
    )
    result = await Runner.run(agent, "Look up customer info for Acme Corp")
    print(result.final_output)

databricks.yml内のUnity Catalog関数へのアクセス権をアプリに付与します：

YAML
resources:
  apps:
    my_agent_app:
      resources:
        - name: 'my_uc_function'
          uc_securable:
            securable_full_name: '<catalog>.<schema>.<function-name>'
            securable_type: 'FUNCTION'
            permission: 'EXECUTE'

関数アイコン。 UCFunctionToolkit を使用する

UCFunctionToolkit の使用​

この例では LangChain を使用していますが、同様のアプローチを他のライブラリにも適用できます。Unity Catalog ツール統合を参照してください。

追加の依存関係をインストールする​

UCFunctionToolkit 用の LangChain 統合パッケージをインストールします。

Python
%pip install unitycatalog-langchain[databricks]==0.2.0

# Install the Databricks LangChain integration package
%pip install databricks-langchain==0.5.0

dbutils.library.restartPython()

UCFunctionToolKit を使用して関数をラップします​

UCFunctionToolkitを使用して関数をラップし、エージェントオーサリングライブラリからアクセスできるようにします。このツールキットは、異なる生成AI ライブラリ間での一貫性を確保し、レトリーバーの自動トレースなどの便利な機能を追加します。

Python
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

エージェントでツールを使用します。​

UCFunctionToolkitからのtoolsプロパティを使用して、LangChainエージェントにツールを追加します。

注記

This example uses LangChain. However you can integrate Unity Catalog tools with other frameworks such as LlamaIndex, OpenAI, Anthropic, and more. See Unity Catalog tool integration.

この例では、わかりやすくするために LangChain AgentExecutor API を使用して単純なエージェントを作成します。本番運用ワークロードの場合は、AIエージェントを作成してDatabricks Appsにデプロイするに示されているエージェント作成ワークフローを使用します。

Python
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 (optional: replace with your LLM of choice)
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?"})

明確なドキュメントでツール呼び出しを改善​

良いドキュメントは、エージェントが各ツールをいつどのように使用するかを理解するのに役立ちます。ツールを文書化するためのベストプラクティスに従ってください。

  • Unity Catalog の関数では、ツール機能とパラメーターを記述するために COMMENT 句を使用してください。
  • 期待される入出力を明確に定義します。
  • エージェントや人間がツールをより使いやすくするために、意味のある説明を記述してください。

例：効果的なツールのドキュメント​

以下の例では、構造化されたテーブルをクエリーするツール用として、明確な COMMENT 文字列を示します。

SQL
CREATE OR REPLACE FUNCTION main.default.lookup_customer_info(
  customer_name STRING COMMENT 'Name of the customer whose info to look up.'
)
RETURNS STRING
COMMENT 'Returns metadata about a specific customer including their email and ID.'
RETURN SELECT CONCAT(
    'Customer ID: ', customer_id, ', ',
    'Customer Email: ', customer_email
  )
  FROM main.default.customer_data
  WHERE customer_name = customer_name
  LIMIT 1;

例: 効果のないツール ドキュメント​

次の例には重要な詳細が不足しているため、エージェントがツールを効果的に使用することが困難になります。

SQL
CREATE OR REPLACE FUNCTION main.default.lookup_customer_info(
  customer_name STRING COMMENT 'Name of the customer.'
)
RETURNS STRING
COMMENT 'Returns info about a customer.'
RETURN SELECT CONCAT(
    'Customer ID: ', customer_id, ', ',
    'Customer Email: ', customer_email
  )
  FROM main.default.customer_data
  WHERE customer_name = customer_name
  LIMIT 1;

Serverlessまたはローカルモードを使用して関数を実行する​

生成AI サービスがツール呼び出しが必要であると判断すると、統合パッケージ (UCFunctionToolkit インスタンス) は DatabricksFunctionClient.execute_function API を実行します。

execute_functionの呼び出しは、Serverless またはローカルの2つの実行モードで関数を実行できます。このモードでは、どのリソースが関数を実行するかが決定されます。

本番運用におけるサーバレスモード​

Unity Catalog 関数をAIエージェントツールとして実行する場合、Serverless モードは本番運用ユースケースでdefaultかつ推奨されるオプションです。このモードは、Serverless ジェネリック コンピュート (Spark Connect Serverless) を使用して関数をリモートで実行し、Lakeguard は、エージェントのプロセスが安全に保たれ、任意のコードをローカルで実行するリスクがないことを保証します。

注記

Unity Catalog 機能は、AI エージェント ツールとして実行される場合、Serverless SQL Warehouse ではなく、サーバーレスジェネリックコンピュート (Spark Connect Serverless) を必要とします。Serverlessジェネリックコンピュートなしでツールを実行しようとすると、PERMISSION_DENIED: Cannot access Spark Connect のようなエラーが発生します。

Python
# Defaults to serverless if `execution_mode` is not specified
client = DatabricksFunctionClient(execution_mode="serverless")

エージェントが Serverless モードでのツール実行を要求すると、以下のようになります：

  1. DatabricksFunctionClient は、定義がローカルにキャッシュされていない場合、関数定義を取得するためにUnity Catalogにリクエストを送信します。
  2. DatabricksFunctionClientは関数定義を抽出し、パラメーター名とタイプを検証します。
  3. DatabricksFunctionClient は、実行を UDF として Serverless 汎用コンピュートに送信します。

開発用ローカルモード​

ローカルモードでは、Serverless汎用コンピュートへのリクエストを行う代わりに、Python関数をローカルサブプロセスで実行します。これにより、ローカルスタックトレースを提供することで、ツール呼び出しのトラブルシューティングをより効果的に行うことができます。これは、Python Unity Catalog関数の開発およびデバッグ用に設計されています。

エージェントが ローカル モードでのツールの実行を要求すると、 DatabricksFunctionClient は次の処理を行います。

  1. 関数定義がローカルにキャッシュされていない場合、Unity Catalogに関数定義の取得をリクエストします。
  2. Python 呼び出し可能定義を抽出し、呼び出し可能ファイルをローカルにキャッシュし、パラメーター名と型を検証します。
  3. 指定されたパラメーターを使用して、タイムアウト保護機能付きの制限されたサブプロセスで呼び出し可能オブジェクトを呼び出します。
Python
# Defaults to serverless if `execution_mode` is not specified
client = DatabricksFunctionClient(execution_mode="local")

"local" モードで実行すると、次の機能が提供されます。

  • CPU時間の上限： 呼び出し可能な実行の合計CPUランタイムを制限して、過度な計算負荷を防ぎます。

    CPU時間制限は、経過実時間ではなく実際のCPU使用量に基づいています。システムスケジューリングと並列プロセスにより、実際のシナリオではCPU時間が経過実時間を超えることがあります。

  • メモリ制限： プロセスに割り当てられる仮想メモリを制限します。

  • タイムアウト保護： 実行中の関数に対して、合計実時間タイムアウトを強制します。

環境変数を使用してこれらの制限をカスタマイズします（詳細はこちら）。

ローカルモードの制限事項​

  • Python関数のみ ：SQLベースの関数はローカルモードではサポートされていません。
  • 信頼されていないコードに関するセキュリティの考慮事項： ローカルモードは、プロセス分離のためにサブプロセスで関数を実行しますが、AIシステムによって生成された任意のコードを実行する際に、潜在的なセキュリティリスクがあります。これは主に、レビューされていない動的に生成されたPythonコードを関数が実行する場合に懸念されます。
  • ライブラリのバージョンの違い: Serverless とローカル実行環境の間でライブラリのバージョンが異なる可能性があり、これにより関数の動作が異なる場合があります。

環境変数​

DatabricksFunctionClient での機能の実行方法を以下の環境変数を使用して構成します:

環境変数

デフォルト値

説明

EXECUTOR_MAX_CPU_TIME_LIMIT

10 秒

最大許容CPU実行時間（ローカルモードのみ）。

EXECUTOR_MAX_MEMORY_LIMIT

100 MB

プロセスに対する最大許容仮想メモリ割り当て（ローカルモードのみ）。

EXECUTOR_TIMEOUT

20 秒

最大合計実測時間（ローカルモードのみ）。

UCAI_DATABRICKS_SESSION_RETRY_MAX_ATTEMPTS

5

トークン失効時にセッションクライアントの更新を再試行する最大試行回数。

UCAI_DATABRICKS_SERVERLESS_EXECUTION_RESULT_ROW_LIMIT

100

Serverlessコンピュートと databricks-connect を使用して関数を実行する際に返される最大行数。

環境変数

デフォルト値

説明

EXECUTOR_MAX_CPU_TIME_LIMIT

10 秒

最大許容CPU実行時間（ローカルモードのみ）。

EXECUTOR_MAX_MEMORY_LIMIT

100 MB

プロセスに対する最大許容仮想メモリ割り当て（ローカルモードのみ）。

EXECUTOR_TIMEOUT

20 秒

最大合計実測時間（ローカルモードのみ）。

UCAI_DATABRICKS_SESSION_RETRY_MAX_ATTEMPTS

5

トークン失効時にセッションクライアントの更新を再試行する最大試行回数。

UCAI_DATABRICKS_SERVERLESS_EXECUTION_RESULT_ROW_LIMIT

100

Serverlessコンピュートと databricks-connect を使用して関数を実行する際に返される最大行数。

http_request を使用して外部APIsを呼び出します（レガシー）​

注記

エージェントを外部サービスに接続するには、DatabricksはMCP ServicesまたはUnity Catalog接続プロキシを推奨します。http_requestをラップするUC関数ツールは引き続きサポートされますが、推奨されるアプローチではなくなりました。

外部サービスを呼び出すhttp_request()をラップするUnity Catalog関数を作成できます。このアプローチは、SQLベースのツールの定義に役立ちます。

次の例は、Slackにメッセージを投稿するUnity Catalog関数ツールを作成します。

SQL
CREATE OR REPLACE FUNCTION main.default.slack_post_message(
  text STRING COMMENT 'message content'
)
RETURNS STRING
COMMENT 'Sends a Slack message by passing in the message and returns the response received from the external service.'
RETURN (http_request(
  conn => 'test_sql_slack',
  method => 'POST',
  path => '/api/chat.postMessage',
  json => to_json(named_struct(
    'channel', "C032G2DAH3",
    'text', text
  ))
)).text

CREATE FUNCTION (SQL、Python、Scala、および Java)をご覧ください。

注記

http_request を使用したSQLアクセスは、ユーザーごとのUser-to-Machineおよび動的クライアント登録接続タイプではブロックされます。代わりにPython Databricks SDKを使用してください。

ノートブックの例​

以下のノートブックは、Unity Catalog 関数を使用して外部サービスに接続するAIエージェントツールの作成を示しています。

Slackメッセージングエージェントツール

ノートブックを新しいタブで開くOpen in Databricks

Microsoft Graph API エージェント ツール

ノートブックを新しいタブで開くOpen in Databricks

Azure AI Search エージェント ツール

ノートブックを新しいタブで開くOpen in Databricks

次のステップ​

このページの見出し