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

Начало работы с агентами ИИ

Создайте вашего первого агента ИИ с помощью Платформы агента ИИ Mosaic. В этом руководстве описано следующее:

  • Создание агента с помощью Agent Framework.
  • Добавьте инструмент для вашего агента.
  • Разверните агент в конечной точке обслуживания модели Databricks.

Общие сведения об агентах и других приложениях ИИ поколения см. в разделе "Основные понятия: создание ИИ в Azure Databricks"

Требования

Рабочая область должна иметь следующие функции:

пример записной книжки

Эта записная книжка содержит весь код, необходимый для создания и развертывания первого агента ИИ. Импортируйте записную книжку в рабочую область Azure Databricks, чтобы её запустить.

Демонстрация агента ИИ Mosaic

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

Определение агента

Агент ИИ состоит из следующих элементов:

  • Большая языковая модель (LLM), которая может рассуждать и принимать решения
  • Инструменты, которые LLM может использовать не только для создания текста, но и для выполнения кода Python или получения данных.

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

  1. Установите необходимые пакеты Python:

    %pip install -U -qqqq mlflow databricks-openai databricks-agents"
dbutils.library.restartPython()
    • mlflow: используется для разработки и отслеживания агентов.
    • databricks-openai: используется для подключения к LLM, размещенным на платформе Databricks, и получения доступа к инструментам Unity Catalog.
    • databricks-agents: используется для упаковки и развертывания агента.

  2. Определите агент. В этом фрагменте кода выполняется следующее:

    • Подключается к конечной точке обслуживания модели Databricks с помощью клиента OpenAI.
    • Включает трассировку MLflow с помощью autolog(). Это добавляет инструменты мониторинга, чтобы вы могли видеть, как агент выполняет свои задачи при запросе.
    • Добавляет инструмент system.ai.python_exec вашему агенту. Эта встроенная функция каталога Unity позволяет агенту запускать код Python.
    • Использует вспомогательные функции MLflow (output_to_responses_items_stream, create_function_call_output_item) для преобразования потоковых выходных данных LLM в формат, совместимый с API ответов.
    import json
import mlflow
from databricks.sdk import WorkspaceClient
from databricks_openai import UCFunctionToolkit, DatabricksFunctionClient
# Import MLflow utilities for converting from chat completions to Responses API format
from mlflow.types.responses import output_to_responses_items_stream, create_function_call_output_item

# Enable automatic tracing for easier debugging
mlflow.openai.autolog()

# Get an OpenAI client configured to connect to Databricks model serving endpoints
openai_client = WorkspaceClient().serving_endpoints.get_open_ai_client()

# Load Databricks built-in tools (Python code interpreter)
client = DatabricksFunctionClient()
builtin_tools = UCFunctionToolkit(function_names=["system.ai.python_exec"], client=client).tools
for tool in builtin_tools:
  del tool["function"]["strict"]

def call_tool(tool_name, parameters):
  if tool_name == "system__ai__python_exec":
    return DatabricksFunctionClient().execute_function("system.ai.python_exec", parameters=parameters).value
  raise ValueError(f"Unknown tool: {tool_name}")

def call_llm(prompt):
  for chunk in openai_client.chat.completions.create(
    model="databricks-claude-sonnet-4-5",
    messages=[{"role": "user", "content": prompt}],
    tools=builtin_tools,
    stream=True
  ):
    yield chunk.to_dict()

def run_agent(prompt):
  """
  Send a user prompt to the LLM, and yield LLM + tool call responses
  The LLM is allowed to call the code interpreter tool if needed, to respond to the user
  """
  # Convert output into Responses API-compatible events
  for chunk in output_to_responses_items_stream(call_llm(prompt)):
    yield chunk.model_dump(exclude_none=True)
  # If the model executed a tool, call it and yield the tool call output in Responses API format
  if chunk.item.get('type') == 'function_call':
    tool_name = chunk.item["name"]
    tool_args = json.loads(chunk.item["arguments"])
    tool_result = call_tool(tool_name, tool_args)
    yield {"type": "response.output_item.done", "item": create_function_call_output_item(call_id=chunk.item["call_id"], output=tool_result)}

Тестирование агента

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

for output_chunk in run_agent("What is the square root of 429?"):
  print(output_chunk)

Помимо выходных данных LLM, вы увидите подробные сведения о трассировке непосредственно в вашей рабочей тетради. Эти следы помогают отлаживать медленные или неудачные вызовы агента. Эти следы были автоматически добавлены с помощью mlflow.openai.autolog() .

Развертывание агента

Теперь, когда у вас есть агент, вы можете упаковать и развернуть его в конечной точке обслуживания Databricks. Начните собирать отзывы о развернутом агенте, делясь им с другими пользователями и общаться с ним с помощью встроенного пользовательского интерфейса чата.

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

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

  1. Чтобы реализовать ResponsesAgent интерфейс, определите методы predict_stream() (для потоковых ответов) и predict() (для непотоковых запросов). Так как базовая логика агента уже выводит события, совместимые с API ответа, реализация проста:

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

class QuickstartAgent(ResponsesAgent):
  def predict_stream(self, request: ResponsesAgentRequest):
    # Extract the user's prompt from the request
    prompt = request.input[-1].content
    # Stream response items from our agent
    for chunk in run_agent(prompt):
      yield ResponsesAgentStreamEvent(**chunk)

  def predict(self, request: ResponsesAgentRequest) -> ResponsesAgentResponse:
    outputs = [
      event.item
      for event in self.predict_stream(request)
      if event.type == "response.output_item.done"
    ]
    return ResponsesAgentResponse(output=outputs)

  2. Добавьте следующий код в записную книжку для тестирования ResponsesAgent класса:

    from mlflow.types.responses import ResponsesAgentRequest

