Älteres Eingabe- und Ausgabe-Agent-Schema

KI-Agents müssen bestimmte Eingabe- und Ausgabeschemaanforderungen einhalten, um mit anderen Features in Databricks kompatibel zu sein. Auf dieser Seite wird erläutert, wie Sie die Legacy-Agent-Erstellungssignaturen und -schnittstellen verwenden: ChatAgent Schnittstelle, ChatModel Schnittstelle, SplitChatMessageRequest Eingabeschema und Ausgabeschema StringResponse .

Erstellen eines älteren ChatAgent-Agents

Die MLflow-Schnittstelle ChatAgent ist mit dem OpenAI-Schema ChatCompletion vergleichbar, aber nicht streng kompatibel.

Informationen zum Erstellen einer ChatAgent finden Sie in den Beispielen im folgenden Abschnitt und unter MLflow-Dokumentation – Was ist die ChatAgent-Schnittstelle.

Um Agents mithilfe ChatAgentvon Agents zu erstellen und bereitzustellen, installieren Sie Folgendes":

• databricks-agents 0.16.0 oder höher
• mlflow 2.20.2 oder höher
• Python 3.10 oder höher.
  • Um diese Anforderung zu erfüllen, können Sie serverloses Computing oder die Databricks Runtime 13.3 LTS oder höher verwenden.

%pip install -U -qqqq databricks-agents>=0.16.0 mlflow>=2.20.2

Was geschieht, wenn ich bereits über einen Agenten verfüge?

Wenn Sie bereits über einen Agent verfügen, der mit LangChain, LangGraph oder einem ähnlichen Framework erstellt wurde, müssen Sie Ihren Agent nicht neu schreiben, um ihn auf Databricks zu verwenden. Schließen Sie stattdessen ihren vorhandenen Agent einfach mit der MLflow-Schnittstelle ChatAgent um:

1. Schreiben Sie eine Python-Wrapperklasse, die von mlflow.pyfunc.ChatAgent erbt.

   Behalten Sie ihren vorhandenen Agent innerhalb der Wrapperklasse als Attribut self.agent = your_existing_agentbei.

2. Die ChatAgent Klasse erfordert, dass Sie eine predict Methode zum Verarbeiten von Nicht-Streaming-Anforderungen implementieren.

   predict muss folgendes akzeptieren:

   • messages: list[ChatAgentMessage], bei dem es sich um eine Liste handelt, die aus ChatAgentMessage besteht, wobei jeder Eintrag eine Rolle (wie "Benutzer" oder "Assistent"), ein Prompt und eine ID enthält.

   • (Optional) context: Optional[ChatContext] und custom_inputs: Optional[dict] für zusätzliche Daten.

   import uuid
   
   # input example
   [
     ChatAgentMessage(
       id=str(uuid.uuid4()),  # Generate a unique ID for each message
       role="user",
       content="What's the weather in Paris?"
     )
   ]
   

   predict muss ein ChatAgentResponse zurückgeben.

   import uuid
   
   # output example
   ChatAgentResponse(
     messages=[
       ChatAgentMessage(
         id=str(uuid.uuid4()),  # Generate a unique ID for each message
         role="assistant",
         content="It's sunny in Paris."
       )
     ]
   )
   

3. Konvertieren zwischen Formaten

   In predict konvertieren Sie die eingehenden Nachrichten von list[ChatAgentMessage] in das Eingabeformat, das Ihr Agent erwartet.

   Nachdem Ihr Agent eine Antwort generiert hat, konvertieren Sie die Ausgabe in ein oder mehrere ChatAgentMessage Objekte, und schließen Sie sie in ein ChatAgentResponse.

Es folgt eine vereinfachte Vorlage zum Konvertieren Ihres Agenten:

from mlflow.pyfunc import ChatAgent
from mlflow.types.agent import ChatAgentMessage, ChatAgentResponse, ChatAgentChunk
import uuid


