Создайте агента ИИ и разместите его в среде обслуживания моделей

На этой странице показано, как создать агент ИИ в Python с помощью настраиваемых агентов и популярных библиотек разработки агентов, таких как LangGraph и OpenAI.

Требования

Чтобы создать и развернуть агентов, используя подход, описанный на этой странице, установите следующее:

• databricks-agents 1.2.0 или более поздней версии
• mlflow 3.1.3 или более поздней версии
• Python 3.10 или более поздней версии.
  • Используйте бессерверные вычисления или Databricks Runtime 13.3 LTS или более поздней версии, чтобы удовлетворить это требование.

%pip install -U -qqqq databricks-agents mlflow

Databricks также рекомендует устанавливать пакеты интеграции Databricks AI Bridge для агентов разработки. Эти пакеты интеграции предоставляют общий уровень API, взаимодействующих с функциями ИИ Databricks, такими как Genie Spaces и векторный поиск, в платформах разработки агентов и пакетах SDK.

Открытый ИИ

%pip install -U -qqqq databricks-openai

LangChain/LangGraph

%pip install -U -qqqq databricks-langchain

DSPy

%pip install -U -qqqq databricks-dspy

Чистые агенты Python

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

Использовать ResponsesAgent для создания агентов

Databricks рекомендует использовать интерфейс ResponsesAgent MLflow для создания агентов промышленного класса. ResponsesAgent позволяет создавать агенты с любой сторонней платформой, а затем интегрировать ее с функциями ИИ Databricks для надежного ведения журнала, трассировки, оценки, развертывания и мониторинга.

Схема ResponsesAgent совместима с схемой OpenAI Responses . Чтобы узнать больше об OpenAIResponses, см. OpenAI: Responses vs. ChatCompletion.

ResponsesAgent предоставляет следующие преимущества:

• Расширенные возможности агента

  • Поддержка нескольких агентов
  • Потоковая передача выходных данных: передавайте выходные данные небольшими блоками.
  • Подробная история сообщений вызова инструментов: Возвращает несколько сообщений, включая промежуточные сообщения вызова инструментов, для повышения качества и более эффективного управления диалогами.
  • Поддержка подтверждения активации инструмента
  • Поддержка долговременно работающего инструмента

• Оптимизация разработки, развертывания и мониторинга

  • Создание агентов с помощью любой платформы: обертывание любого существующего агента с помощью ResponsesAgent интерфейса для обеспечения совместимости с ИИ-площадкой, оценкой агента и мониторингом агентов.
  • Интерфейсы разработки с типизацией: создание кода агента с использованием типизированных классов Python, что позволяет использовать преимущества автозаполнения в записных книжках и интегрированной среды разработки.
  • Автоматическое вывод подписи: MLflow автоматически выводит ResponsesAgent подписи при ведении журнала агента, упрощая регистрацию и развертывание. См. определение сигнатуры модели при журналировании.
  • Автоматическая трассировка: MLflow автоматически трассирует ваши функции predict и predict_stream, агрегируя потоковые ответы для упрощения оценки и отображения.
  • таблицы выводов, улучшенные с помощью шлюза ИИ: таблицы выводов шлюза ИИ автоматически включены для развернутых агентов, предоставляя доступ к подробным метаданным журнала запросов.

Чтобы узнать, как создать модель ResponsesAgent, см. примеры в следующем разделе и документации по MLflow — ResponsesAgent для службы моделей.

ResponsesAgent Примеры

В следующих записных книжках показано, как создавать потоковую передачу и не потоковую передачу ResponsesAgent с помощью популярных библиотек. Сведения о расширении возможностей этих агентов см. в разделе "Средства агента ИИ".

Открытый ИИ

Простой агент чата OpenAI с помощью моделей, размещенных в Databricks

Получите ноутбук

Агент вызова инструментов OpenAI MCP

Получите ноутбук

Агент вызова инструментов OpenAI с помощью моделей, размещенных в Databricks

Получите ноутбук

Агент вызова инструментов OpenAI с помощью моделей, размещенных в OpenAI

Получите ноутбук

LangGraph

Агент вызова инструментов LangGraph MCP

Получите ноутбук

DSPy

Агент однократного вызова инструмента DSPy

Получите ноутбук

Пример нескольких агентов

Сведения о создании многоагентной системы см. в статье "Использование Genie" в системах с несколькими агентами (обслуживание моделей).

