Aller au contenu principal
Dernière mise à jour le

Créer un agent d'IA et le déployer sur Model Serving

info

Pour les nouveaux cas d'utilisation, Databricks recommande de déployer des agents sur Databricks Apps pour un contrôle total sur le code des agents, la configuration du serveur et le workflow de déploiement. Consultez Créer un agent IA et le déployer sur Databricks Apps. Pour migrer un agent existant, consultez Migrer un agent de Model Serving vers Databricks Apps.

Cette page montre comment créer un agent d'IA en Python en utilisant des agents personnalisés et des bibliothèques populaires de création d'agents comme LangGraph et OpenAI.

Exigences​

astuce

Databricks recommande d’installer la dernière version du client Python MLflow lors du développement d’agents.

Pour créer et déployer des agents en utilisant l'approche décrite sur cette page, installez les éléments suivants :

  • databricks-agents 1.2.0 ou supérieur
  • mlflow 3.1.3 ou plus
  • Python 3,10 ou version ultérieure.
    • Utilisez un compute serverless ou Databricks Runtime 13.3 LTS ou une version ultérieure pour répondre à cette exigence.
%pip install -U -qqqq databricks-agents mlflow

Databricks recommande également d’installer les packages d’intégration Databricks AI Bridge pour l’auteur des agents. Ces packages d’intégration fournissent une couche partagée d’APIs qui interagissent avec les fonctionnalités Databricks AI, telles que les agents Genie et la recherche AI, à travers les frameworks de création d’agents et les SDK.

%pip install -U -qqqq databricks-openai

Utiliser ResponsesAgent pour créer des agents​

Databricks recommande l'interface MLflow ResponsesAgent pour créer des agents de niveau production. ResponsesAgent vous permet de créer des agents avec n'importe quel framework tiers, puis de l'intégrer aux fonctionnalités Databricks AI pour des capacités robustes de journalisation, de traçage, d'évaluation, de déploiement et de monitoring.

Le schéma ResponsesAgent est compatible avec le schéma Responses d'OpenAI. Pour en savoir plus sur OpenAI Responses, consultez OpenAI : Réponses et ChatCompletion.

remarque

L’ancienne interface ChatAgent est toujours prise en charge sur Databricks. Cependant, pour les nouveaux agents, Databricks recommande d’utiliser la dernière version de MLflow et l’interface ResponsesAgent.

Consultez le schéma de l'agent d'entrée et de sortie hérité (Model Serving).

RéponsesAgent permet d'envelopper facilement les agents existants pour la compatibilité Databricks.

ResponsesAgent offre les avantages suivants :

  • Capacités d'agent avancées

    • Prise en charge multi-agent
    • Sortie en streaming : diffusez la sortie par petits blocs.
    • **Historique complet des messages d'appel d'outils** : Retourne plusieurs messages, y compris les messages d'appel d'outils intermédiaires, pour une qualité et une gestion des conversations améliorées.
    • Prise en charge de la confirmation de l'appel d'outil
    • Prise en charge des outils à exécution longue

  • Développement, déploiement et monitoring rationalisés

    • Créez des agents à l’aide de n’importe quel framework : enveloppez tout agent existant à l’aide de l’interface ResponsesAgent pour bénéficier d’une compatibilité prête à l’emploi avec AI Playground, Agent Evaluation et le monitoring des agents.
    • Interfaces d'édition typées : rédigez du code d'agent à l'aide de classes Python typées, en bénéficiant de l'autocomplétion des IDE et notebooks.
    • Inférence automatique de signature : MLflow déduit automatiquement ResponsesAgent signatures lors de l'enregistrement d'un agent, ce qui simplifie l'enregistrement et le déploiement. Consultez l'inférence de la signature du modèle lors de l'enregistrement.
    • **Traçage automatique** : MLflow trace predict predict_stream automatiquement vos fonctions et, agrégeant les réponses Stream pour une évaluation et un affichage plus faciles.
    • Tables d’inférence optimisées par AI Gateway : les tables d’inférence AI Gateway sont automatiquement activées pour les agents déployés, ce qui permet d’accéder à des métadonnées de Logs de requêtes détaillées.

Pour savoir comment créer un ResponsesAgent, consultez les exemples de la section suivante et la documentation MLflow - ResponsesAgent pour Model Serving.

ResponsesAgent exemples​

Les Notebooks suivants montrent comment créer ResponsesAgent en streaming et non-streaming à l'aide de bibliothèques populaires. Pour savoir comment étendre les capacités de ces agents, consultez Connecter les agents aux outils.

