Udostępnij za pośrednictwem

Tworzenie agentów AI w kodzie

Na tej stronie pokazano, jak utworzyć agenta sztucznej inteligencji w języku Python przy użyciu programu Mosaic AI Agent Framework i popularnych bibliotek tworzenia agentów, takich jak LangGraph, PyFunc i OpenAI.

Wymagania

Wskazówka

Usługa Databricks zaleca zainstalowanie najnowszej wersji klienta języka Python MLflow podczas opracowywania agentów.

Aby utworzyć i wdrożyć agentów przy użyciu podejścia na tej stronie, 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

Usługa Databricks zaleca również instalowanie pakietów integracji Databricks AI Bridge podczas tworzenia agentów. Te pakiety integracyjne (takie jak databricks-langchain, databricks-openai) udostępniają wspólną warstwę interfejsów API do interakcji z funkcjami sztucznej inteligencji Databricks, takimi jak Databricks AI/BI Genie i Vector Search, w ramach ramy tworzenia agentów i zestawów SDK.

LangChain/LangGraph

%pip install -U -qqqq databricks-langchain

OpenAI

%pip install -U -qqqq databricks-openai

Agenci czystego Pythona

%pip install -U -qqqq databricks-ai-bridge

Użyj ChatAgent do tworzenia agentów

Usługa Databricks zaleca korzystanie z interfejsu MLflow ChatAgent do tworzenia agentów gotowych do wdrożenia w środowisku produkcyjnym. Ta specyfikacja schematu czatu jest podobna do tego, ale nie jest ściśle zgodna ze schematem OpenAI ChatCompletion .

Aplikacja ChatAgent łatwo opakowuje istniejących agentów pod kątem zgodności usługi Databricks.

ChatAgent zapewnia następujące korzyści:

  • Zaawansowane możliwości agenta

    • Obsługa wielu agentów
    • Strumieniowe dane wyjściowe: Włącz interaktywne doświadczenia użytkownika, przesyłając dane strumieniowe w małych partiach.
    • Kompleksowa historia komunikatów wywołujących narzędzia: Zwracanie wielu wiadomości, w tym pośrednich komunikatów wywołujących narzędzia, dla lepszej jakości obsługi i zarządzania przebiegiem rozmów.
    • obsługa potwierdzenia wywołania narzędzi

  • Usprawnione programowanie, wdrażanie i monitorowanie

    • Utwórz agenta przy użyciu dowolnego środowiska: Opakuj dowolnego istniejącego agenta przy użyciu interfejsu ChatAgent, aby uzyskać wbudowaną zgodność z AI Playground, oceną agenta i monitorowaniem agenta.
    • Interfejsy programowania z typowanymi danymi: Pisanie kodu agenta przy użyciu typowanych klas języka Python, korzystając z funkcji autouzupełniania w IDE i notatniku.
    • automatyczne wnioskowanie sygnatur: MLflow automatycznie rozpoznaje ChatAgent sygnatury podczas rejestrowania agenta, upraszczając proces rejestracji i wdrażania. Zobacz Podpis modelu podczas procesu logowania.
    • Tabele wnioskowań wzmocnione przez AI Gateway: Tabele wnioskowań AI Gateway są automatycznie uruchamiane dla wdrożonych agentów, co zapewnia dostęp do szczegółowych metadanych logów żądań.

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

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. KlasaChatAgent wymaga zaimplementowania metody predict do obsługi żądań niebędących strumieniowymi.

    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.

Wskazówka

Automatyczne konwertowanie danych wyjściowych aplikacji LangChain

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.

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 notatnik

OpenAI

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

Pobierz notatnik

Agent do wywoływania narzędzi OpenAI Responses API

Pobierz notatnik

Bot OpenAI tylko do rozmowy

Pobierz notatnik

Autogen

Agent narzędzi wywoływanych przez AutoGen

Pobierz notatnik

DSPy

Agent DSPy przeznaczony wyłącznie do czatu

Pobierz notatnik

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

Przykład wielu agentów

Aby dowiedzieć się, jak utworzyć system z wieloma agentami przy użyciu usługi Genie, zobacz Use Genie in multi-agent systems.

Agenty strumieniowego przesyłania wyjściowego

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. Aby dowiedzieć się więcej na temat idealnego zachowania przesyłania strumieniowego ChatAgent, zapoznaj się z dokumentacją 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"]
      )

Opracowywanie gotowych do wdrożenia ChatAgent dla usługi Databricks Model Serving

Databricks wdraża ChatAgent w środowisku rozproszonym na platformie Databricks Model Serving, co oznacza, że podczas konwersacji wieloetapowej ta sama replika serwera może nie przetwarzać wszystkich żądań. Zwróć uwagę na następujące implikacje dotyczące zarządzania stanem agenta:

  • Unikaj lokalnego buforowania: Podczas wdrażania ChatAgent, nie zakładaj, że ta sama replika będzie obsługiwać wszystkie żądania w konwersacji wieloetapowej. Zrekonstruuj stan wewnętrzny, używając schematu słownikowego ChatAgentRequest dla każdej iteracji.

  • stan bezpieczny wątkowo: projektuj stan agenta, aby stan był bezpieczny wątkowo, zapobiegając konfliktom w środowiskach wielowątkowych.

  • Zainicjuj stan w funkcji predict: za każdym razem, kiedy wywoływana jest funkcja predict, a nie podczas inicjowania ChatAgent. Przechowywanie stanu na poziomie ChatAgent może spowodować przeciek informacji między konwersacjami i powodować konflikty, ponieważ pojedyncza replika ChatAgent może obsługiwać żądania z wielu konwersacji.