Пример состояниевого агента

Чтобы узнать, как создать агентов с отслеживанием состояния с краткосрочной и долгосрочной памятью, используя Lakebase в качестве хранилища памяти, см. в разделе AI agent memory (Model Serving).

Пример агента без общения

В отличие от разговорных агентов, которые ведут диалоги с несколькими этапами, недиалоговые агенты сосредоточены на эффективном выполнении четко определенных задач. Эта оптимизированная архитектура обеспечивает более высокую пропускную способность для независимых запросов.

Чтобы узнать, как создать неконверсированного агента, см. раздел "Агенты неконверсированного искусственного интеллекта с помощью MLflow".

Что делать, если у меня уже есть агент?

Если у вас уже есть агент, созданный с помощью LangChain, LangGraph или аналогичной платформы, вам не нужно переписать агент, чтобы использовать его в Databricks. Вместо этого просто оберните существующего агента в интерфейс MLflow ResponsesAgent.

1. Напишите класс оболочки Python, наследующий от mlflow.pyfunc.ResponsesAgent.

   В классе-оболочке используйте ссылку на существующий агент в качестве атрибута self.agent = your_existing_agent.

2. Класс ResponsesAgent требует реализации метода predict, который возвращает ResponsesAgentResponse для обработки запросов, отличных от потоковых. Ниже приведен пример 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 преобразуйте входящие сообщения из ResponsesAgentRequest в формат, ожидаемый агентом. После того как агент создает ответ, преобразуйте его результат ResponsesAgentResponse в объект.

Ознакомьтесь со следующими примерами кода, чтобы узнать, как преобразовать существующие агенты в ResponsesAgent:

Базовое преобразование

Для агентов без потоковой передачи преобразуйте входные и выходные данные в функцию predict .

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

Передача данных с повторным использованием кода

Для агентов потоковой передачи вы можете быть умными и повторно использовать логику, чтобы избежать дублирования кода, который преобразует сообщения:

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

Миграция из ChatCompletions

Если существующий агент использует API OpenAI ChatCompletions, его можно перенести ResponsesAgent без перезаписи основной логики. Добавьте оболочку, которая:

1. Преобразует входящие ResponsesAgentRequest сообщения в ChatCompletions формат, который ожидает агент.
2. ChatCompletions Преобразует выходные данные в схемуResponsesAgentResponse.
3. При необходимости поддерживает потоковую передачу путем сопоставления инкрементальных дельт из ChatCompletions в ResponsesAgentStreamEvent объекты.

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)

Для полноты картины смотрите ResponsesAgent примеры.

Потоковая передача ответов

Потоковая передача позволяет агентам отправлять ответы в блоках в режиме реального времени вместо ожидания полного ответа. Чтобы реализовать потоковую передачу данных с помощью ResponsesAgent, выведите ряд разностных событий, за которым следует окончательное событие завершения.

1. Отправка разностных событий: отправка нескольких output_text.delta событий с одинаковыми item_id для потоковой передачи фрагментов текста в режиме реального времени.
2. Завершите с событием "done": отправьте финальное response.output_item.done событие с тем же item_id, что и разностные события, содержащие полный итоговый выходной текст.

Каждое разностное событие передает фрагмент текста клиенту. Последнее готовое событие содержит полный текст ответа и сигнал Databricks выполнить следующее:

• Отслеживайте выходные данные агента с помощью трассировки MLflow
• Агрегированные потоковые ответы в таблицах вывода шлюза искусственного интеллекта
• Отображение полных выходных данных в пользовательском интерфейсе игровой площадки ИИ

Распространение ошибок потоковой передачи

Azure Databricks передаёт все ошибки, возникающие в ходе потоковой обработки, с последним токеном под databricks_output.error. Ответственность за правильную обработку и отображение этой ошибки лежит на вызывающем клиенте.

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

Расширенные функции

пользовательские настройки входов и выходов

В некоторых сценариях могут потребоваться дополнительные данные агента, такие как client_type, или выходные данные, такие как ссылки на источник извлечения session_id, которые не должны быть включены в журнал чата для будущих взаимодействий.

В этих сценариях MLflow ResponsesAgent изначально поддерживает поля custom_inputs и custom_outputs. Вы можете получить доступ к пользовательским входным данным через request.custom_inputs во всех приведенных выше примерах в ResponsesAgent Examples.