Agent de discussion simple OpenAI utilisant des modèles hébergés par Databricks

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

Agent d'appel d'outils OpenAI MCP

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

Agent d'appel d'outils OpenAI utilisant des modèles hébergés par Databricks

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

Agent d'appel d'outils OpenAI utilisant des modèles hébergés par OpenAI

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

Exemple multi‑agent​

Pour apprendre à créer un système multi-agents, consultez Utiliser Genie dans les systèmes multi-agents (Model Serving).

Exemple d'agent à états​

Pour apprendre à créer des agents avec état dotés d'une mémoire à court terme et à long terme en utilisant Lakebase comme magasin de mémoire, consultez Mémoire des agents IA (Model Serving).

Exemple d'agent non conversationnel​

Contrairement aux agents conversationnels qui gèrent les dialogues à plusieurs tours, les agents non conversationnels se concentrent sur l'exécution efficace de tâches bien définies. Cette architecture simplifiée permet un throughput plus élevé pour les requêtes indépendantes.

Pour apprendre à créer un agent non conversationnel, consultez Agents IA non conversationnels utilisant MLflow.

Et si vous avez déjà un agent ?​

Si vous disposez déjà d'un agent développé avec LangChain, LangGraph ou un framework similaire, vous n'avez pas besoin de réécrire votre agent pour l'utiliser sur Databricks. Au lieu de cela, enveloppez simplement votre agent existant avec l'interface MLflow ResponsesAgent :

  1. Écrire une classe wrapper Python qui hérite de mlflow.pyfunc.ResponsesAgent.

    Dans la classe wrapper, référencez l’agent existant en tant qu’attribut self.agent = your_existing_agent.

  2. La classe ResponsesAgent" nécessite l'implémentation d'une méthode predict" qui renvoie un ResponsesAgentResponse" pour gérer les requêtes de non-streaming. Voici un exemple du schéma ResponsesAgentResponses :

    Python
    import uuid
    # input as a dict
    {"input": [{"role": "user", "content": "What did the data scientist say when their Spark job finally completed?"}]}

    # output example
    ResponsesAgentResponse(
        output=[
            {
                "type": "message",
                "id": str(uuid.uuid4()),
                "content": [{"type": "output_text", "text": "Well, that really sparked joy!"}],
                "role": "assistant",
            },
        ]
    )

  3. Dans la fonction predict, convertissez les messages entrants de ResponsesAgentRequest au format attendu par l'agent. Après que l'agent génère une réponse, convertissez sa sortie en un objet ResponsesAgentResponse.

Consultez les exemples de code suivants pour savoir comment convertir les agents existants en ResponsesAgent:

Pour les agents non-streaming, convertissez les entrées et sorties dans la fonction predict.

Python
from uuid import uuid4

from mlflow.pyfunc import ResponsesAgent
from mlflow.types.responses import (
    ResponsesAgentRequest,
    ResponsesAgentResponse,
)


class MyWrappedAgent(ResponsesAgent):
    def __init__(self, agent):
        # Reference your existing agent
        self.agent = agent

    def predict(self, request: ResponsesAgentRequest) -> ResponsesAgentResponse:
        # Convert incoming messages to your agent's format
        # prep_msgs_for_llm is a function you write to convert the incoming messages
        messages = self.prep_msgs_for_llm([i.model_dump() for i in request.input])

        # Call your existing agent (non-streaming)
        agent_response = self.agent.invoke(messages)

        # Convert your agent's output to ResponsesAgent format, assuming agent_response is a str
        output_item = (self.create_text_output_item(text=agent_response, id=str(uuid4())),)

        # Return the response
        return ResponsesAgentResponse(output=[output_item])

Pour des exemples complets, consultez des exemplesResponsesAgent.

Réponses de streaming​

Le Streaming permet aux agents d'envoyer des réponses par blocs en temps réel au lieu d'attendre la réponse complète. Pour implémenter le streaming avec ResponsesAgent, émettez une série d'événements delta suivie d'un événement de finalisation :

  1. Émettre des événements delta : Envoyez plusieurs événements output_text.delta avec le même item_id pour stream des fragments de texte en temps réel.
  2. Terminer par l'événement d'achèvement : envoyez un événement response.output_item.done final avec le même item_id que les événements delta contenant le texte de sortie final complet.

Chaque événement delta Stream un bloc de texte au client. L'événement final terminé contient le texte de réponse complet et indique à Databricks d'effectuer les opérations suivantes :

  • Tracez la sortie de votre agent avec le traçage MLflow
  • Agréger les réponses Stream dans les tables d’inférence AI Gateway
  • Afficher le résultat complet dans l'interface utilisateur de l'AI Playground.