class MyWrappedAgent(ChatAgent):
  def __init__(self, agent):
    self.agent = agent

  def predict(self, messages, context=None, custom_inputs=None):
    # Convert messages to your agent's format
    agent_input = ... # build from messages
    agent_output = self.agent.invoke(agent_input)
    # Convert output to ChatAgentMessage
    return ChatAgentResponse(
      messages=[ChatAgentMessage(role="assistant", content=agent_output, id=str(uuid.uuid4()),)]
    )

  def predict_stream(self, messages, context=None, custom_inputs=None):
    # If your agent supports streaming
    for chunk in self.agent.stream(...):
      yield ChatAgentChunk(delta=ChatAgentMessage(role="assistant", content=chunk, id=str(uuid.uuid4())))

Vollständige Beispiele finden Sie in den Notizbüchern im folgenden Abschnitt.

Beispiele für ChatAgent

Die folgenden Notizbücher zeigen, wie Sie Streaming und Nicht-Streaming ChatAgents mit den beliebten Bibliotheken OpenAI, LangGraph und AutoGen verfassen können.

LangGraph

Wenn Sie einen LangChain-Agenten umschließen, können Sie mlflow.langchain.output_parsers.ChatAgentOutputParser verwenden, um LangChain-Ausgaben automatisch in das MLflow-ChatAgentMessage- und ChatAgentResponse-Schema zu konvertieren.

Toolaufruf-Agent für LangGraph

Notebook abrufen

OpenAI

Toolaufruf-Agent für OpenAI

Notebook abrufen

Toolaufruf-Agent der OpenAI-Antwort-API

Notebook abrufen

Chat-Agent für OpenAI

Notebook abrufen

AutoGen

AutoGen-Toolaufruf-Agent

Notebook abrufen

DSPy

Nur-DSPy-Chat-Agent

Notebook abrufen

Informationen zum Erweitern der Funktionen dieser Agents durch Hinzufügen von Tools finden Sie unter KI-Agent-Tools.

Streamen von ChatAgent-Antworten

Streaming-Agents liefern Antworten in einem kontinuierlichen Stream kleinerer, inkrementeller Blöcke. Streaming reduziert die wahrgenommene Latenzzeit und verbessert die Benutzererfahrung bei konversationalen Agenten.

Um ein Streaming ChatAgent zu erstellen, definieren Sie eine predict_stream-Methode, die einen Generator liefert, der ChatAgentChunk-Objekte erzeugt – jedes ChatAgentChunk enthält einen Teil der Antwort. Weitere Informationen zum idealen ChatAgent Streamingverhalten in den MLflow-Dokumenten.

Der folgende Code zeigt ein Beispiel für eine predict_stream-Funktion, die vollständige Beispiele für Streaming-Agents finden Sie unter ChatAgent-Beispiele:

def predict_stream(
  self,
  messages: list[ChatAgentMessage],
  context: Optional[ChatContext] = None,
  custom_inputs: Optional[dict[str, Any]] = None,
) -> Generator[ChatAgentChunk, None, None]:
  # Convert messages to a format suitable for your agent
  request = {"messages": self._convert_messages_to_dict(messages)}

  # Stream the response from your agent
  for event in self.agent.stream(request, stream_mode="updates"):
    for node_data in event.values():
      # Yield each chunk of the response
      yield from (
        ChatAgentChunk(**{"delta": msg}) for msg in node_data["messages"]
      )

Erstellen eines älteren ChatModel-Agents

ChatModel ist eine ältere Agent-Erstellungsschnittstelle in MLflow, die das ChatCompletion-Schema von OpenAI erweitert, sodass Sie die Kompatibilität mit Plattformen beibehalten können, die den ChatCompletion-Standard unterstützen, während sie benutzerdefinierte Funktionen hinzufügen. Weitere Details finden Sie unter MLflow: Erste Schritte mit ChatModel .

Die Erstellung Ihres Agents als Unterklasse von mlflow.pyfunc.ChatModel bietet die folgenden Vorteile:

• Aktiviert die Ausgabe des Streaming-Agents beim Aufrufen eines bereitgestellten Agents (Umgehung {stream: true} im Anforderungstext).

• Aktiviert automatisch AI-Gateway-Ableitungstabellen, wenn Ihr Agent bereitgestellt wird, und ermöglicht den Zugriff auf erweiterte Anforderungsprotokollmetadaten, z. B. den Namen des Antragstellers.

  Warnung

  Anforderungsprotokolle und Bewertungsprotokolle sind veraltet und werden in einer zukünftigen Version entfernt. Weitere Informationen zur Migration finden Sie unter Außerbetriebnahme von Anforderungsprotokollen und Bewertungsprotokollen.

