Legacy giriş ve çıkış ajan şeması (Model Servis Etme)

Important

Yeni kullanım örnekleri için Databricks aracı kodu, sunucu yapılandırması ve dağıtım iş akışı üzerinde tam denetim için Databricks Uygulamalarına aracı dağıtmanızı önerir. Bkz. Bir yapay zeka aracısı yazma ve Databricks Uygulamalarında dağıtma. Mevcut bir ajanı taşımak için Model Sunumu'ndan Databricks Uygulamalarına ajan göç ettirme bölümüne bakın.

Note

Databricks, aracıları yazmak için ResponsesAgent şemaya geçiş yapmanızı önerir. Bkz. Bir yapay zeka aracısı yazma ve Databricks Uygulamalarında dağıtma.

Yapay zeka aracılarının Databricks'te diğer özelliklerle uyumlu olması için belirli giriş ve çıkış şeması gereksinimlerine uyması gerekir. Bu sayfada eski aracı yazma imzalarının ve arabirimlerinin nasıl kullanılacağı açıklanmaktadır: ChatAgent arabirim, ChatModel arabirim, SplitChatMessageRequest giriş şeması ve StringResponse çıkış şeması.

Eski bir ChatAgent aracısı yazma

MLflow ChatAgent arabirimi, OpenAI ChatCompletion şemasına benzer ancak bu şemayla tam olarak uyumlu değildir.

ChatAgent, Databricks uyumluluğu için mevcut aracıları kolayca sarmalar.

ChatAgentoluşturmayı öğrenmek için, aşağıdaki bölümdeki örneklere ve MLflow belgelerine bakın: ChatAgent arabiriminedir?

ChatAgent kullanarak aracılar yazmak ve dağıtmak için aşağıdakileri yükleyin:

  • databricks-agents 0.16.0 veya üzeri
  • mlflow 2.20.2 veya üzeri
  • Python 3.10 veya üzeri.
    • Bu gereksinimi karşılamak için sunucusuz işlem veya Databricks Runtime 13.3 LTS veya üzerini kullanabilirsiniz.
%pip install -U -qqqq databricks-agents==0.16.0 mlflow==2.20.2

Ya zaten bir temsilcim varsa?

Zaten LangChain, LangGraph veya benzer bir çerçeveyle oluşturulmuş bir aracınız varsa, aracınızı Databricks'te kullanmak için yeniden yazmanız gerekmez. Bunun yerine, mevcut aracınızı MLflow ChatAgent arabirimiyle sarmalamanız gerekir:

  1. mlflow.pyfunc.ChatAgent'dan devralan bir Python sarmalayıcı sınıfı yazın.

    Sarmalayıcı sınıfının içinde, mevcut aracınızı özniteliği self.agent = your_existing_agentolarak tutun.

  2. Bir ChatAgent sınıfı, akış içermeyen istekleri işlemek için bir predict yöntem uygulamanızı gerektirir.

    predict kabul etmelidir:

    • messages: list[ChatAgentMessage], rolü olan her birinin ChatAgentMessage listesidir ("kullanıcı" veya "yardımcı"), istem ve kimlik.

    • (İsteğe bağlı) context: Optional[ChatContext] ve custom_inputs: Optional[dict] ek veriler için.

    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 bir ChatAgentResponse döndürmelidir.

    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. Biçimler arasında dönüştürme

    predict içinde, gelen iletileri list[ChatAgentMessage] temsilcinizin beklediği giriş biçimine dönüştürün.

    Ajanınız bir yanıt oluşturduktan sonra çıktısını bir veya daha fazla ChatAgentMessage nesneye dönüştürün ve bir ChatAgentResponse içinde sarmalayın.

Tavsiye

LangChain çıkışını otomatik olarak dönüştürme

Bir LangChain aracısını sarmalıyorsanız, LangChain çıkışlarını otomatik olarak MLflow mlflow.langchain.output_parsers.ChatAgentOutputParser ve ChatAgentMessage şeması formatına dönüştürmek için ChatAgentResponse kullanabilirsiniz.

Aşağıda temsilcinizi dönüştürmek için basitleştirilmiş bir şablon bulunmaktadır:

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

Tam örnekler için aşağıdaki bölümdeki not defterlerine bakın.

ChatAgent örnekleri

Aşağıdaki not defterleri, popüler OpenAI, LangGraph ve AutoGen kitaplıklarını kullanarak akış tabanlı ve akış tabanı olmayan yazılım oluşturmayı gösterir.

LangGraph

Bir LangChain aracısını sarmalıyorsanız, LangChain çıkışlarını otomatik olarak MLflow mlflow.langchain.output_parsers.ChatAgentOutputParser ve ChatAgentMessage şeması formatına dönüştürmek için ChatAgentResponse kullanabilirsiniz.

LangGraph araç çağırma aracısı

Dizüstü bilgisayar al

OpenAI

