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 i OpenAI.

Wymagania

Aby utworzyć i wdrożyć agentów przy użyciu podejścia na tej stronie, zainstaluj następujące elementy:

• databricks-agents 1.2.0 lub nowszy
• mlflow 3.1.3 lub nowsze
• Środowisko Python w wersji 3.10 lub nowszej.
  • Użyj bezserwerowych obliczeń lub środowiska Databricks Runtime 13.3 LTS lub nowszego, aby spełnić to wymaganie.

%pip install -U -qqqq databricks-agents mlflow

Usługa Databricks zaleca również instalowanie pakietów integracji mostka AI usługi Databricks w celu tworzenia agentów. Te pakiety integracji udostępniają udostępnioną warstwę interfejsów API, które współdziałają z funkcjami sztucznej inteligencji usługi Databricks, takimi jak databricks AI/BI Genie i wyszukiwanie wektorów, między strukturami tworzenia agentów i zestawami SDK.

OpenAI

%pip install -U -qqqq databricks-openai

LangChain/LangGraph

%pip install -U -qqqq databricks-langchain

DSPy

%pip install -U -qqqq databricks-dspy

Agenci czystego Pythona

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

Użyj ResponsesAgent do tworzenia agentów

Usługa Databricks zaleca interfejs ResponsesAgent MLflow do tworzenia agentów klasy produkcyjnej. ResponsesAgent Umożliwia tworzenie agentów z dowolną strukturą innej firmy, a następnie integrowanie jej z funkcjami sztucznej inteligencji usługi Databricks w celu uzyskania niezawodnych funkcji rejestrowania, śledzenia, oceny, wdrażania i monitorowania.

ResponsesAgent Schemat jest zgodny ze schematem OpenAIResponses. Aby dowiedzieć się więcej na temat interfejsu Responses: odpowiedzi a chatCompletion).

ResponsesAgent zapewnia następujące korzyści:

• Zaawansowane możliwości agenta

  • Obsługa wielu agentów
  • Dane wyjściowe przesyłania strumieniowego: przesyłaj strumieniowo dane wyjściowe w mniejszych fragmentach.
  • 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
  • Obsługa długotrwałych narzędzi

• Usprawnione programowanie, wdrażanie i monitorowanie

  • Tworzenie agentów przy użyciu dowolnej platformy: Opakuj dowolnego istniejącego agenta przy użyciu interfejsu ResponsesAgent , aby uzyskać wbudowaną zgodność z narzędziami AI Playground, Agent Evaluation i Agent Monitoring.
  • 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 podpisów: narzędzie MLflow automatycznie wywnioskuje ResponsesAgent podpisy podczas rejestrowania agenta, upraszczając rejestrację i wdrażanie. Zobacz Podpis modelu podczas procesu logowania.
  • Automatyczne śledzenie: narzędzie MLflow automatycznie śledzi dane predict i predict_stream funkcje, agregując przesyłane strumieniowo odpowiedzi w celu łatwiejszej oceny i wyświetlania.
  • 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ć obiekt ResponsesAgent, zapoznaj się z przykładami w poniższej sekcji i dokumentacją platformy MLflow — ResponsesAgent for Model Serving (OdpowiedziAgent dla obsługi modeli).

ResponsesAgent Przykłady

W poniższych notesach przedstawiono sposób tworzenia przesyłania strumieniowego i przesyłania strumieniowego bez przesyłania strumieniowego ResponsesAgent przy użyciu popularnych bibliotek. Aby dowiedzieć się, jak rozszerzyć możliwości tych agentów, zobacz Narzędzia agenta sztucznej inteligencji.

OpenAI

Prost agent czatu OpenAI używający modeli hostowanych na Databricks

Pobierz notatnik

Agent wywołujący narzędzia OpenAI przy użyciu modeli hostowanych w usłudze Databricks

Pobierz notatnik

Agent wywołujący narzędzia OpenAI przy użyciu modeli hostowanych za pomocą interfejsu OpenAI

Pobierz notatnik

LangGraph

Agent wywoływania narzędzi LangGraph

Pobierz notatnik

DSPy

Agent jednokrotnego wywoływania narzędzi DSPy

Pobierz notatnik

Przykład wielu agentów

Aby dowiedzieć się, jak utworzyć system z wieloma agentami, zobacz Używanie usługi Genie w systemach z wieloma agentami.

Przykład agenta stanowego

Aby dowiedzieć się, jak tworzyć agentów stanowych z krótkoterminową i długoterminową pamięcią przy użyciu usługi Lakebase jako magazynu pamięci, zobacz Pamięć agenta sztucznej inteligencji.

Przykład agenta nieinteraktywnego

