AI エージェントの使用を開始する

Mosaic AI Agent Framework を使用して、初めての AI エージェントを構築します。このチュートリアルでは、次のことを行います。

Agent Frameworkを使用してエージェントを作成します。
エージェントにツールを追加します。
エージェントを Databricks モデルサービングエンドポイントにデプロイします。

エージェントとその他の生成AIアプリの概念の概要については、「生成AI AI アプリとは」を参照してください。

必要条件

ワークスペースで次の機能が有効になっている必要があります。

Unity Catalog
モザイク AI エージェントフレームワーク
基盤モデル (トークン単位の従量課金, プロビジョニングスループット, または外部モデル). 利用可能な地域が限られている機能を見る

ノートブックの例

このノートブックには、最初の AI エージェントを作成してデプロイするために必要なすべてのコードが含まれています。ノートブックを Databricks ワークスペースにインポートして実行します。

Mosaic AIエージェントのデモ

Open notebook in new tab

エージェントの定義

AI エージェントは、次のもので構成されます。

推論と意思決定が可能な大規模言語モデル (LLM)
LLM がテキストを生成するだけでなく、Python コードの実行やデータの取得など、さまざまなことを行うために使用できるツール

Databricks ノートブックで次のコードを実行して、単純なツール呼び出しエージェントを定義します。

必要なPythonパッケージをインストールします。
Python
```
%pip install -U -qqqq mlflow databricks-openai databricks-agents"
dbutils.library.restartPython()
```
- mlflow: エージェント開発とエージェントトレースに使用されます。
- databricks-openai: Databricks がホストする LLM に接続し、Unity Catalog ツールにアクセスするために使用されます。
- databricks-agents: エージェントをパッケージ化してデプロイするために使用されます。

エージェントを定義します。このコードスニペットでは、次のことを行います。

OpenAI クライアントを使用して Databricks モデルサービングエンドポイントに接続します。
autolog()を使用して MLflow トレースを有効にします。これにより、インストルメンテーションが追加され、クエリを送信したときにエージェントが何をするかを確認できます。
system.ai.python_execツールをエージェントに追加します。この組み込み Unity Catalog 関数を使用すると、エージェントは Python コードを実行できます。
MLflow ヘルパー関数 ( output_to_responses_items_stream 、 create_function_call_output_item ) を使用して、ストリーミング LLM 出力を Responses API 互換形式に変換します。

Python
import json
import mlflow
from databricks.sdk import WorkspaceClient
from databricks_openai import UCFunctionToolkit, DatabricksFunctionClient
# Import MLflow utilities for converting from chat completions to Responses API format
from mlflow.types.responses import output_to_responses_items_stream, create_function_call_output_item

# Enable automatic tracing for easier debugging
mlflow.openai.autolog()

# Get an OpenAI client configured to connect to Databricks model serving endpoints
openai_client = WorkspaceClient().serving_endpoints.get_open_ai_client()

# Load Databricks built-in tools (Python code interpreter)
client = DatabricksFunctionClient()
builtin_tools = UCFunctionToolkit(function_names=["system.ai.python_exec"], client=client).tools
for tool in builtin_tools:
  del tool["function"]["strict"]

def call_tool(tool_name, parameters):
  if tool_name == "system__ai__python_exec":
    return DatabricksFunctionClient().execute_function("system.ai.python_exec", parameters=parameters).value
  raise ValueError(f"Unknown tool: {tool_name}")

def call_llm(prompt):
  for chunk in openai_client.chat.completions.create(
    model="databricks-claude-sonnet-4-5",
    messages=[{"role": "user", "content": prompt}],
    tools=builtin_tools,
    stream=True
  ):
    yield chunk.to_dict()

def run_agent(prompt):
  """
  Send a user prompt to the LLM, and yield LLM + tool call responses
  The LLM is allowed to call the code interpreter tool if needed, to respond to the user
  """
  # Convert output into Responses API-compatible events
  for chunk in output_to_responses_items_stream(call_llm(prompt)):
    yield chunk.model_dump(exclude_none=True)
  # If the model executed a tool, call it and yield the tool call output in Responses API format
  if chunk.item.get('type') == 'function_call':
    tool_name = chunk.item["name"]
    tool_args = json.loads(chunk.item["arguments"])
    tool_result = call_tool(tool_name, tool_args)
    yield {"type": "response.output_item.done", "item": create_function_call_output_item(call_id=chunk.item["call_id"], output=tool_result)}

