Schemat agenta wejścia i wyjścia typu legacy

Agenci sztucznej inteligencji muszą przestrzegać określonych wymagań dotyczących schematu danych wejściowych i wyjściowych, aby były zgodne z innymi funkcjami w usłudze Databricks. Na tej stronie opisano, jak korzystać ze starszych agentów do tworzenia podpisów i interfejsów: ChatAgent interfejs, ChatModel interfejs, SplitChatMessageRequest schemat wejściowy i schemat wyjściowy StringResponse.

Opracowanie przestarzałego agenta ChatAgent

Interfejs MLflow ChatAgent jest podobny do tego, ale nie jest ściśle zgodny ze schematem OpenAI ChatCompletion .

Aby dowiedzieć się, jak utworzyć ChatAgent, zobacz przykłady w następnej sekcji oraz w dokumentacji MLflow - Czym jest interfejs ChatAgent.

Aby utworzyć i wdrożyć agentów przy użyciu programu ChatAgent, zainstaluj następujące elementy:

• databricks-agents 0.16.0 lub nowszy
• mlflow 2.20.2 lub nowszy
• Środowisko Python w wersji 3.10 lub nowszej.
  • Aby spełnić to wymaganie, możesz użyć bezserwerowych obliczeń lub środowiska Databricks Runtime 13.3 LTS lub nowszego.

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

Co zrobić, jeśli mam już agenta?

Jeśli masz już agenta skompilowany za pomocą oprogramowania LangChain, LangGraph lub podobnej struktury, nie musisz ponownie pisać agenta, aby używać go w usłudze Databricks. Zamiast tego po prostu owiń swojego obecnego agenta interfejsem MLflow ChatAgent.

1. Napisz klasę opakowującą w Pythonie, która dziedziczy z mlflow.pyfunc.ChatAgent klasy.

   Wewnątrz klasy opakowania zachowaj istniejącego agenta jako atrybut self.agent = your_existing_agent.

2. Klasa ChatAgent wymaga zaimplementowania predict metody do obsługi żądań typu non-streaming.

   predict musi zaakceptować:

   • messages: list[ChatAgentMessage], która jest listą ChatAgentMessage, z których każdy zawiera rolę (taką jak "użytkownik" lub "asystent"), polecenie oraz identyfikator.

   • (Opcjonalnie) context: Optional[ChatContext] i custom_inputs: Optional[dict] w przypadku dodatkowych danych.

   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 musi zwrócić wartość ChatAgentResponse.

   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. Konwertowanie między formatami

   W predict, przekonwertuj nadchodzące komunikaty z list[ChatAgentMessage] na format wejściowy oczekiwany przez agenta.

   Po wygenerowaniu odpowiedzi przez agenta, przekonwertuj jego dane wyjściowe na jeden lub więcej obiektów ChatAgentMessage i opakuj je w ChatAgentResponse.

Poniżej przedstawiono uproszczony szablon do konwersji Twojego agenta:

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

Aby uzyskać pełne przykłady, zobacz notesy w poniższej sekcji.

przykłady ChatAgent

W poniższych notatnikach pokazano, jak tworzyć aplikacje strumieniowe i niestrumieniowe ChatAgents przy użyciu popularnych bibliotek OpenAI, LangGraph i AutoGen.

LangGraph

Jeśli opakowujesz agenta LangChain, możesz użyć mlflow.langchain.output_parsers.ChatAgentOutputParser do automatycznego konwertowania danych wyjściowych LangChain na schemat MLflow ChatAgentMessage i ChatAgentResponse.

Agent wywoływania narzędzi LangGraph

Pobierz laptopa

OpenAI

Narzędzie do wywoływania modułów OpenAI

Pobierz laptopa

Agent do wywoływania narzędzi OpenAI Responses API

Pobierz laptopa

Bot OpenAI tylko do rozmowy

Pobierz laptopa

AutoGen

Agent narzędzi wywoływanych przez AutoGen

Pobierz laptopa

DSPy

Agent DSPy przeznaczony wyłącznie do czatu

Pobierz laptopa