OpenAI araç çağırma aracısı

Dizüstü bilgisayar al

OpenAI Yanıtlar API'sinde araç çağırma aracı

Dizüstü bilgisayar al

OpenAI yalnızca sohbet aracısı

Dizüstü bilgisayar al

AutoGen

AutoGen araç çağırma aracısı

Dizüstü bilgisayar al

DSPy

DSPy yalnızca sohbet aracısı

Dizüstü bilgisayar al

Araçlar ekleyerek bu aracıların özelliklerini genişletmeyi öğrenmek için bkz. Aracıları araçlara bağlama.

ChatAgent yanıtlarını akışla aktarma

Akış ajanları, yanıtları sürekli bir şekilde küçük, artımlı parçalar halinde sunar. Akış algılanan gecikme süresini azaltır ve konuşma aracıları için kullanıcı deneyimini geliştirir.

Akış ChatAgent yazmak için, predict_stream objeleri oluşturan ve bunları sunan bir jeneratör döndüren bir ChatAgentChunk yöntemi tanımlayın; her bir ChatAgentChunk, yanıtın bir bölümünü içerir. ChatAgent ideal akış davranışı hakkında daha fazla bilgi edinin.

Aşağıdaki kodda örnek bir predict_stream işlevi gösterilmektedir. Akış aracılarının tam örnekleri için bkz. ChatAgent örnekleri:

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

Eski bir ChatModel aracısı yazma

Important

Databricks, aracı veya üretken yapay zeka uygulamaları oluşturmak için ChatAgent arabirimini önerir. ChatModel'den ChatAgent'a geçiş yapmak için bkz. MLflow belgeleri - ChatModel'den ChatAgent'a geçiş.

ChatModel , MLflow'da OpenAI'nin ChatCompletion şemasını genişleten ve özel işlevsellik eklerken ChatCompletion standardını destekleyen platformlarla uyumluluğu korumanızı sağlayan eski bir aracı yazma arabirimidir. Ek ayrıntılar için bkz . MLflow: ChatModel'i Kullanmaya Başlama .

Aracınızı mlflow.pyfunc.ChatModel alt sınıfı olarak yazmak aşağıdaki avantajları sağlar:

  • Bir hizmet aracısını çağırırken, istek gövdesinde {stream: true} atlayarak akış aracısı çıktısını etkinleştirir.

  • Aracınız kullanıma sunulduğunda Yapay Zeka Ağ Geçidi Karar Tablolarını otomatik olarak etkinleştirir ve istek sahibinin adı gibi gelişmiş istek günlüğü meta verilerine erişim sağlar.

    Warning

    İstek günlükleri ve değerlendirme günlükleri kullanım dışıdır ve gelecek bir sürümde kaldırılacaktır. Geçiş kılavuzu için bkz. istek günlükleri ve değerlendirme günlüklerinin kaldırılması.

  • Yazılan Python sınıflarını kullanarak ChatCompletion şemasıyla uyumlu aracı kodu yazmanızı sağlar.

  • MLflow, ajan günlüğe kaydedilirken bir input_example olmasa bile sohbet tamamlamasıyla uyumlu imzayı otomatik olarak çıkartır. Bu, aracıyı kaydetme ve dağıtma işlemini basitleştirir. Bkz. Günlükleme sırasında Model İmzasını Çıkart.

Aşağıdaki kod en iyi şekilde Databricks not defterinde çalıştırılır. Not defterleri, ajanlarınızı geliştirmek, test etmek ve yinelemek için uygun bir ortam sağlar.

MyAgent sınıfı, gerekli mlflow.pyfunc.ChatModel yöntemini uygulayarak predictgenişletir. Bu, Custom Agents ile uyumluluğu sağlar.

sınıfı ayrıca akış çıkışlarını işlemek için _create_chat_completion_chunk ve predict_stream isteğe bağlı yöntemleri içerir.

# Install a pinned version of mlflow
%pip install -U mlflow==2.20.2
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)

Aracı sınıfını MyAgent tek bir not defterinde tanımlarken, ayrı bir sürücü not defteri oluşturmanızı öneririz. Sürücü not defteri, aracıyı Model Kayıt Defteri'ne kaydeder ve Model Sunma kullanarak aracıyı dağıtır.

Bu ayrım, MLflow'un Koddan Alınan Modeller metodolojisini kullanarak modelleri günlüğe kaydetmek için Databricks tarafından önerilen iş akışını izler.

SplitChatMessageRequest giriş şeması (kullanım dışı)

SplitChatMessagesRequest, geçerli sorguyu ve geçmişi ayrı ayrı aracı girişi olarak geçirmenizi sağlar.

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

StringResponse çıkış şeması (kullanım dışı)

StringResponse , aracının yanıtını tek content dize alanına sahip bir nesne olarak döndürmenizi sağlar:

{"content": "This is an example string response"}
  • Last updated on