dostosowane dane wejściowe i wyjściowe

Niektóre scenariusze mogą wymagać dodatkowych danych wejściowych agenta, takich jak client_type i session_id, lub danych wyjściowych, takich jak odnośniki do źródeł, które nie powinny być uwzględnione w historii czatu w przyszłych interakcjach.

W przypadku tych scenariuszy platforma MLflow ChatAgent natywnie obsługuje pola custom_inputs i custom_outputs.

Ostrzeżenie

Aplikacja do oceny agentów nie obsługuje obecnie generowania śladów dla agentów z dodatkowymi polami wejściowymi.

Zapoznaj się z poniższymi notesami, aby dowiedzieć się, jak ustawić niestandardowe dane wejściowe i wyjściowe.

Notatnik niestandardowego agenta dla schematów OpenAI + PyFunc

Pobierz notatnik

Notatnik agenta niestandardowego schematu LangGraph

Pobierz notatnik

Podaj custom_inputs w AI Playground i w aplikacji do przeglądu agentów

Jeśli agent akceptuje dodatkowe dane wejściowe przy użyciu pola custom_inputs, możesz ręcznie wprowadzać te dane zarówno w aplikacji AI Playground, jak i w aplikacji przeglądającej agenta .

  1. W aplikacji Plac Zabaw AI lub Aplikacja Przegląd Agenta wybierz ikonę koła zębatego ikona koła zębatego.

  2. Włącz custom_inputs.

  3. Podaj obiekt JSON zgodny ze zdefiniowanym schematem wejściowym agenta.

    Podaj własne_wnioski w środowisku testowym AI.

Określanie niestandardowych schematów modułu pobierania

Agenci sztucznej inteligencji często używają modułów pobierania do znajdowania i wykonywania zapytań dotyczących danych bez struktury z indeksów wyszukiwania wektorowego. Na przykład narzędzia do pobierania i śledzenia można znaleźć w temacie Tworzenie i śledzenie narzędzi do pobierania danych nieustrukturyzowanych.

Śledź te detektory w swoim agencie za pomocą obszarów MLflow RETRIEVER, aby włączyć funkcje produktów Databricks, w tym:

  • Automatyczne wyświetlanie linków do pobranych dokumentów źródłowych w interfejsie użytkownika środowiska zabaw sztucznej inteligencji
  • Automatyczne uruchamianie oceny ugruntowania i istotności kryteriów w procesie ewaluacji agenta

Notatka

Databricks zaleca używanie narzędzi pobierania dostarczanych przez pakiety Databricks AI Bridge, takie jak databricks_langchain.VectorSearchRetrieverTool i databricks_openai.VectorSearchRetrieverTool, ponieważ są one już zgodne ze schematem pobierania MLflow. Sprawdź Lokalnie opracuj narzędzia do wyszukiwania wektorowego przy użyciu mostka AI.

Jeśli agent zawiera zakresy pobierania z niestandardowym schematem, wywołaj mlflow.models.set_retriever_schema podczas definiowania agenta w kodzie. Mapuje to kolumny wyjściowe retrievera na oczekiwane pola MLflow (primary_key, text_column, doc_uri).

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 an open-source platform, purpose-built to assist machine learning practitioners...',
#         'doc_uri': 'https://mlflow.org/docs/latest/index.html',
#         'title': 'MLflow: A Tool for Managing the Machine Learning Lifecycle'
#     },
#     {
#         '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"],
)

Notatka

Kolumna doc_uri jest szczególnie ważna podczas oceniania wydajności programu retriever. doc_uri jest głównym identyfikatorem dokumentów zwracanych przez narzędzie wyszukujące, co umożliwia porównywanie ich z zestawami oceny danych rzeczywistych. Zobacz Zestawy ewaluacyjne (MLflow 2).

Parametryzowanie kodu agenta na potrzeby wdrażania w różnych środowiskach

Można parametryzować kod agenta, aby ponownie użyć tego samego kodu agenta w różnych środowiskach.

Parametry to pary klucz-wartość definiowane w słowniku języka Python lub pliku .yaml.

Aby skonfigurować kod, utwórz ModelConfig przy użyciu słownika języka Python lub pliku .yaml . ModelConfig to zestaw parametrów typu klucz-wartość, który umożliwia elastyczne zarządzanie konfiguracją. Na przykład można użyć słownika podczas tworzenia oprogramowania, a następnie przekonwertować go na plik .yaml na potrzeby wdrożenia produkcyjnego i CI/CD.

Aby uzyskać szczegółowe informacje o ModelConfig, zobacz dokumentację MLflow.

Poniżej przedstawiono przykład ModelConfig:

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

W kodzie agenta możesz odwołać się do konfiguracji domyślnej (programowania) z pliku .yaml lub słownika:

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

Następnie, podczas rejestrowania agenta, określ parametr model_config jako log_model, aby określić niestandardowy zestaw parametrów do użycia podczas ładowania zalogowanego agenta. Zobacz dokumentację MLflow — ModelConfig.

Propagacja błędów przesyłania strumieniowego

Mosaic AI rozprzestrzenia wszelkie błędy napotkane podczas przesyłania strumieniowego pod ostatnim tokenem databricks_output.error. Klient wywołujący musi prawidłowo obsłużyć i wyświetlić ten błąd.

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

Następne kroki