W przeciwieństwie do agentów konwersacyjnych, które zarządzają dialogami wieloetapowymi, agenci niekonwersacyjni koncentrują się na wydajnym realizowaniu dobrze zdefiniowanych zadań. Ta usprawniona architektura zapewnia większą przepływność dla żądań niezależnych.

Aby dowiedzieć się, jak utworzyć agenta niekonwersacyjnego, zobacz Niekonwersacyjne agenci sztucznej inteligencji przy użyciu platformy MLflow.

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 ResponsesAgent.

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

   Wewnątrz klasy otoki odwołaj się do istniejącego agenta jako atrybutu self.agent = your_existing_agent.

2. Klasa ResponsesAgent wymaga zaimplementowania predict metody, która zwraca ResponsesAgentResponse obiekt w celu obsługi żądań nieprzesyłania strumieniowego. Poniżej przedstawiono przykład schematu ResponsesAgentResponses :

   import uuid
   # input as a dict
   {"input": [{"role": "user", "content": "What did the data scientist say when their Spark job finally completed?"}]}
   
   # output example
   ResponsesAgentResponse(
       output=[
           {
               "type": "message",
               "id": str(uuid.uuid4()),
               "content": [{"type": "output_text", "text": "Well, that really sparked joy!"}],
               "role": "assistant",
           },
       ]
   )
   

3. predict W funkcji przekonwertuj komunikaty przychodzące z ResponsesAgentRequest na format oczekiwany przez agenta. Po wygenerowaniu odpowiedzi przez agenta przekonwertuj ResponsesAgentResponse jego dane wyjściowe na obiekt.

Zobacz następujące przykłady kodu, aby zobaczyć, jak przekonwertować istniejących agentów na ResponsesAgent:

Konwersja podstawowa

W przypadku agentów spoza przesyłania strumieniowego przekonwertuj predict dane wejściowe i wyjściowe w funkcji .

from uuid import uuid4

from mlflow.pyfunc import ResponsesAgent
from mlflow.types.responses import (
    ResponsesAgentRequest,
    ResponsesAgentResponse,
)


class MyWrappedAgent(ResponsesAgent):
    def __init__(self, agent):
        # Reference your existing agent
        self.agent = agent

    def predict(self, request: ResponsesAgentRequest) -> ResponsesAgentResponse:
        # Convert incoming messages to your agent's format
        # prep_msgs_for_llm is a function you write to convert the incoming messages
        messages = self.prep_msgs_for_llm([i.model_dump() for i in request.input])

        # Call your existing agent (non-streaming)
        agent_response = self.agent.invoke(messages)

        # Convert your agent's output to ResponsesAgent format, assuming agent_response is a str
        output_item = (self.create_text_output_item(text=agent_response, id=str(uuid4())),)

        # Return the response
        return ResponsesAgentResponse(output=[output_item])

Przesyłanie strumieniowe za pomocą ponownego użycia kodu

W przypadku agentów przesyłania strumieniowego możesz być sprytną i ponownie używać logiki, aby uniknąć duplikowania kodu, który konwertuje komunikaty:

from typing import Generator
from uuid import uuid4

from mlflow.pyfunc import ResponsesAgent
from mlflow.types.responses import (
    ResponsesAgentRequest,
    ResponsesAgentResponse,
    ResponsesAgentStreamEvent,
)


class MyWrappedStreamingAgent(ResponsesAgent):
    def __init__(self, agent):
        # Reference your existing agent
        self.agent = agent

    def predict(self, request: ResponsesAgentRequest) -> ResponsesAgentResponse:
        """Non-streaming predict: collects all streaming chunks into a single response."""
        # Reuse the streaming logic and collect all output items
        output_items = []
        for stream_event in self.predict_stream(request):
            if stream_event.type == "response.output_item.done":
                output_items.append(stream_event.item)

        # Return all collected items as a single response
        return ResponsesAgentResponse(output=output_items)

    def predict_stream(
        self, request: ResponsesAgentRequest
    ) -> Generator[ResponsesAgentStreamEvent, None, None]:
        """Streaming predict: the core logic that both methods use."""
        # Convert incoming messages to your agent's format
        # prep_msgs_for_llm is a function you write to convert the incoming messages, included in full examples linked below
        messages = self.prep_msgs_for_llm([i.model_dump() for i in request.input])

        # Stream from your existing agent
        item_id = str(uuid4())
        aggregated_stream = ""
        for chunk in self.agent.stream(messages):
            # Convert each chunk to ResponsesAgent format
            yield self.create_text_delta(delta=chunk, item_id=item_id)
            aggregated_stream += chunk

        # Emit an aggregated output_item for all the text deltas with id=item_id
        yield ResponsesAgentStreamEvent(
            type="response.output_item.done",
            item=self.create_text_output_item(text=aggregated_stream, id=item_id),
        )