Propagation des erreurs de streaming​

Databricks propage les erreurs rencontrées lors du streaming avec le dernier jeton sous databricks_output.error. Il appartient au client appelant de gérer et de signaler correctement cette erreur.

Bash
{
  "delta": …,
  "databricks_output": {
    "trace": {...},
    "error": {
      "error_code": BAD_REQUEST,
      "message": "TimeoutException: Tool XYZ failed to execute."
    }
  }
}

Fonctionnalités avancées​

Entrées et sorties personnalisées​

Certains scénarios peuvent nécessiter des entrées d'agent supplémentaires, telles que client_type et session_id, ou des sorties comme des liens de source de récupération qui ne devraient pas être inclus dans l'historique de discussion pour les interactions futures.

Pour ces scénarios, MLflow ResponsesAgent prend en charge en mode natif les champs custom_inputs et custom_outputs. Vous pouvez accéder aux entrées personnalisées via request.custom_inputs dans tous les exemples liés ci-dessus dans Exemples de ResponsesAgent.

attention

L'application d'examen d'Agent Evaluation ne prend pas en charge le rendu des traces pour les agents avec des champs d'entrée supplémentaires.

Consultez les Notebooks suivants pour apprendre à définir des entrées et sorties personnalisées.

Fournissez custom_inputs dans l'AI Playground et examinez l'application​

Si votre agent accepte des entrées supplémentaires à l'aide du champ custom_inputs, vous pouvez fournir ces entrées manuellement dans l'AI Playground et l'application d'examen.

  1. Dans l'AI Playground ou l'application d'examen d'agent, sélectionnez l'icône d'engrenage Icône d'engrenage..

  2. Activer custom_inputs .

  3. Fournissez un objet JSON qui correspond au schéma d'entrée défini de votre agent.

    Fournissez des custom_inputs dans l'AI Playground.

Spécifier des schémas de récupération personnalisés​

Les agents d'IA utilisent couramment des récupérateurs pour trouver et interroger des données non structurées à partir d'index de recherche IA. Par exemple, pour les outils de récupérateur, consultez Connectez des agents à des données non structurées.

Tracez ces extracteurs au sein de votre agent à l'aide de spans RETRIEVER MLflow afin d'activer les fonctionnalités du produit Databricks, notamment :

  • Affichage automatique des links vers les documents source récupérés dans l'interface utilisateur de l'AI Playground
  • Exécution automatique des juges d'ancrage de récupération et de pertinence dans Agent Evaluation
remarque

Databricks recommande d’utiliser les outils de récupération fournis par les packages Databricks AI Bridge comme databricks_langchain.VectorSearchRetrieverTool et databricks_openai.VectorSearchRetrieverTool, car ils sont déjà conformes au schéma de récupération MLflow. Consultez Développer un outil de récupération localement à l'aide d'AI Bridge.

Si votre agent inclut des étendues de récupérateur avec un schéma personnalisé, appelez mlflow.models.set_retriever_schema lorsque vous définissez votre agent dans le code. Ceci mappe les colonnes de sortie de votre récupérateur aux champs attendus de MLflow (primary_key, text_column, doc_uri).

Python
import mlflow
# Define the retriever's schema by providing your column names
# For example, the following call specifies the schema of a retriever that returns a list of objects like
# [
#     {
#         'document_id': '9a8292da3a9d4005a988bf0bfdd0024c',
#         'chunk_text': 'MLflow is the largest open source AI engineering platform for agents, LLMs, and ML models...',
#         'doc_uri': 'https://mlflow.org/docs/latest/index.html',
#         'title': 'MLflow: The Largest Open Source AI Engineering Platform'
#     },
#     {
#         'document_id': '7537fe93c97f4fdb9867412e9c1f9e5b',
#         'chunk_text': 'A great way to get started with MLflow is to use the autologging feature. Autologging automatically logs your model...',
#         'doc_uri': 'https://mlflow.org/docs/latest/getting-started/',
#         'title': 'Getting Started with MLflow'
#     },
# ...
# ]
mlflow.models.set_retriever_schema(
    # Specify the name of your retriever span
    name="mlflow_docs_vector_search",
    # Specify the output column name to treat as the primary key (ID) of each retrieved document
    primary_key="document_id",
    # Specify the output column name to treat as the text content (page content) of each retrieved document
    text_column="chunk_text",
    # Specify the output column name to treat as the document URI of each retrieved document
    doc_uri="doc_uri",
    # Specify any other columns returned by the retriever
    other_columns=["title"],
)
remarque

