Začínáme s agenty AI

Sestavte svého prvního agenta AI pomocí architektury agenta Mosaic AI. V tomto kurzu:

Vytvoření agenta pomocí rozhraní Agent Framework
Agentovi přidejte nástroj.
Nasaďte agenta do koncového bodu obsluhy modelu Databricks.

Pro koncepční úvod do agentů a dalších aplikací generativní umělé inteligence viz Koncepty: Generativní umělá inteligence v Azure Databricks

Požadavky

Váš pracovní prostor musí mít povolené následující funkce:

Katalog Unity
Architektura agenta systému Mosaic AI
Základní modely (pay-per-token, zřízená propustnost nebo externí modely). Viz Funkce s omezenou regionální dostupností

ukázkový poznámkový blok

Tento poznámkový blok obsahuje veškerý kód, který potřebujete k vytvoření a nasazení prvního agenta AI. Naimportujte poznámkový blok do pracovního prostoru Azure Databricks a spusťte ho.

Ukázka agenta AI v systému Mosaic

Pořiďte si notebook

Definování agenta

Agent AI se skládá z následujících:

Velký jazykový model (LLM), který může zdůvodnit a rozhodovat
Nástroje, které LLM může použít k provádění více než jen generování textu, jako je spuštění kódu Pythonu nebo načítání dat

Spuštěním následujícího kódu v poznámkovém bloku Databricks definujte jednoduchého agenta pro volání nástrojů:

Nainstalujte požadované balíčky Pythonu:
```
%pip install -U -qqqq mlflow databricks-openai databricks-agents"
dbutils.library.restartPython()
```
- mlflow: Používá se pro vývoj agentů a trasování agentů.
- databricks-openai: Slouží k připojení k LLM hostovaným Databricks a přístupu k nástrojům Unity Catalog.
- databricks-agents: Slouží k zabalení a nasazení agenta.

Definujte agenta. Tento fragment kódu provede následující:

Připojí se k modelu Databricks obsluhující koncový bod pomocí klienta OpenAI.
Umožňuje trasování MLflow pomocí autolog(). Tím se přidá instrumentace, abyste viděli, co váš agent dělá při odesílání dotazu.
Přidává nástroj system.ai.python_exec do vašeho agenta. Tato integrovaná funkce katalogu Unity umožňuje vašemu agentu spouštět kód Pythonu.
Používá pomocné funkce MLflow (output_to_responses_items_stream, create_function_call_output_item) k převodu datového toku LLM na formát kompatibilní s 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-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)}

Otestujte agenta

Otestujte agenta dotazováním pomocí výzvy, která vyžaduje spuštění kódu Pythonu:

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

Kromě výstupu LLM uvidíte podrobné záznamy přímo ve vašem poznámkovém bloku. Tyto záznamy vám pomohou ladit pomalá nebo neúspěšná volání agenta. Tyto stopy byly automaticky přidány pomocí mlflow.openai.autolog().

Nasazení agenta

Teď, když máte agenta, můžete ho zabalit a nasadit do koncového bodu služby Databricks. Začněte shromažďovat zpětnou vazbu na nasazeného agenta tím, že ho nasdílíte s ostatními a budete s ním chatovat pomocí integrovaného chatovacího uživatelského rozhraní.

Příprava kódu agenta pro nasazení

Pokud chcete připravit kód agenta pro nasazení, zabalte ho ResponsesAgent pomocí rozhraní MLflow. Rozhraní ResponsesAgent je doporučený způsob balení agentů pro nasazení v Azure Databricks.

Pokud chcete implementovat ResponsesAgent rozhraní, definujte predict_stream() metody (pro odpovědi na streamování) i predict() (pro požadavky bez streamování). Vzhledem k tomu, že logika základního agenta už vypíše události kompatibilní s rozhraním API pro odpovědi, implementace je jednoduchá:

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)

Přidejte následující kód do svého poznámkového bloku pro otestování třídy 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)

Zkombinujte veškerý kód agenta do jednoho souboru, abyste ho mohli protokolovat a nasadit.

Sloučit veškerý kód agenta do jedné buňky poznámkového bloku
V horní části buňky přidejte magický příkaz %%writefile quickstart_agent.py pro uložení agenta do souboru.
V dolní části buňky použijte mlflow.models.set_model(), abyste zavolali objekt agenta. To říká MLflow, který objekt agenta se má použít při poskytování předpovědí. Tento krok efektivně nakonfiguruje vstupní bod do kódu agenta.

Buňka poznámkového bloku by měla vypadat takto:

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

Zaznamenejte agenta

Zaregistrujte agenta a zaregistrujte ho do katalogu Unity. Tím se váš agent a jeho závislosti zabalí do jednoho artefaktu pro nasazení.

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
)

Nasazení agenta

Nasaďte registrovaného agenta na obslužný koncový bod:

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
)

Jakmile se koncový bod agenta spustí, můžete s ním chatovat pomocí AI Playgroundu nebo ho sdílet se zúčastněnými stranami , abyste mohli získat zpětnou vazbu.

Další kroky

Na základě svých cílů zvolte, kam se chcete vrátit:

Měřte a zlepšujte kvalitu svého agenta: Podívejte se na rychlý start hodnocení agenta.

Vytváření pokročilejších agentů: Vytvořte agenta, který provádí RAG pomocí nestrukturovaných dat, zpracovává konverzace s více kroky a používá hodnocení agentů k měření kvality. Viz kurz: Sestavení, vyhodnocení a nasazení agenta načítání.

Naučte se vytvářet agenty pomocí jiných architektur: Naučte se vytvářet agenty pomocí oblíbených knihoven, jako jsou LangGraph, pure Python a OpenAI. Viz Vytváření agentů AI v kódu

Váš názor

Byla tato stránka užitečná?

Last updated on 2025-12-09

Sdílet prostřednictvím

Začínáme s agenty AI

Požadavky

ukázkový poznámkový blok

Ukázka agenta AI v systému Mosaic

Definování agenta

Otestujte agenta

Nasazení agenta

Příprava kódu agenta pro nasazení

Zaznamenejte agenta

Nasazení agenta

Další kroky

Váš názor

Další materiály