Kom igång med AI-agenter

Skapa din första AI-agent med Hjälp av Mosaic AI Agent Framework. I den här handledningen kommer du att:

Skapa en agent med Agent Framework.
Lägg till ett verktyg i din agent.
Distribuera din agent till en Databricks-modell som betjänar slutpunkten.

En konceptuell introduktion till agenter och andra gen-AI-appar finns i Vad är gen AI-appar?

Kravspecifikation

Arbetsytan måste ha följande funktioner aktiverade:

Unity-katalog
Mosaic AI Agent Framework
Grundmodeller (betala per token, etablerat dataflöde eller externa modeller). Se Funktioner med begränsad regional tillgänglighet

Exempelanteckningsbok

Den här notebook-filen innehåller all kod som du behöver för att skapa och distribuera din första AI-agent. Importera anteckningsboken till din Azure Databricks-arbetsyta för att köra den.

Demo av Mosaic AI-agent

Hämta anteckningsbok

Definiera agenten

En AI-agent består av följande:

En stor språkmodell (LLM) som kan resonera och fatta beslut
Verktyg som LLM kan använda för att göra mer än att bara generera text, till exempel att köra Python-kod eller hämta data

Kör följande kod i en Databricks-anteckningsbok för att definiera en enkel verktygssamtalsagent:

Installera nödvändiga Python-paket:
```
%pip install -U -qqqq mlflow databricks-openai databricks-agents"
dbutils.library.restartPython()
```
- mlflow: Används för agentutveckling och agentspårning.
- databricks-openai: Används för att ansluta till Databricks-värdbaserade LLM och komma åt Unity Catalog-verktyg.
- databricks-agents: Används för att paketera och distribuera agenten.

Definiera agenten. Det här kodfragmentet gör följande:

Ansluter till Databricks-modellen som betjänar slutpunkten med hjälp av OpenAI-klienten.
Aktiverar MLflow-spårning med .autolog() Detta lägger till instrumentation så att du kan se vad din agent gör när du skickar en fråga.
Lägger till system.ai.python_exec-verktyget till din agent. Med den här inbyggda Unity Catalog-funktionen kan din agent köra Python-kod.
Använder MLflow-hjälpfunktioner (output_to_responses_items_stream, create_function_call_output_item) för att konvertera strömmande LLM-utdata till ett API-kompatibelt svarsformat.

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

Testa agenten

Testa agenten genom att fråga den med en fråga som kräver att Python-kod körs:

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

Förutom LLM:s resultat ser du detaljerad spårningsinformation direkt i din notebook. Dessa spårningar hjälper dig att felsöka långsamma eller misslyckade agentanrop. Dessa spårningar lades automatiskt till med hjälp av mlflow.openai.autolog() .

Distribuera agenten

Nu när du har en agent kan du paketera och distribuera den till en Databricks-serverslutpunkt. Börja samla in feedback om en distribuerad agent genom att dela den med andra och chatta med den med hjälp av ett inbyggt chattgränssnitt.

Förbereda agentkod för distribution

Om du vill förbereda agentkoden för distributionen omsluter du den med MLflows ResponsesAgent gränssnitt. Gränssnittet ResponsesAgent är det rekommenderade sättet att paketera agenter för distribution på Azure Databricks.

Om du vill implementera ResponsesAgent gränssnittet definierar du metoderna både predict_stream() (för direktuppspelningssvar) och predict() (för icke-strömmande begäranden). Eftersom den underliggande agentlogiken redan matar ut API-kompatibla händelser för svar är implementeringen enkel:

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)

Lägg till följande kod i notebook-filen för att testa klassen 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)

Kombinera all agentkod i en enda fil så att du kan logga och distribuera den.

Konsolidera all din agentkod till en notebook-cell.
Överst i cellen lägger du till det %%writefile quickstart_agent.py magiska kommandot för att spara agenten i en fil.
"Anropa mlflow.models.set_model() med ditt agentobjekt längst ned i cellen." Detta talar om för MLflow vilket agentobjekt som ska användas när förutsägelser hanteras. Det här steget konfigurerar effektivt startpunkten för vår agentkod.

Notebook-cellen bör se ut så här:

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

Logga agenten

Logga din agent och registrera den i Unity Catalog. Detta paketerar din agent och dess beroenden i en enda artefakt för distribution.

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
)

Distribuera agenten

Distribuera din registrerade agent till en serverslutpunkt:

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
)

När agentslutpunkten har startat kan du chatta med den med AI Playground eller dela den med intressenter för feedback.

Nästa steg

Välj vart du ska gå härnäst baserat på dina mål:

Mät och förbättra agentens kvalitet: Se Snabbstart för agentutvärdering.

Skapa mer avancerade agenter: Skapa en agent som utför RAG med ostrukturerade data, hanterar konversationer med flera turer och använder Agentutvärdering för att mäta kvalitet. Se Självstudie: Skapa, utvärdera och distribuera en hämtningsagent.

Lär dig hur du skapar agenter med andra ramverk: Lär dig hur du skapar agenter med hjälp av populära bibliotek som LangGraph, ren Python och OpenAI. Se Skapa AI-agenter i kod

Feedback

Var den här sidan till hjälp?

Last updated on 2025-12-03

Dela via

Kom igång med AI-agenter

Kravspecifikation

Exempelanteckningsbok

Demo av Mosaic AI-agent

Definiera agenten

Testa agenten

Distribuera agenten

Förbereda agentkod för distribution

Logga agenten

Distribuera agenten

Nästa steg

Feedback

Ytterligare resurser