開始使用 AI 代理程式

使用馬賽克 AI 代理程式架構建置您的第一個 AI 代理程式。在本教學課程中，您將：

使用 Agent Framework 撰寫代理程式。
將工具新增至您的代理程式。
將您的代理程式部署到提供端點的 Databricks 模型。

如需代理程式和其他 Gen AI 應用程式的概念簡介，請參閱什麼是 Gen AI 應用程式？

需求

您的工作區必須啟用下列功能：

Unity 目錄
馬賽克智能代理框架
基礎模型（按令牌付費、設定的吞吐量或外部模型）。請參閱具有有限區域可用性的功能

範例筆記本

此筆記本包含撰寫及部署第一個 AI 代理程式所需的所有程式代碼。將筆記本匯入 Azure Databricks 工作區以執行。

馬賽克 AI 代理人演示

拿筆記本

定義代理程式

AI 代理程式包含下列各項：

大型語言模型（LLM）可以推理和做出決策
LLM 可用來執行不只是產生文字的工具，例如執行 Python 程式代碼或擷取數據

在 Databricks 筆記本中執行下列程式代碼，以定義簡單的工具呼叫代理程式：

安裝必要 Python 套件：
```
%pip install -U -qqqq mlflow databricks-openai databricks-agents"
dbutils.library.restartPython()
```
- mlflow：用於代理開發和代理追蹤。
- databricks-openai：用於連接 Databricks 託管的大型語言模型並存取 Unity 目錄工具。
- databricks-agents：用來封裝和部署代理程式。

定義代理程式。此代碼段會執行下列動作：

使用 OpenAI 用戶端連接到 Databricks 模型服務端點。
使用 autolog()啟用 MLflow 追蹤。這會新增檢測，讓您可以在提交查詢時查看代理程式的功能。
將 system.ai.python_exec 工具新增至您的代理程式。此內建 Unity 目錄函式可讓您的代理程式執行 Python 程式代碼。
使用 MLflow 輔助函式（output_to_responses_items_stream， create_function_call_output_item）將串流 LLM 輸出轉換為相容 Responses API 格式。

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-3-7-sonnet",
    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 程式代碼的提示來查詢代理程式，以測試代理程式：

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

除了 LLM 的輸出之外，您也會直接在筆記本中看到詳細的追蹤資訊。這些追蹤可協助您偵錯緩慢或失敗的代理程式呼叫。這些追蹤是使用 mlflow.openai.autolog() 自動新增的。

部署代理程式

現在您已有代理程式，您可以封裝並部署至 Databricks 服務端點。開始收集已部署代理的意見反饋，可以先與他人共享，然後使用內建聊天室介面與其聊天。

準備代理程式代碼以進行部署

若要準備代理程式代碼以進行部署，請使用 MLflow 的 ResponsesAgent 介面加以包裝。介面 ResponsesAgent 是封裝代理程式以在 Azure Databricks 上部署的建議方式。

要實作 ResponsesAgent 介面，必須定義 predict_stream()（用於串流回應）和 predict()（用於非串流請求）兩種方法。由於底層代理邏輯已輸出 Responses 與 API 相容的事件，實作方式相當直接：

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 類別：

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)

將所有代理程式代碼合併成單一檔案，以便記錄和部署它。

將所有代理程式代碼合併到一個筆記本數據格中。
在儲存格頂端，新增 %%writefile quickstart_agent.py Magic 命令以將代理程式儲存至檔案。
在儲存格底部，使用您的代理程式物件呼叫 mlflow.models.set_model() 。這會告訴 MLflow 在提供預測時要使用的代理程序物件。此步驟可有效設定我們代理程式碼的進入點。

您的筆記本儲存格應該看起來如下所示：

%%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-3-7-sonnet",
    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。這會將您的代理程式及其相依性封裝到單一成品以進行部署。

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,
  scale_to_zero=True
)

代理程式端點啟動之後，您可以使用 AI 遊樂場與其聊天，或與項目關係人分享以取得意見反應。

後續步驟

根據您的目標選擇下一步的位置：

測量並改善代理程序的品質：請參閱代理程序評估快速入門。

建置更進階的代理程式：建立使用非結構化數據執行RAG的代理程式、處理多回合交談，並使用代理程式評估來測量品質。請參閱教學課程：建置、評估和部署擷取代理程式。

瞭解如何使用其他架構建置代理程序：瞭解如何使用 LangGraph、純 Python 和 OpenAI 等熱門連結庫來建置代理程式。請參閱在程式代碼中撰寫 AI 代理程式

意見反應

此頁面對您有幫助嗎？

Last updated on 2025-12-03

共用方式為

開始使用 AI 代理程式

需求

範例筆記本

馬賽克 AI 代理人演示

定義代理程式

測試代理程式

部署代理程式

準備代理程式代碼以進行部署

記錄代理程式

部署代理程式

後續步驟

意見反應

其他資源