エージェントをテストする

エージェントをテストするには、Pythonコードの実行が必要なプロンプトでエージェントをクエリします。

Python
for output_chunk in run_agent("What is the square root of 429?"):
  print(output_chunk)

LLM の出力に加えて、詳細なトレース情報がノートブックに直接表示されます。これらのトレースは、低速または失敗したエージェントコールをデバッグするのに役立ちます。これらのトレースは、 mlflow.openai.autolog() .

エージェントをデプロイする

エージェントが作成されたので、それをパッケージ化して Databricks サービスエンドポイントにデプロイできます。デプロイされたエージェントに関するフィードバックの収集を開始するには、エージェントを他のユーザーと共有し、組み込みのチャットUIを使用してチャットします。

デプロイ用のエージェントコードの準備

デプロイ用にエージェントコードを準備するには、MLflow の ResponsesAgent インターフェイスを使用してラップします。ResponsesAgent インターフェイスは、Databricks にデプロイするエージェントをパッケージ化するための推奨される方法です。

ResponsesAgentインターフェースを実装するには、 predict_stream() （ストリーミング応答用）メソッドとpredict() （非ストリーミング要求用）メソッドの両方を定義します。基盤となるエージェントロジックはすでに Responses API 互換のイベントを出力しているため、実装は簡単です。

Python
from mlflow.pyfunc import ResponsesAgent
from mlflow.types.responses import ResponsesAgentRequest, ResponsesAgentResponse, ResponsesAgentStreamEvent

class QuickstartAgent(ResponsesAgent):
  def predict_stream(self, request: ResponsesAgentRequest):
    # Extract the user's prompt from the request
    prompt = request.input[-1].content
    # Stream response items from our agent
    for chunk in run_agent(prompt):
      yield ResponsesAgentStreamEvent(**chunk)

  def predict(self, request: ResponsesAgentRequest) -> ResponsesAgentResponse:
    outputs = [
      event.item
      for event in self.predict_stream(request)
      if event.type == "response.output_item.done"
    ]
    return ResponsesAgentResponse(output=outputs)

次のコードをノートブックに追加して、 ResponsesAgent クラスをテストします。

Python
from mlflow.types.responses import ResponsesAgentRequest

AGENT = QuickstartAgent()

# Create a ResponsesAgentRequest with input messages
request = ResponsesAgentRequest(
  input=[
    {
      "role": "user",
      "content": "What's the square root of 429?"
    }
  ]
)

for event in AGENT.predict_stream(request):
  print(event)

すべてのエージェントコードを1つのファイルに結合して、ログに記録してデプロイできるようにします。

すべてのエージェントコードを 1 つのノートブックセルに統合します。
セルの上部に、エージェントをファイルに保存するための %%writefile quickstart_agent.py マジックコマンドを追加します。
セルの下部で、エージェントオブジェクトを使用して mlflow.models.set_model() に電話をかけます。これにより、予測を提供するときに使用するエージェントオブジェクトが MLflow に指示されます。この手順により、エージェントコードへのエントリポイントが効果的に構成されます。

ノートブックのセルは、次のようになります。

Python
%%writefile quickstart_agent.py

import json
from databricks.sdk import WorkspaceClient
from databricks_openai import UCFunctionToolkit, DatabricksFunctionClient

import mlflow
from mlflow.pyfunc import ResponsesAgent
from mlflow.types.responses import (
  ResponsesAgentRequest,
  ResponsesAgentResponse,
  ResponsesAgentStreamEvent,
  output_to_responses_items_stream,
  create_function_call_output_item
)

# Enable automatic tracing for deployed agent
mlflow.openai.autolog()

# Get an OpenAI client configured to talk to Databricks model serving endpoints
openai_client = WorkspaceClient().serving_endpoints.get_open_ai_client()

# Load Databricks built-in tools (Python code interpreter)
client = DatabricksFunctionClient()
builtin_tools = UCFunctionToolkit(function_names=["system.ai.python_exec"], client=client).tools
for tool in builtin_tools:
  del tool["function"]["strict"]

def call_tool(tool_name, parameters):
  if tool_name == "system__ai__python_exec":
    return DatabricksFunctionClient().execute_function("system.ai.python_exec", parameters=parameters).value
  raise ValueError(f"Unknown tool: {tool_name}")

