Pular para o conteúdo principal

Esquema de agente de entrada e saída legado

nota

A Databricks recomenda a migração para o esquema ResponsesAgent para criar agentes. Consulte os agentes " AI " no código.

AI Os agentes devem aderir a requisitos específicos de esquema de entrada e saída para serem compatíveis com outros recursos em Databricks. Esta página explica como usar as assinaturas e interfaces de criação do agente legado: interface ChatAgent, interface ChatModel, esquema de entrada SplitChatMessageRequest e esquema de saída StringResponse.

Crie um agente antigo do ChatAgent​

A interface do MLflow ChatAgent é semelhante, mas não estritamente compatível, com o esquema OpenAI ChatCompletion.

O ChatAgent integra facilmente os agentes existentes para compatibilidade com o Databricks.

Para aprender como criar um ChatAgent, consulte os exemplos na seção a seguir e a documentação do MLflow - O que é a interface ChatAgent.

Para criar e implantar agentes utilizando ChatAgent, instale o seguinte:

  • databricks-agents0.16.0 ou superior
  • mlflow 2.20.2 ou superior
  • Python 3,10 ou mais.
    • Para atender a esse requisito, é possível utilizar serverless compute ou Databricks Runtime 13.3 LTS ou acima.
Python
%pip install -U -qqqq databricks-agents>=0.16.0 mlflow>=2.20.2

E se eu já tiver um agente?​

Caso já possua um agente desenvolvido com LangChain, LangGraph ou uma estrutura semelhante, não é necessário reescrevê-lo para utilizá-lo no Databricks. Em vez disso, basta envolver o seu agente existente com a interface MLflow ChatAgent:

  1. Escreva uma classe wrapper Python que herde de mlflow.pyfunc.ChatAgent.

    Dentro da classe wrapper, mantenha seu agente existente como um atributo self.agent = your_existing_agent.

  2. A classe ChatAgent requer que você implemente um método para lidar com solicitações de não transmissão. predict método para lidar com solicitações não transmissíveis.

    predict deve aceitar:

    • messages: list[ChatAgentMessage], que é uma lista de ChatAgentMessage, cada um com uma função (como " user " ou " assistant "), o prompt e um ID.

    • (Opcional) context: Optional[ChatContext] e custom_inputs: Optional[dict] para dados extras.

    Python
    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 deve retornar um ChatAgentResponse.

    Python
    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. Converter entre formatos

    Em predict, converta as mensagens recebidas de list[ChatAgentMessage] no formato de entrada que seu agente espera.

    Depois que seu agente gerar uma resposta, converta sua saída em um ou mais objetos ChatAgentMessage e envolva-os em ChatAgentResponse.

dica

Converter automaticamente a saída do LangChain

Se estiver envolvendo um agente LangChain, é possível utilizar mlflow.langchain.output_parsers.ChatAgentOutputParser para converter automaticamente as saídas do LangChain para o esquema MLflow ChatAgentMessage e ChatAgentResponse.

A seguir, apresentamos um padrão simplificado para converter seu agente:

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

Para obter exemplos completos, consulte o Caderno na seção a seguir.

Exemplos deChatAgent​

O Notebook a seguir demonstra como criar transmissão e não transmissão ChatAgents utilizando as bibliotecas populares OpenAI, LangGraph e AutoGen.

Se estiver envolvendo um agente LangChain, é possível utilizar mlflow.langchain.output_parsers.ChatAgentOutputParser para converter automaticamente as saídas do LangChain para o esquema MLflow ChatAgentMessage e ChatAgentResponse.

Agente de chamada de ferramentas LangGraph

Open notebook in new tab

Para saber como expandir os recursos desses agentes adicionando ferramentas, consulte Ferramentas do agente doAI.

transmissão Respostas do ChatAgent​

Os agentes de transmissão fornecem respostas em uma transmissão contínua de partes menores e incrementais. A transmissão reduz a latência percebida e melhora a experiência do usuário para agentes conversacionais.

Para criar uma transmissão ChatAgent, defina um método predict_stream que retorne um gerador que produza objetos ChatAgentChunk - cada ChatAgentChunk contém uma parte da resposta. Leia mais sobre o comportamento ideal de transmissão do ideal ChatAgent nos documentosMLflow.

O código a seguir apresenta um exemplo da função predict_stream. Para exemplos completos de agentes de transmissão, consulte os exemplos do ChatAgent:

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

Crie um agente legado do ChatModel​

important

Databricks recomenda a interface ChatAgent para criar agentes ou gerar aplicativos AI. Para migrar do ChatModel para o ChatAgent, consulte a documentação do MLflow - Migrar do ChatModel para o ChatAgent.

ChatModel é uma interface de criação de agente legado no MLflow que estende o esquema ChatCompletion da OpenAI, permitindo que o senhor mantenha a compatibilidade com plataformas que suportam o padrão ChatCompletion e, ao mesmo tempo, adicione funcionalidades personalizadas. MLflowConsulte: Começando com o ChatModel para obter mais detalhes.

Criar seu agente como uma subclasse de mlflow.pyfunc.ChatModel oferece os seguintes benefícios:

  • Ativa a saída do agente de transmissão ao invocar um agente atendido (ignorando {stream: true} no corpo da solicitação).
  • Ativa automaticamente as tabelas de inferência do AI Gateway quando o seu agente é atendido, fornecendo acesso a metadados aprimorados do log de solicitações, como o nome do solicitante.
  • Permite que o senhor escreva código de agente compatível com o esquema ChatCompletion usando classes Python tipadas.
  • O MLflow infere automaticamente uma assinatura compatível com a conclusão do chat ao registrar o agente, mesmo sem um input_example. Isso simplifica o processo de registro e implantação do agente. Consulte Inferir assinatura do modelo durante o registro.

O código a seguir é melhor executado em um notebook com Databricks. O Notebook oferece um ambiente conveniente para desenvolver, testar e iterar seu agente.

A classe MyAgent estende mlflow.pyfunc.ChatModel, implementando o método predict necessário. Isso garante a compatibilidade com o Mosaic AI Agent Framework.

A classe também inclui os métodos opcionais _create_chat_completion_chunk e predict_stream para lidar com as saídas de transmissão.

Python
# Install the latest version of mlflow
%pip install -U mlflow
dbutils.library.restartPython()
Python
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)

Enquanto define a classe do agente MyAgent em um Notebook, recomendamos criar um Notebook de driver separado. O driver Notebook logs o agente em Model Registry e implantado o agente utilizando o modelo servindo.

Essa separação segue o fluxo de trabalho recomendado pela Databricks para modelos de registro usando a metodologia Models from Code do MLflow.

Esquema de entrada SplitChatMessageRequest (obsoleto)​

SplitChatMessagesRequest permite que o senhor passe a consulta atual e o histórico separadamente como entrada do agente.

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

Esquema de saída StringResponse (obsoleto)​

StringResponse permite que o senhor retorne a resposta do agente como um objeto com um único campo de strings content:

{"content": "This is an example string response"}
Última atualização em