• Ermöglicht Ihnen das Schreiben von Agentcode, der mit dem ChatCompletion-Schema mit typierten Python-Klassen kompatibel ist.

• MLflow leitet automatisch eine chat-kompatible Signatur ab, wenn der Agent protokolliert wird, auch ohne input_example. Dies vereinfacht das Registrieren und Bereitstellen des Agents. Weitere Informationen finden Sie unter Ableiten der Modellsignatur während der Protokollierung.

Der folgende Code wird am besten in einem Databricks-Notizbuch ausgeführt. Notebooks bieten eine praktische Umgebung zum Entwickeln, Testen und Iterieren Ihres Agents.

Die MyAgent Klasse erweitert mlflow.pyfunc.ChatModel, wobei die erforderliche predict-Methode implementiert wird. Dadurch wird die Kompatibilität mit Dem Mosaik AI Agent Framework sichergestellt.

Die Klasse enthält auch die optionalen Methoden _create_chat_completion_chunk und predict_stream zum Behandeln von Streamingausgaben.

# Install the latest version of mlflow
%pip install -U mlflow
dbutils.library.restartPython()

import re
from typing import Optional, Dict, List, Generator
from mlflow.pyfunc import ChatModel
from mlflow.types.llm import (
  # Non-streaming helper classes
  ChatCompletionRequest,
  ChatCompletionResponse,
  ChatCompletionChunk,
  ChatMessage,
  ChatChoice,
  ChatParams,
  # Helper classes for streaming agent output
  ChatChoiceDelta,
  ChatChunkChoice,
)

class MyAgent(ChatModel):
  """
  Defines a custom agent that processes ChatCompletionRequests
  and returns ChatCompletionResponses.
  """
  def predict(self, context, messages: list[ChatMessage], params: ChatParams) -> ChatCompletionResponse:
    last_user_question_text = messages[-1].content
    response_message = ChatMessage(
      role="assistant",
      content=(
        f"I will always echo back your last question. Your last question was: {last_user_question_text}. "
      )
    )
    return ChatCompletionResponse(
      choices=[ChatChoice(message=response_message)]
    )

  def _create_chat_completion_chunk(self, content) -> ChatCompletionChunk:
    """Helper for constructing a ChatCompletionChunk instance for wrapping streaming agent output"""
    return ChatCompletionChunk(
      choices=[ChatChunkChoice(
        delta=ChatChoiceDelta(
          role="assistant",
          content=content
        )
      )]
    )

  def predict_stream(
    self, context, messages: List[ChatMessage], params: ChatParams
  ) -> Generator[ChatCompletionChunk, None, None]:
    last_user_question_text = messages[-1].content
    yield self._create_chat_completion_chunk(f"Echoing back your last question, word by word.")
    for word in re.findall(r"\S+\s*", last_user_question_text):
      yield self._create_chat_completion_chunk(word)

agent = MyAgent()
model_input = ChatCompletionRequest(
  messages=[ChatMessage(role="user", content="What is Databricks?")]
)
response = agent.predict(context=None, messages=model_input.messages, params=None)
print(response)

Während Sie die Agentklasse MyAgent in einem Notizbuch definieren, empfehlen wir das Erstellen eines separaten Treibernotizbuchs. Das Treibernotizbuch protokolliert den Agenten im Modellregister und stellt ihn mithilfe von Model Serving bereit.

Diese Trennung folgt dem von Databricks empfohlenen Workflow für die Protokollierung von Modellen mithilfe der Methodik 'Models from Code' von MLflow.

SplitChatMessageRequest-Eingabeschema (veraltet)

mit SplitChatMessagesRequest können Sie die aktuelle Abfrage und den aktuellen Verlauf separat als Agenteingabe übergeben.

  question = {
    "query": "What is MLflow",
    "history": [
      {
        "role": "user",
        "content": "What is Retrieval-augmented Generation?"
      },
      {
        "role": "assistant",
        "content": "RAG is"
      }
    ]
  }

StringResponse-Ausgabeschema (veraltet)

StringResponse ermöglicht es Ihnen, die Antwort des Agents als Objekt mit einem einzelnen Zeichenfolgenfeld content zurückzugeben:

{"content": "This is an example string response"}