Поделиться через

Создание агентов ИИ посредством программирования того в коде

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

Требования

Подсказка

Databricks рекомендует установить последнюю версию клиента Python MLflow при разработке агентов.

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

  • databricks-agents 0.16.0 или более поздней версии
  • mlflow 2.20.2 или более поздней версии
  • Python 3.10 или более поздней версии.
    • Для удовлетворения этого требования можно использовать бессерверные вычисления или Databricks Runtime 13.3 LTS или более поздней версии.
%pip install -U -qqqq databricks-agents>=0.16.0 mlflow>=2.20.2

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

LangChain/LangGraph

%pip install -U -qqqq databricks-langchain

Открытый ИИ

%pip install -U -qqqq databricks-openai

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

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

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

Databricks рекомендует использовать интерфейс MLflow ChatAgent для разработки агентов промышленного класса. Эта спецификация схемы чата аналогична, но не совместима со схемой OpenAI ChatCompletion .

ChatAgent легко упаковывает существующие агенты для совместимости Databricks.

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

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

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

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

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

Чтобы узнать, как создать ChatAgent, смотрите примеры в следующем разделе и документацию MLflow - Что такое интерфейс ChatAgent.

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

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

  1. Создание класса оболочки Python, наследуемого от mlflow.pyfunc.ChatAgent.

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

  2. КлассChatAgent требует реализации predict метода для обработки непотоковых запросов.

    predict должен принять:

    • messages: list[ChatAgentMessage] — это список ChatAgentMessage, каждая из которых имеет роль (например, "пользователь" или "помощник"), содержит запрос и идентификатор.

    • (Необязательно) context: Optional[ChatContext] и custom_inputs: Optional[dict] для дополнительных данных.

    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 должен возвращать 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. Преобразование между форматами

    В predict преобразуйте входящие сообщения из list[ChatAgentMessage] в формат, который ожидает ваш агент.

    После того как агент создает ответ, преобразуйте выходные данные в один или несколько ChatAgentMessage объектов и заключите их в ChatAgentResponse.

Подсказка

Автоматическое преобразование выходных данных LangChain

Если вы упаковываете агент LangChain, можно использовать mlflow.langchain.output_parsers.ChatAgentOutputParser для автоматического преобразования выходных данных LangChain в MLflow ChatAgentMessage и ChatAgentResponse схему.

Ниже приведен упрощенный шаблон для конвертации вашего агента.

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

Полные примеры см. в записных книжках в следующем разделе.

примеры ChatAgent

В следующих записных книжках показано, как создавать потоковую передачу и не потоковую передачу ChatAgents с помощью популярных библиотек OpenAI, LangGraph и AutoGen.

LangGraph

Если вы упаковываете агент LangChain, можно использовать mlflow.langchain.output_parsers.ChatAgentOutputParser для автоматического преобразования выходных данных LangChain в MLflow ChatAgentMessage и ChatAgentResponse схему.

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

Возьмите записную книжку

Открытый ИИ

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

Возьмите записную книжку

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

Возьмите записную книжку

Агент только для чата OpenAI

Возьмите записную книжку

AutoGen

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

Возьмите записную книжку

DSPy

Агент только для чата DSPy

Возьмите записную книжку

Чтобы узнать, как расширить возможности этих агентов, добавив инструменты, см. инструменты агента ИИ.

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

Сведения о создании многоагентной системы с помощью Genie см. в статье Использование Genie в системах с несколькими агентами.

Агенты потоковой передачи выходных данных

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

Чтобы создать потоковую передачу ChatAgent, определите predict_stream метод, возвращающий генератор, который передаёт ChatAgentChunk объекты, каждая ChatAgentChunk из которых содержит часть ответа. Узнайте больше об идеальном поведении потоковой передачи ChatAgent в документации MLflow.

В следующем коде показан пример функции predict_stream, полные примеры агентов потоковой передачи см. в примерах 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"]
      )

Создание готовых к развертыванию ChatAgentдля службы моделей Databricks

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

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

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

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

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

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

В этих сценариях MLflow ChatAgent изначально поддерживает поля custom_inputs и custom_outputs.

Предупреждение

Приложение Agent Evaluation review app в данный момент не поддерживает отображение логов для агентов с дополнительными полями ввода.

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

Ноутбук настраиваемого агента схемы OpenAI + PyFunc

Возьмите записную книжку

Блокнот агента специальной схемы LangGraph

Возьмите записную книжку

Предоставить custom_inputs в AI Playground и приложении для оценки агентов

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

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

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

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

    Предоставьте пользовательские входные данные в песочнице для ИИ.

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

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

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

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

Заметка

Databricks рекомендует использовать средства извлечения, предоставляемые пакетами Databricks AI Bridge, такими как databricks_langchain.VectorSearchRetrieverTool и databricks_openai.VectorSearchRetrieverTool, так как они уже соответствуют схеме извлекателя MLflow. См. локальную разработку инструментов векторного поиска с помощью AI Bridge.

Если ваш агент включает в себя диапазоны извлечения с пользовательской схемой, вызовите 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 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"],
)

Заметка

Столбец doc_uri особенно важен при оценке производительности извлекателя. doc_uri — это основной идентификатор документов, возвращаемых извлекателем, позволяющий сравнивать их с эталонными оценочными наборами данных. См. раздел "Оценочные наборы" (MLflow 2).

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

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

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

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

Дополнительные сведения о ModelConfig см. в документации по MLflow.

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

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

Обнаруженные при потоковой передаче ошибки Mosaic AI передаются с последним токеном под databricks_output.error. Ответственность за правильную обработку и отображение этой ошибки лежит на вызывающем клиенте.

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

Дальнейшие действия