Сведения о настройке пользовательских входных и выходных данных см. в следующих записных книжках.

Предоставьте custom_inputs в AI Playground и приложении для обзоров

Если агент принимает дополнительные входные данные с помощью custom_inputs поля, вы можете вручную указать эти входные данные как в игровой площадке ИИ , так и в приложении проверки.

1. На игровой площадке ИИ или в приложении проверки агента выберите

2. Включите custom_inputs.

3. Укажите объект JSON, соответствующий определенной схеме входных данных агента.

   

Определите пользовательские схемы извлекателя

Агенты ИИ обычно используют средства извлечения для поиска и запроса неструктурированных данных из индексов векторного поиска. Например, средства извлечения см. в разделе "Подключение агентов к неструктурированным данным".

Отслеживайте этих ретриверов с помощью вашего агента, используя интервалы MLflow RETRIEVER, чтобы задействовать функции продукта Databricks, такие как:

• Автоматическое отображение ссылок на извлеченные исходные документы в пользовательском интерфейсе игровой площадки ИИ
• Автоматическое выполнение оценки обоснованности и релевантности в системе оценки агентов

Если агент включает диапазоны извлекателя с пользовательской схемой, вызовите mlflow.models.set_retriever_schema при определении агента в коде. Это сопоставляет выходные столбцы извлекателя с ожидаемыми полями 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 the largest open source AI engineering platform for agents, LLMs, and ML models...',
#         'doc_uri': 'https://mlflow.org/docs/latest/index.html',
#         'title': 'MLflow: The Largest Open Source AI Engineering Platform'
#     },
#     {
#         '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"],
)

Рекомендации по развертыванию

Подготовка к обслуживанию модели Databricks

Databricks развертывает ResponsesAgentв распределенной среде в Службе моделей Databricks. Это означает, что во время многоэтапной беседы одна и та же реплика обслуживания может не обрабатывать все запросы. Обратите внимание на следующие импликации для управления состоянием агента:

• Избегайте локального кэширования: ResponsesAgentпри развертывании не предполагается, что одна и та же реплика обрабатывает все запросы в многоэтапной беседе. Восстанавливать внутреннее состояние с помощью схемы словаря ResponsesAgentRequest для каждого шага.

• Потокобезопасное состояние: состояние агента проектирования должно быть потокобезопасным для предотвращения конфликтов в многопоточных средах.

• инициализировать состояние в функции predict: инициализировать состояние при каждом вызове функции predict, а не во время инициализации ResponsesAgent. Хранение состояния на уровне ResponsesAgent может привести к утечке информации между разговорами и конфликтам, так как одна реплика ResponsesAgent может обрабатывать запросы из нескольких разговоров.

Параметризуйте код для развертывания в разных средах

Параметризуйте код агента для повторного использования одного и того же кода агента в разных средах.

Параметры — это пары "ключ-значение", которые определяются в словаре Python или файле .yaml.

Чтобы настроить код, создайте ModelConfig с помощью словаря Python или файла .yaml. ModelConfig — это набор параметров типа "ключ-значение", который обеспечивает гибкое управление конфигурацией. Например, можно использовать словарь во время разработки, а затем преобразовать его в файл .yaml для рабочего развертывания и CI/CD.

Ниже показан пример 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

В коде агента можно ссылаться на конфигурацию по умолчанию (разработка) из файла или словаря .yaml:

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

Затем при логировании вашего агента укажите параметр model_config как log_model, чтобы задать настраиваемый набор параметров, используемых при загрузке записанного агента. См. документацию по MLflow — ModelConfig.

Использование синхронного кода или шаблонов обратного вызова

Чтобы обеспечить стабильность и совместимость, используйте синхронный код или шаблоны на основе обратных вызовов в реализации агента.

Azure Databricks автоматически управляет асинхронным взаимодействием, чтобы обеспечить оптимальную параллелизм и производительность при развертывании агента. Введение пользовательских циклов событий или асинхронных платформ может привести к ошибкам, таким как RuntimeError: This event loop is already running and caused unpredictable behavior.

Azure Databricks рекомендует избегать асинхронного программирования, например использования asyncio или создания пользовательских циклов обработки событий при разработке агентов.

Дальнейшие шаги

• Создайте свои собственные инструменты агента.
• Запишите действие агента ИИ.
• Добавить трассировки для агента ИИ.
• Разверните агента ИИ.