AGENT = QuickstartAgent()

# Create a ResponsesAgentRequest with input messages
request = ResponsesAgentRequest(
  input=[
    {
      "role": "user",
      "content": "What's the square root of 429?"
    }
  ]
)

for event in AGENT.predict_stream(request):
  print(event)

  3. Объедините весь код агента в один файл, чтобы вы могли регистрировать и развертывать его.

  • Консолидируйте весь код агента в одну ячейку блокнота.
  • В верхней части ячейки добавьте магическую %%writefile quickstart_agent.py команду, чтобы сохранить агент в файле.
  • В нижней части ячейки вызовите объект агента с помощью mlflow.models.set_model(). Это сообщает MLflow, какой объект агента следует использовать при обслуживании прогнозов. Этот шаг эффективно настраивает точку входа в код агента.

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

%%writefile quickstart_agent.py

import json
from databricks.sdk import WorkspaceClient
from databricks_openai import UCFunctionToolkit, DatabricksFunctionClient

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

# Enable automatic tracing for deployed agent
mlflow.openai.autolog()

# Get an OpenAI client configured to talk to Databricks model serving endpoints
openai_client = WorkspaceClient().serving_endpoints.get_open_ai_client()

# Load Databricks built-in tools (Python code interpreter)
client = DatabricksFunctionClient()
builtin_tools = UCFunctionToolkit(function_names=["system.ai.python_exec"], client=client).tools
for tool in builtin_tools:
  del tool["function"]["strict"]

def call_tool(tool_name, parameters):
  if tool_name == "system__ai__python_exec":
    return DatabricksFunctionClient().execute_function("system.ai.python_exec", parameters=parameters).value
  raise ValueError(f"Unknown tool: {tool_name}")

def call_llm(prompt):
  for chunk in openai_client.chat.completions.create(
    model="databricks-claude-sonnet-4-5",
    messages=[{"role": "user", "content": prompt}],
    tools=builtin_tools,
    stream=True
  ):
    yield chunk.to_dict()

def run_agent(prompt):
  """
  Send a user prompt to the LLM, and yield LLM + tool call responses
  The LLM is allowed to call the code interpreter tool if needed, to respond to the user
  """
  # Convert output into Responses API-compatible events
  for chunk in output_to_responses_items_stream(call_llm(prompt)):
    yield chunk.model_dump(exclude_none=True)
  # If the model executed a tool, call it and yield the tool call output in Responses API format
  if chunk.item.get('type') == 'function_call':
    tool_name = chunk.item["name"]
    tool_args = json.loads(chunk.item["arguments"])
    tool_result = call_tool(tool_name, tool_args)
    yield {"type": "response.output_item.done", "item": create_function_call_output_item(call_id=chunk.item["call_id"], output=tool_result)}

class QuickstartAgent(ResponsesAgent):
  def predict_stream(self, request: ResponsesAgentRequest):
    # Extract the user's prompt from the request
    prompt = request.input[-1].content
    # Stream response items from our agent
    for chunk in run_agent(prompt):
      yield ResponsesAgentStreamEvent(**chunk)

  def predict(self, request: ResponsesAgentRequest) -> ResponsesAgentResponse:
    outputs = [
      event.item
      for event in self.predict_stream(request)
      if event.type == "response.output_item.done"
    ]
    return ResponsesAgentResponse(output=outputs)

AGENT = QuickstartAgent()
mlflow.models.set_model(AGENT)

Захегистрируйте агента

Внесите данные о вашем агенте и зарегистрируйте его в каталоге Unity. Это упаковывает агента и его зависимости в один артефакт для развертывания.

import mlflow
from mlflow.models.resources import DatabricksFunction, DatabricksServingEndpoint
from pkg_resources import get_distribution

# Change the catalog name ("main") and schema name ("default") to register the agent to a different location
registered_model_name = "main.default.quickstart_agent"

# Specify Databricks resources that the agent needs to access.
# This step lets Databricks automatically configure authentication
# so the agent can access these resources when it's deployed.
resources = [
  DatabricksServingEndpoint(endpoint_name="databricks-claude-sonnet-4-5"),
  DatabricksFunction(function_name="system.ai.python_exec"),
]

mlflow.set_registry_uri("databricks-uc")
logged_agent_info = mlflow.pyfunc.log_model(
  artifact_path="agent",
  python_model="quickstart_agent.py",
  extra_pip_requirements=[f"databricks-connect=={get_distribution('databricks-connect').version}"],
  resources=resources,
  registered_model_name=registered_model_name
)

Развертывание агента

Разверните зарегистрированного агента в точке доступа:

from databricks import agents

deployment_info = agents.deploy(
  model_name=registered_model_name,
  model_version=logged_agent_info.registered_model_version,
  scale_to_zero=True
)

После запуска конечной точки агента вы можете общаться с ним с помощью ИИ-площадки или поделиться им с заинтересованными лицами для получения отзывов.

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

Выберите, куда идти дальше на основе ваших целей:

Измерение и улучшение качества агента. См. краткое руководство по оценке агента.

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

Узнайте, как создавать агенты с помощью других платформ: узнайте, как создавать агенты с помощью популярных библиотек, таких как LangGraph, чистый Python и OpenAI. См . раздел "Создание агентов ИИ" в коде

  • Last updated on