Erste Schritte mit KI-Agents

Erstellen Sie Ihren ersten KI-Agent mit Mosaik AI Agent Framework. In diesem Tutorial werden Sie Folgendes lernen:

Erstellen Sie einen Agent mit Agent Framework.
Fügen Sie Ihrem Agent ein Tool hinzu.
Stellen Sie Ihren Agenten an einem Databricks-Modellbereitstellungs-Endpunkt bereit.

Eine konzeptionelle Einführung in Agenten und andere generative KI-Apps finden Sie unter Was sind generative KI-Apps?

Anforderungen

Ihr Arbeitsbereich muss die folgenden Features aktiviert haben:

Unity-Katalog
Mosaik AI Agent Framework
Foundation-Modelle (Pay-per-Token, bereitgestellter Durchsatz oder externe Modelle). Features mit eingeschränkter regionaler Verfügbarkeit sehen

Beispielnotizbuch

Dieses Notizbuch enthält den gesamten Code, den Sie zum Erstellen und Bereitstellen Ihres ersten KI-Agents benötigen. Importieren Sie das Notizbuch in Ihren Azure Databricks-Arbeitsbereich, um es auszuführen.

Demo des Mosaic-AI-Agenten

Notebook abrufen

Definieren des Agents

Ein KI-Agent besteht aus folgenden Komponenten:

Ein großes Sprachmodell (LLM), das argumentieren und Entscheidungen treffen kann
Tools, die der LLM verwenden kann, um mehr zu tun als nur Text zu generieren, z. B. Ausführen von Python-Code oder Abrufen von Daten

Führen Sie den folgenden Code in einem Databricks-Notizbuch aus, um einen einfachen Toolanruf-Agent zu definieren:

Installieren Sie die erforderlichen Python-Pakete:
```
%pip install -U -qqqq mlflow databricks-openai databricks-agents"
dbutils.library.restartPython()
```
- mlflow: Wird für die Agententwicklung und Agent-Tracing verwendet.
- databricks-openai: Wird verwendet, um eine Verbindung mit den von Databricks gehosteten LLM herzustellen und auf Unity Catalog-Tools zuzugreifen.
- databricks-agents: Wird zum Packen und zur Bereitstellung des Agents verwendet.

Definieren Sie den Agent. Dieser Codeausschnitt führt die folgenden Aktionen aus:

Stellt mithilfe des OpenAI-Clients eine Verbindung mit dem Databricks-Modell her, das den Endpunkt bedient.
Aktiviert die MLflow-Ablaufverfolgung mithilfe von autolog(). Dadurch wird eine Instrumentierung hinzugefügt, damit Sie sehen können, was Ihr Agent tut, wenn Sie eine Abfrage senden.
Fügt das system.ai.python_exec Tool Ihrem Agent hinzu. Mit dieser integrierten Unity Catalog-Funktion kann Ihr Agent Python-Code ausführen.
Verwendet MLflow-Hilfsfunktionen (output_to_responses_items_stream, create_function_call_output_item), um die Streaming-LLM-Ausgabe in ein mit der Antwort-API kompatibles Format zu konvertieren.

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)}

Testen des Agents

Testen Sie den Agent, indem Sie ihn mit einer Eingabeaufforderung abfragen, für die Python-Code ausgeführt werden muss:

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

Zusätzlich zur LLM-Ausgabe werden detaillierte Protokolldaten direkt in Ihrem Notizbuch angezeigt. Diese Ablaufverfolgungen helfen Ihnen beim Debuggen langsamer oder fehlgeschlagener Agentanrufe. Diese Spuren wurden automatisch mithilfe von mlflow.openai.autolog() hinzugefügt.

Bereitstellen des Agents

Nachdem Sie nun über einen Agent verfügen, können Sie ihn in einem Databricks-Dienstendpunkt packen und bereitstellen. Beginnen Sie mit dem Sammeln von Feedback zu einem bereitgestellten Agent, indem Sie es mit anderen teilen und mit diesem mithilfe einer integrierten Chatbenutzeroberfläche chatten.

Vorbereiten des Agentcodes für die Bereitstellung

Um Ihren Agent-Code für die Bereitstellung vorzubereiten, umschließen Sie ihn mithilfe der MLflow-Schnittstelle ResponsesAgent. Die ResponsesAgent Schnittstelle ist die empfohlene Methode zum Packen von Agents für die Bereitstellung in Azure Databricks.

Um die ResponsesAgent Schnittstelle zu implementieren, definieren Sie sowohl (für Streamingantworten) als predict_stream() auch predict() (für Nicht-Streaming-Anforderungen) Methoden. Da die zugrunde liegende Agentlogik bereits mit der Antwort-API kompatible Ereignisse ausgibt, ist die Implementierung einfach:

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)

Fügen Sie Ihrem Notizbuch den folgenden Code hinzu, um die ResponsesAgent Klasse zu testen.

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)

Kombinieren Sie ihren gesamten Agentcode in einer einzigen Datei, damit Sie ihn protokollieren und bereitstellen können.

Konsolidieren Sie Ihren gesamten Agentcode in einer Notebook-Zelle.
Fügen Sie oben in der Zelle den %%writefile quickstart_agent.py magischen Befehl hinzu, um Ihren Agent in einer Datei zu speichern.
Am unteren Rand der Zelle rufen Sie mlflow.models.set_model() mit Ihrem Agentenobjekt auf. Dadurch wird MLflow mitgeteilt, welches Agentobjekt beim Verarbeiten von Vorhersagen verwendet werden soll. In diesem Schritt wird der Einstiegspunkt für unseren Agentcode optimal konfiguriert.

Ihre Notizbuchzelle sollte wie folgt aussehen:

%%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)

Protokollieren des Agents

Protokollieren Sie Ihre Agenten sowie registrieren Sie diese im Unity-Katalog. Dadurch wird Ihr Agent und seine Abhängigkeiten in ein einzelnes Artefakt für die Bereitstellung verpackt.

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
)

Bereitstellen des Agents

Stellen Sie Ihren registrierten Agenten an einem Bereitstellungspunkt bereit.

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
)

Nachdem der Agentendpunkt gestartet wurde, können Sie mit dem Agent-Playground chatten oder ihn mit den Projektbeteiligten teilen , um Feedback zu erhalten.

Nächste Schritte

Wählen Sie, wo Sie als Nächstes hingehen möchten basierend auf Ihren Zielen.

Messen und verbessern Sie die Qualität Ihres Agenten: Siehe Schnellstart zur Agentauswertung.

Erstellen Sie fortgeschrittene Agenten: Erstellen Sie einen Agenten, der RAG mithilfe unstrukturierter Daten durchführt, mehrere Gesprächsrunden verarbeitet und die Agentenbewertung verwendet, um die Qualität zu messen. Siehe Lernprogramm: Erstellen, Auswerten und Bereitstellen eines Abruf-Agents.

Erfahren Sie, wie Sie Agents mit anderen Frameworks erstellen: Erfahren Sie, wie Sie Agents mit beliebten Bibliotheken wie LangGraph, pure Python und OpenAI erstellen. Anzeigen von Autoren-KI-Agents im Code

Feedback

War diese Seite hilfreich?

Last updated on 2025-12-09

Freigeben über

Erste Schritte mit KI-Agents

Anforderungen

Beispielnotizbuch

Demo des Mosaic-AI-Agenten

Definieren des Agents

Testen des Agents

Bereitstellen des Agents

Vorbereiten des Agentcodes für die Bereitstellung

Protokollieren des Agents

Bereitstellen des Agents

Nächste Schritte

Feedback

Zusätzliche Ressourcen