Aby dowiedzieć się, jak rozszerzyć możliwości tych agentów, dodając narzędzia, zobacz narzędzia agenta sztucznej inteligencji.

Przesyłanie strumieniowe odpowiedzi usługi ChatAgent

Agenci przesyłania strumieniowego dostarczają odpowiedzi w ciągłym strumieniu mniejszych, przyrostowych fragmentów. Przesyłanie strumieniowe zmniejsza postrzegane opóźnienie i poprawia doświadczenie użytkownika dla agentów konwersacyjnych.

Aby utworzyć przesyłanie strumieniowe ChatAgent, zdefiniuj predict_stream metodę zwracającą generator, który generuje ChatAgentChunk obiekty — każdy z nich ChatAgentChunk zawiera część odpowiedzi. Przeczytaj więcej na temat idealnego ChatAgent zachowania przesyłania strumieniowego w dokumentacji platformy MLflow.

Poniższy kod przedstawia przykładową funkcję predict_stream. Pełne przykłady agentów przesyłania strumieniowego można znaleźć w przykładach ChatAgent.

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"]
      )

Utworzyć agenta starszej wersji ChatModel

ChatModel to dotychczasowy interfejs tworzenia agentów w narzędziu MLflow, który rozszerza schemat ChatCompletion firmy OpenAI, co pozwala zachować zgodność z platformami obsługującymi standard ChatCompletion, umożliwiając jednocześnie dodawanie niestandardowych funkcji. Aby uzyskać dodatkowe informacje, zobacz MLflow: Getting Started with ChatModel (Wprowadzenie do modelu ChatModel ).

Tworzenie agenta jako podklasy mlflow.pyfunc.ChatModel zapewnia następujące korzyści:

• Włącza dane wyjściowe agenta przesyłania strumieniowego podczas wywoływania obsługiwanego agenta (pomijając {stream: true} treść żądania).

• Automatycznie włącza tabele inferowania Bramy Sztucznej Inteligencji, gdy Twój agent jest uruchamiany, zapewniając dostęp do rozszerzonych metadanych dziennika żądań, takich jak nazwa wnioskodawcy.

  Ostrzeżenie

  Dzienniki żądań i dzienniki oceny są przestarzałe i zostaną usunięte w przyszłej wersji. Aby uzyskać wskazówki dotyczące migracji, zobacz wycofywanie dzienników żądań i dzienników oceny .

• Umożliwia pisanie kodu agenta zgodnego ze schematem ChatCompletion przy użyciu wpisanych klas języka Python.

• Narzędzie MLflow automatycznie wywnioskuje podpis zgodny z uzupełnianiem czatu podczas rejestrowania agenta, nawet bez elementu input_example. Upraszcza to proces rejestrowania i wdrażania agenta. Zobacz Podpis modelu podczas procesu logowania.

Poniższy kod najlepiej działa w notatniku usługi Databricks. Notesy zapewniają wygodne środowisko do tworzenia, testowania i iterowania agenta.

Klasa MyAgent rozszerza mlflow.pyfunc.ChatModel, implementuje wymaganą metodę predict. Zapewnia to zgodność z frameworkiem Mosaic AI Agent Framework.

Klasa zawiera również opcjonalne metody _create_chat_completion_chunk i predict_stream do obsługi wyjść strumieniowych.

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

Podczas definiowania klasy agenta MyAgent w jednym notatniku zalecamy utworzenie oddzielnego notatnika kierowcy. Notatnik sterownika rejestruje agenta w rejestrze modeli i wdraża agenta za pomocą Model Serving.

Ten podział jest zgodny z przepływem pracy zalecanym przez Databricks do rejestrowania modeli przy użyciu modeli MLflow zgodnie z metodologią Modeli z Kodem.

SplitChatMessageRequest schemat danych wejściowych (przestarzały)

SplitChatMessagesRequest umożliwia oddzielne przekazywanie bieżącego zapytania i historii jako danych wejściowych agenta.

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

Schemat danych wyjściowych StringResponse (przestarzały)

StringResponse Umożliwia zwrócenie odpowiedzi agenta jako obiektu z pojedynczym polem ciągu content :

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