La colonne doc_uri est particulièrement importante lors de l'évaluation des performances de l'extracteur. doc_uri est l'identifiant principal des documents renvoyés par l'extracteur, vous permettant de les comparer à des jeux d'évaluation de vérité terrain. Consultez les jeux d'évaluation (MLflow 2).

Considérations de déploiement​

Préparer pour Databricks Model Serving​

Databricks déploie des ResponsesAgents dans un environnement distribué sur Databricks Model Serving. Cela signifie que pendant une conversation multi-tours, la même réplique de service pourrait ne pas gérer toutes les requêtes. Portez attention aux implications suivantes pour la gestion de l'état de l'agent :

  • **Évitez la mise en cache locale** : Lors du déploiement ResponsesAgent d'un, ne supposez pas que la même réplique gère toutes les requêtes dans une conversation à plusieurs tours. Reconstruisez l'état interne à l'aide d'un schéma ResponsesAgentRequest de dictionnaire pour chaque tour.

  • **État thread-safe** : Concevez l'état de l'agent pour qu'il soit thread-safe, évitant les conflits dans les environnements multithread.

  • **Initialisez l'état dans la predict fonction ** : Initialisez l'état chaque fois que la predict fonction est appelée, pas pendant ResponsesAgent l'initialisation de. Stocker l'état au niveau ResponsesAgent pourrait laisser fuir des informations entre les conversations et causer des conflits, car une seule réplique ResponsesAgent pourrait gérer les requêtes de plusieurs conversations.

Paramétrer le code pour le déploiement dans plusieurs environnements​

Paramétrez le code de l'agent pour réutiliser le même code d'agent dans différents environnements.

Les paramètres sont des paires clé-valeur que vous définissez dans un dictionnaire Python ou un fichier .yaml.

Pour configurer le code, créez un ModelConfig à l'aide d'un dictionnaire Python ou d'un fichier .yaml. ModelConfig est un ensemble de paramètres clé-valeur qui permet une gestion flexible de la configuration. Par exemple, vous pouvez utiliser un dictionnaire pendant le développement, puis le convertir en un fichier .yaml pour le déploiement en production et la CI/CD.

Un exemple ModelConfig est présenté ci-dessous :

YAML
llm_parameters:
  max_tokens: 500
  temperature: 0.01
model_serving_endpoint: databricks-meta-llama-3-3-70b-instruct
vector_search_index: ml.docs.databricks_docs_index
prompt_template: 'You are a hello world bot. Respond with a reply to the user''s
  question that indicates your prompt template came from a YAML file. Your response
  must use the word "YAML" somewhere. User''s question: {question}'
prompt_template_input_vars:
  - question

Dans votre code d'agent, vous pouvez référencer une configuration default (de développement) à partir du fichier ou du dictionnaire .yaml :

Python
import mlflow
# Example for loading from a .yml file
config_file = "configs/hello_world_config.yml"
model_config = mlflow.models.ModelConfig(development_config=config_file)

# Example of using a dictionary
config_dict = {
    "prompt_template": "You are a hello world bot. Respond with a reply to the user's question that is fun and interesting to the user. User's question: {question}",
    "prompt_template_input_vars": ["question"],
    "model_serving_endpoint": "databricks-meta-llama-3-3-70b-instruct",
    "llm_parameters": {"temperature": 0.01, "max_tokens": 500},
}

model_config = mlflow.models.ModelConfig(development_config=config_dict)

# Use model_config.get() to retrieve a parameter value
# You can also use model_config.to_dict() to convert the loaded config object
# into a dictionary
value = model_config.get('sample_param')

Ensuite, lorsque vous enregistrez votre agent, spécifiez le model_config paramètre log_model vers"> pour spécifier un ensemble personnalisé de paramètres à utiliser lors du chargement de l'agent enregistré. Consultez la documentation MLflow - ModelConfig.

Utilisez du code synchrone ou des modèles de rappel​

Pour garantir la stabilité et la compatibilité, utilisez du code synchrone ou des modèles basés sur des rappels dans l'implémentation de votre agent.

Databricks gère automatiquement la communication asynchrone afin de fournir une concurrence et des performances optimales lors du déploiement d'un agent. L'introduction de boucles d'événements personnalisées ou de frameworks asynchrones pourrait entraîner des erreurs comme RuntimeError: This event loop is already running and caused unpredictable behavior.

Databricks recommande d'éviter la programmation asynchrone, telle que l'utilisation d'asyncio ou la création de boucles d'événements personnalisées, lors du développement d'agents.

Ressources supplémentaires​

Sur cette page