def call_llm(prompt):
  for chunk in openai_client.chat.completions.create(
    model="databricks-claude-sonnet-4-5",
    messages=[{"role": "user", "content": prompt}],
    tools=builtin_tools,
    stream=True
  ):
    yield chunk.to_dict()

def run_agent(prompt):
  """
  Send a user prompt to the LLM, and yield LLM + tool call responses
  The LLM is allowed to call the code interpreter tool if needed, to respond to the user
  """
  # Convert output into Responses API-compatible events
  for chunk in output_to_responses_items_stream(call_llm(prompt)):
    yield chunk.model_dump(exclude_none=True)
  # If the model executed a tool, call it and yield the tool call output in Responses API format
  if chunk.item.get('type') == 'function_call':
    tool_name = chunk.item["name"]
    tool_args = json.loads(chunk.item["arguments"])
    tool_result = call_tool(tool_name, tool_args)
    yield {"type": "response.output_item.done", "item": create_function_call_output_item(call_id=chunk.item["call_id"], output=tool_result)}

class QuickstartAgent(ResponsesAgent):
  def predict_stream(self, request: ResponsesAgentRequest):
    # Extract the user's prompt from the request
    prompt = request.input[-1].content
    # Stream response items from our agent
    for chunk in run_agent(prompt):
      yield ResponsesAgentStreamEvent(**chunk)

  def predict(self, request: ResponsesAgentRequest) -> ResponsesAgentResponse:
    outputs = [
      event.item
      for event in self.predict_stream(request)
      if event.type == "response.output_item.done"
    ]
    return ResponsesAgentResponse(output=outputs)

AGENT = QuickstartAgent()
mlflow.models.set_model(AGENT)

エージェントのログを記録する

エージェントをログに記録し、 Unity Catalogに登録するこれにより、エージェントとその依存関係がデプロイ用の 1 つのアーティファクトにパッケージ化されます。

Python
import mlflow
from mlflow.models.resources import DatabricksFunction, DatabricksServingEndpoint
from pkg_resources import get_distribution

# Change the catalog name ("main") and schema name ("default") to register the agent to a different location
registered_model_name = "main.default.quickstart_agent"

# Specify Databricks resources that the agent needs to access.
# This step lets Databricks automatically configure authentication
# so the agent can access these resources when it's deployed.
resources = [
  DatabricksServingEndpoint(endpoint_name="databricks-claude-sonnet-4-5"),
  DatabricksFunction(function_name="system.ai.python_exec"),
]

mlflow.set_registry_uri("databricks-uc")
logged_agent_info = mlflow.pyfunc.log_model(
  artifact_path="agent",
  python_model="quickstart_agent.py",
  extra_pip_requirements=[f"databricks-connect=={get_distribution('databricks-connect').version}"],
  resources=resources,
  registered_model_name=registered_model_name
)

エージェントをデプロイする

登録済みエージェントをサービスエンドポイントにデプロイします。

Python
from databricks import agents

deployment_info = agents.deploy(
  model_name=registered_model_name,
  model_version=logged_agent_info.registered_model_version,
  scale_to_zero=True
)

エージェントエンドポイントが起動したら、 AI Playground を使用してチャットしたり、関係者と共有してフィードバックを得たりできます。

次のステップ

目標に基づいて次に進む場所を選択してください。

エージェントの品質を測定して改善する : 「エージェント評価のクイックスタート」を参照してください。

より高度なエージェントの構築 :非構造化データを使用してRAGを実行し、マルチターンの会話を処理し、エージェント評価を使用して品質を測定するエージェントを作成します。チュートリアル: 取得エージェントのビルド、評価、デプロイを参照してください。

他のフレームワークを使用してエージェントを構築する方法を学ぶ : LangGraph、純粋なPython、OpenAIなどの一般的なライブラリを使用してエージェントを構築する方法を学びます。「コードで AI エージェントを作成する」を参照してください

必要条件​

ノートブックの例​

Mosaic AIエージェントのデモ

エージェントの定義​

エージェントをテストする​

エージェントをデプロイする​

デプロイ用のエージェント コードの準備​

エージェントのログを記録する​

エージェントをデプロイする​

次のステップ​

必要条件

ノートブックの例

エージェントの定義

エージェントをテストする

エージェントをデプロイする

デプロイ用のエージェントコードの準備

エージェントのログを記録する

エージェントをデプロイする

次のステップ