Migrowanie z aplikacji ChatCompletions

Jeśli istniejący agent używa interfejsu API OpenAI ChatCompletions, możesz przeprowadzić migrację do ResponsesAgent niego bez ponownego zapisywania podstawowej logiki. Dodaj otokę, która:

1. Konwertuje komunikaty przychodzące ResponsesAgentRequest na ChatCompletions format oczekiwany przez agenta.
2. ChatCompletions Przekształca dane wyjściowe w ResponsesAgentResponse schemat.
3. Opcjonalnie obsługuje przesyłanie strumieniowe przez mapowanie przyrostowych różnic z ChatCompletions do ResponsesAgentStreamEvent obiektów.

from typing import Generator
from uuid import uuid4

from databricks.sdk import WorkspaceClient
from mlflow.pyfunc import ResponsesAgent
from mlflow.types.responses import (
    ResponsesAgentRequest,
    ResponsesAgentResponse,
    ResponsesAgentStreamEvent,
)


# Legacy agent that outputs ChatCompletions objects
class LegacyAgent:
    def __init__(self):
        self.w = WorkspaceClient()
        self.OpenAI = self.w.serving_endpoints.get_open_ai_client()

    def stream(self, messages):
        for chunk in self.OpenAI.chat.completions.create(
            model="databricks-claude-sonnet-4-5",
            messages=messages,
            stream=True,
        ):
            yield chunk.to_dict()


# Wrapper that converts the legacy agent to a ResponsesAgent
class MyWrappedStreamingAgent(ResponsesAgent):
    def __init__(self, agent):
        # `agent` is your existing ChatCompletions agent
        self.agent = agent

    def prep_msgs_for_llm(self, messages):
        # dummy example of prep_msgs_for_llm
        # real example of prep_msgs_for_llm included in full examples linked below
        return [{"role": "user", "content": "Hello, how are you?"}]

    def predict(self, request: ResponsesAgentRequest) -> ResponsesAgentResponse:
        """Non-streaming predict: collects all streaming chunks into a single response."""
        # Reuse the streaming logic and collect all output items
        output_items = []
        for stream_event in self.predict_stream(request):
            if stream_event.type == "response.output_item.done":
                output_items.append(stream_event.item)

        # Return all collected items as a single response
        return ResponsesAgentResponse(output=output_items)

    def predict_stream(
        self, request: ResponsesAgentRequest
    ) -> Generator[ResponsesAgentStreamEvent, None, None]:
        """Streaming predict: the core logic that both methods use."""
        # Convert incoming messages to your agent's format
        messages = self.prep_msgs_for_llm([i.model_dump() for i in request.input])

        # process the ChatCompletion output stream
        agent_content = ""
        tool_calls = []
        msg_id = None
        for chunk in self.agent.stream(messages):  # call the underlying agent's stream method
            delta = chunk["choices"][0]["delta"]
            msg_id = chunk.get("id", None)
            content = delta.get("content", None)
            if tc := delta.get("tool_calls"):
                if not tool_calls:  # only accommodate for single tool call right now
                    tool_calls = tc
                else:
                    tool_calls[0]["function"]["arguments"] += tc[0]["function"]["arguments"]
            elif content is not None:
                agent_content += content
                yield ResponsesAgentStreamEvent(**self.create_text_delta(content, item_id=msg_id))

        # aggregate the streamed text content
        yield ResponsesAgentStreamEvent(
            type="response.output_item.done",
            item=self.create_text_output_item(agent_content, msg_id),
        )

        for tool_call in tool_calls:
            yield ResponsesAgentStreamEvent(
                type="response.output_item.done",
                item=self.create_function_call_item(
                    str(uuid4()),
                    tool_call["id"],
                    tool_call["function"]["name"],
                    tool_call["function"]["arguments"],
                ),
            )


agent = MyWrappedStreamingAgent(LegacyAgent())

for chunk in agent.predict_stream(
    ResponsesAgentRequest(input=[{"role": "user", "content": "Hello, how are you?"}])
):
    print(chunk)

Aby zapoznać się z kompletnymi przykładami, zobacz ResponsesAgent przykłady.

Odpowiedzi przesyłania strumieniowego

Przesyłanie strumieniowe umożliwia agentom wysyłanie odpowiedzi we fragmentach w czasie rzeczywistym zamiast czekać na pełną odpowiedź. Aby zaimplementować przesyłanie strumieniowe za pomocą ResponsesAgentprogramu , emituj serię zdarzeń różnicowych, po których następuje końcowe zdarzenie ukończenia:

