Bien démarrer avec les agents IA

Créez votre premier agent IA à l’aide de Mosaic AI Agent Framework. Dans ce tutoriel, vous allez :

Créez un agent à l’aide d’Agent Framework.
Ajoutez un outil à votre agent.
Déployez votre agent sur un point de terminaison de service de modèle Databricks.

Pour une présentation conceptuelle des agents et d’autres applications IA de génération, consultez Présentation des applications IA de génération ?

Spécifications

Votre espace de travail doit avoir les fonctionnalités suivantes activées :

Catalogue Unity
Cadre de l’assistant IA de Mosaic
Modèles de base (paiement par jeton, débit provisionné ou modèles externes). Afficher les fonctionnalités avec une disponibilité régionale limitée

Exemple de bloc-notes

Ce notebook contient tout le code dont vous avez besoin pour créer et déployer votre premier agent IA. Importez le notebook dans votre espace de travail Azure Databricks pour l’exécuter.

Démonstration de l’agent Mosaïque AI

Obtenir un ordinateur portable

Définir l’agent

Un agent IA se compose des éléments suivants :

Modèle de langage volumineux (LLM) qui peut raisonner et prendre des décisions
Outils que le LLM peut utiliser pour effectuer plus que simplement générer du texte, par exemple exécuter du code Python ou extraire des données

Exécutez le code suivant dans un notebook Databricks pour définir un agent d’appel d’outils simple :

Installez les packages Python requis :
```
%pip install -U -qqqq mlflow databricks-openai databricks-agents"
dbutils.library.restartPython()
```
- mlflow: utilisé pour le développement de l’agent et son suivi.
- databricks-openai: permet de se connecter au LLM hébergé par Databricks et d’accéder aux outils du catalogue Unity.
- databricks-agents: utilisé pour empaqueter et déployer l’agent.

Définissez l’agent. Cet extrait de code effectue les opérations suivantes :

Se connecte au point de terminaison de service du modèle Databricks à l'aide du client OpenAI.
Active le suivi MLflow à l’aide de autolog(). Cela ajoute l’instrumentation afin de voir ce que fait votre agent lorsque vous envoyez une requête.
Ajoute l’outil system.ai.python_exec à votre agent. Cette fonction de catalogue Unity intégrée permet à votre agent d’exécuter du code Python.
Utilise les fonctions d’assistance MLflow (output_to_responses_items_stream, create_function_call_output_item) pour convertir la sortie LLM en streaming dans un format compatible avec l’API Réponses.

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

Tester l’agent

Testez l’agent en l’interrogeant avec une invite qui nécessite l’exécution du code Python :

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

En plus de la sortie de LLM, vous verrez des informations de trace détaillées directement dans votre bloc-notes. Ces traces vous aident à déboguer des appels d’agent lents ou ayant échoué. Ces traces ont été ajoutées automatiquement à l’aide de mlflow.openai.autolog().

Déployer l’agent

Maintenant que vous disposez d’un agent, vous pouvez empaqueter et le déployer sur un point de terminaison de service Databricks. Commencez à recueillir des commentaires sur un agent déployé en le partageant avec d’autres personnes et en discutant avec lui à l’aide d’une interface utilisateur de conversation intégrée.

Préparer le code de l’agent pour le déploiement

Pour préparer votre code d’agent pour le déploiement, encapsulez-le à l’aide de l’interface de ResponsesAgent MLflow. L’interface ResponsesAgent est la méthode recommandée pour empaqueter des agents pour le déploiement sur Azure Databricks.

Pour implémenter l’interface ResponsesAgent, définissez à la fois les méthodes predict_stream() (pour les réponses en streaming) et predict() (pour les requêtes non streaming). Étant donné que la logique de l’agent sous-jacente génère déjà des événements compatibles avec l’API Réponses, l’implémentation est simple :

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)

Ajoutez le code suivant à votre notebook pour tester votre ResponsesAgent classe :

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)

Combinez tout votre code d’agent en un seul fichier pour pouvoir le journaliser et le déployer.

Consolidez tout votre code d’agent dans une cellule de bloc-notes.
En haut de la cellule, ajoutez la %%writefile quickstart_agent.py commande magique pour enregistrer votre agent dans un fichier.
En bas de la cellule, appelez mlflow.models.set_model() avec votre objet agent. Cela indique à MLflow quel objet agent utiliser lors du service des prédictions. Cette étape configure efficacement le point d’entrée sur notre code d’agent.

Votre cellule de bloc-notes doit se présenter comme suit :

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

Enregistrer l'agent

Connectez votre agent et enregistrez-le dans le catalogue Unity. Cela empaquette votre agent et ses dépendances en un seul artefact pour le déploiement.

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
)

Déployer l’agent

Déployez votre agent inscrit sur un point de terminaison de service :

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
)

Une fois le point de terminaison de l’agent démarré, vous pouvez discuter avec lui à l’aide d’AI Playground ou le partager avec les parties prenantes pour obtenir des commentaires.

Étapes suivantes

Choisissez l’emplacement suivant en fonction de vos objectifs :

Mesurez et améliorez la qualité de votre agent : consultez le guide de démarrage rapide Évaluation de l’agent.

Créer des agents plus avancés : créez un agent qui effectue RAG à l’aide de données non structurées, gère les conversations multitours et utilise l’évaluation de l’agent pour mesurer la qualité. Consultez le tutoriel : Générer, évaluer et déployer un agent de récupération.

Découvrez comment créer des agents à l’aide d’autres frameworks : découvrez comment créer des agents à l’aide de bibliothèques populaires telles que LangGraph, Python pur et OpenAI. Voir Créer des agents IA dans le code

Commentaires

Cette page a-t-elle été utile ?

Last updated on 2025-12-09

Partager via

Bien démarrer avec les agents IA

Spécifications

Exemple de bloc-notes

Démonstration de l’agent Mosaïque AI

Définir l’agent

Tester l’agent

Déployer l’agent

Préparer le code de l’agent pour le déploiement

Enregistrer l'agent

Déployer l’agent

Étapes suivantes

Commentaires

Ressources supplémentaires