AI エージェントの概要

2025-05-14

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

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

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

要求事項

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

Unity カタログ
モザイク人工知能エージェントフレームワーク
基盤モデル (トークンごとの支払い、プロビジョニングされたスループット、または外部モデル)。リージョンの可用性が制限された機能を参照してください

ノートブックの例

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

モザイクAIエージェントデモ

ノートブックを入手

エージェントを定義する

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

意思決定を行うことができる大規模言語モデル (LLM)
PYTHON コードの実行やデータのフェッチなど、テキストの生成以外にも LLM が使用できるツール

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

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

エージェントを定義します。このコードスニペットでは、次の処理が行われます。

OpenAI クライアントを使用して、エンドポイントを提供する Databricks モデルに接続します。
autolog()を使用して MLflow トレースを有効にします。これによりインストルメンテーションが追加され、クエリの送信時にエージェントが何を行うかを確認できます。
system.ai.python_exec ツールをエージェントに追加します。この組み込みの Unity カタログ関数を使用すると、エージェントで Python コードを実行できます。
プロンプトを使用して LLM にクエリを実行し、応答を処理する関数を定義します。

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

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

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

# 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)
  raise ValueError(f"Unknown tool: {tool_name}")

def run_agent(prompt):
  """
  Send a user prompt to the LLM and return a list of LLM response messages
  The LLM is allowed to call the code interpreter tool, if needed, to respond to the user
  """
  result_msgs = []
  response = openai_client.chat.completions.create(
    model="databricks-claude-3-7-sonnet",
    messages=[{"role": "user", "content": prompt}],
    tools=builtin_tools,
  )
  msg = response.choices[0].message
  result_msgs.append(msg.to_dict())

  # If the model executed a tool, call it
  if msg.tool_calls:
    call = msg.tool_calls[0]
    tool_result = call_tool(call.function.name, json.loads(call.function.arguments))
    result_msgs.append({"role": "tool", "content": tool_result.value, "name": call.function.name, "tool_call_id": call.id})
  return result_msgs

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

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

answer = run_agent("What is the 100th fibonacci number?")
for message in answer:
  print(f'{message["role"]}: {message["content"]}')

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

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

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

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

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

ChatAgent インターフェイスを実装するには、ユーザーのメッセージをエージェントに送信し、エージェントの応答を収集して、predict()形式で返すChatAgentResponses関数を定義する必要があります。

import uuid
from typing import Any, Optional

from mlflow.pyfunc import ChatAgent
from mlflow.types.agent import ChatAgentMessage, ChatAgentResponse, ChatContext

class QuickstartAgent(ChatAgent):
  def predict(
    self,
    messages: list[ChatAgentMessage],
    context: Optional[ChatContext] = None,
    custom_inputs: Optional[dict[str, Any]] = None,
  ) -> ChatAgentResponse:
    # 1. Extract the last user prompt from the input messages
    prompt = messages[-1].content

    # 2. Call run_agent to get back a list of response messages
    raw_msgs = run_agent(prompt)

    # 3. Map each response message into a ChatAgentMessage and return
    # the response
    out = []
    for m in raw_msgs:
      out.append(ChatAgentMessage(
        id=uuid.uuid4().hex,
        **m
      ))

    return ChatAgentResponse(messages=out)

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

AGENT = QuickstartAgent()
for response_message in AGENT.predict({"messages": [{"role": "user", "content": "What's the 100th fibonacci number?"}]}).messages:
  print(f"role: {response_message.role}, content: {response_message.content}")

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

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

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

%%writefile quickstart_agent.py

import json
import uuid
from databricks.sdk import WorkspaceClient
from databricks_openai import UCFunctionToolkit, DatabricksFunctionClient
from typing import Any, Optional

import mlflow
from mlflow.pyfunc import ChatAgent
from mlflow.types.agent import ChatAgentMessage, ChatAgentResponse, ChatContext

# Add an mlflow.openai.autolog() call to capture traces in the serving endpoint

# Get an OpenAI client configured to talk to Databricks model serving endpoints
# We'll use this to query an LLM in our agent
openai_client = WorkspaceClient().serving_endpoints.get_open_ai_client()

# Load Databricks built-in tools (a stateless Python code interpreter tool)
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)
  raise ValueError(f"Unknown tool: {tool_name}")

def run_agent(prompt):
  """
  Send a user prompt to the LLM, and return a list of LLM response messages
  The LLM is allowed to call the code interpreter tool if needed, to respond to the user
  """
  result_msgs = []
  response = openai_client.chat.completions.create(
    model="databricks-claude-3-7-sonnet",
    messages=[{"role": "user", "content": prompt}],
    tools=builtin_tools,
  )
  msg = response.choices[0].message
  result_msgs.append(msg.to_dict())

  # If the model executed a tool, call it
  if msg.tool_calls:
    call = msg.tool_calls[0]
    tool_result = call_tool(call.function.name, json.loads(call.function.arguments))
    result_msgs.append({"role": "tool", "content": tool_result.value, "name": call.function.name, "tool_call_id": call.id})
  return result_msgs

class QuickstartAgent(ChatAgent):
  def predict(
    self,
    messages: list[ChatAgentMessage],
    context: Optional[ChatContext] = None,
    custom_inputs: Optional[dict[str, Any]] = None,
  ) -> ChatAgentResponse:
    prompt = messages[-1].content
    raw_msgs = run_agent(prompt)
    out = []
    for m in raw_msgs:
      out.append(ChatAgentMessage(
        id=uuid.uuid4().hex,
        **m
      ))

    return ChatAgentResponse(messages=out)

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

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

エージェントをログに記録し、Unity カタログに登録します。これにより、エージェントとその依存関係がデプロイ用の 1 つの成果物にパッケージ化されます。

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-3-7-sonnet"),
  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
)

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

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

from databricks import agents

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

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

次のステップ

目標に基づいて次に進む場所を選択します。

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

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

他のフレームワークを使用してエージェントを構築する方法について説明します。LangGraph、純粋な Python、OpenAI などの一般的なライブラリを使用してエージェントを構築する方法について説明します。「ChatAgentを使用してエージェントを作成する」を参照してください

次の方法で共有

AI エージェントの概要

要求事項

ノートブックの例

モザイクAIエージェントデモ

エージェントを定義する

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

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

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

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

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

次のステップ

フィードバック

その他のリソース

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