1. Emituj zdarzenia różnicowe: wysyłaj wiele output_text.delta zdarzeń o tym samym item_id rozmiarze, aby przesyłać strumieniowo fragmenty tekstu w czasie rzeczywistym.
2. Zakończ z gotowym zdarzeniem: wyślij ostateczne response.output_item.done zdarzenie z takim samym item_id , jak zdarzenia różnicowe zawierające pełny końcowy tekst wyjściowy.

Każde zdarzenie różnicowe przesyła strumieniowo fragment tekstu do klienta. Ostateczne gotowe zdarzenie zawiera pełny tekst odpowiedzi i sygnały usługi Databricks w celu wykonania następujących czynności:

• Śledzenie danych wyjściowych agenta za pomocą śledzenia MLflow
• Agregowanie odpowiedzi przesyłanych strumieniowo w tabelach wnioskowania bramy sztucznej inteligencji
• Wyświetlanie pełnych danych wyjściowych w interfejsie użytkownika platformy AI Playground

Propagacja błędów przesyłania strumieniowego

Mozaika AI propaguje wszelkie błędy napotkane podczas przesyłania strumieniowego przy użyciu ostatniego tokenu w obszarze 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."
    }
  }
}

Zaawansowane funkcje

dostosowane dane wejściowe i wyjściowe

W niektórych scenariuszach mogą być wymagane dodatkowe dane wejściowe agenta, takie jak i , client_typelub dane wyjściowe, takie jak session_id linki źródłowe pobierania, które nie powinny być uwzględniane w historii czatów na potrzeby przyszłych interakcji.

W przypadku tych scenariuszy platforma MLflow ResponsesAgent natywnie obsługuje pola custom_inputs i custom_outputs. Dostęp do niestandardowych danych wejściowych można uzyskać za pośrednictwem request.custom_inputs we wszystkich przykładach połączonych powyżej w sekcji ResponsesAgent Examples (Przykłady odpowiedzi).

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

Podaj custom_inputs w środowisku zabaw dla sztucznej inteligencji i przejrzyj aplikację

Jeśli agent akceptuje dodatkowe dane wejściowe przy użyciu custom_inputs pola, możesz ręcznie podać te dane wejściowe zarówno w środowisku zabaw dla sztucznej inteligencji, jak i w aplikacji do przeglądu.

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

2. Włącz custom_inputs.

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

   

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

Jeśli agent zawiera zakresy pobierania z niestandardowym schematem, wywołaj mlflow.models.set_retriever_schema metodę 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"],
)

Uwagi dotyczące wdrażania

Przygotowywanie do obsługi modeli usługi Databricks

Usługa Databricks wdraża ResponsesAgentpliki w środowisku rozproszonym w usłudze Databricks Model Serving. Oznacza to, że podczas konwersacji wieloeściowej ta sama replika obsługująca może nie obsługiwać wszystkich żądań. Zwróć uwagę na następujące implikacje dotyczące zarządzania stanem agenta:

• Unikaj buforowania lokalnegoResponsesAgent: podczas wdrażania elementu nie zakładaj, że ta sama replika obsługuje wszystkie żądania w konwersacji wieloeściowej. Zrekonstruuj stan wewnętrzny, używając schematu słownikowego ResponsesAgentRequest 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 ResponsesAgent. Przechowywanie stanu na poziomie ResponsesAgent może spowodować przeciek informacji między konwersacjami i powodować konflikty, ponieważ pojedyncza replika ResponsesAgent może obsługiwać żądania z wielu konwersacji.

Parametryzowanie kodu na potrzeby wdrażania w środowiskach

Parametryzowanie kodu agenta w celu ponownego użycia 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 element przy użyciu słownika języka Python lub .yaml pliku. 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.

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.

Używanie synchronicznego kodu lub wzorców wywołań zwrotnych

Aby zapewnić stabilność i zgodność, użyj synchronicznego kodu lub wzorców opartych na wywołaniu zwrotnym w implementacji agenta.

Usługa Azure Databricks automatycznie zarządza komunikacją asynchroniczną w celu zapewnienia optymalnej współbieżności i wydajności podczas wdrażania agenta. Wprowadzenie niestandardowych pętli zdarzeń lub struktur asynchronicznych może prowadzić do błędów, takich jak RuntimeError: This event loop is already running and caused unpredictable behavior.

Usługa Azure Databricks zaleca unikanie programowania asynchronicznego, takich jak używanie asyncio lub tworzenie niestandardowych pętli zdarzeń podczas opracowywania agentów.

Następne kroki

• Utwórz własne narzędzia agenta.
• Zarejestruj agenta sztucznej inteligencji.
• Dodaj ślady do agenta sztucznej inteligencji.
• wdróż agenta sztucznej inteligencji.