Partilhar via


Introdução aos agentes de IA

Crie seu primeiro agente de IA usando o Mosaic AI Agent Framework. Neste tutorial, você irá:

  • Crie um agente usando o Agent Framework.
  • Adicione uma ferramenta ao seu agente.
  • Implemente o seu agente para um ponto de serviço de modelo Databricks.

Para obter uma introdução conceitual aos agentes e outros aplicativos de IA de geração, consulte O que são aplicativos de IA de geração?

Requerimentos

Seu espaço de trabalho deve ter os seguintes recursos habilitados:

Exemplo de bloco de notas

Este bloco de anotações contém todo o código que você precisa para criar e implantar seu primeiro agente de IA. Importe o bloco de anotações para seu espaço de trabalho do Azure Databricks para executá-lo.

Demonstração do agente Mosaic AI

Obter caderno

Definir o agente

Um agente de IA consiste no seguinte:

  • Um modelo de linguagem grande (LLM) que pode raciocinar e tomar decisões
  • Ferramentas que o LLM pode usar para fazer mais do que apenas gerar texto, como executar código Python ou buscar dados

Execute o seguinte código em um bloco de anotações Databricks para definir um agente de chamada de ferramenta simples:

  1. Instale os pacotes Python necessários:

    %pip install -U -qqqq mlflow databricks-openai databricks-agents"
    dbutils.library.restartPython()
    
    • mlflow: Usado para desenvolvimento de agentes e rastreamento de agentes.
    • databricks-openai: Usado para se ligar ao LLM alojado no Databricks e aceder às ferramentas do Unity Catalog.
    • databricks-agents: Usado para empacotar e implantar o agente.
  2. Defina o agente. Este trecho de código faz o seguinte:

    • Conecta-se ao modelo Databricks que serve o endpoint usando o cliente OpenAI.
    • Habilita o rastreio do MLflow usando autolog(). Isso adiciona instrumentação para que você possa ver o que seu agente faz quando você envia uma consulta.
    • Adiciona a ferramenta system.ai.python_exec ao seu agente. Esta função integrada do Unity Catalog permite que seu agente execute código Python.
    • Utiliza funções auxiliares do MLflow (output_to_responses_items_stream, create_function_call_output_item) para converter saída de LLM em streaming para um formato compatível com Responses 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)}
    

Testar o agente

Teste o agente consultando-o com um prompt que requer a execução de código Python:

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

Além da saída do LLM, você verá informações detalhadas de rastreamento diretamente em seu notebook. Esses rastreamentos ajudam a depurar chamadas de agente lentas ou com falha. Esses rastreios foram adicionados automaticamente usando mlflow.openai.autolog().

Implantar o agente

Agora que tem um agente, pode empacotá-lo e implantá-lo num endpoint de serviço do Databricks. Comece a coletar feedback sobre um agente implantado compartilhando-o com outras pessoas e conversando com ele usando uma interface do usuário de bate-papo integrada.

Preparar o código do agente para implantação

Para preparar o código do agente para implantação, envolva-o usando a ResponsesAgent interface do MLflow. A ResponsesAgent interface é a maneira recomendada de empacotar agentes para implantação no Azure Databricks.

  1. Para implementar a ResponsesAgent interface, defina tanto o método predict_stream() (para respostas em streaming) como o método predict() (para pedidos não-streaming). Como a lógica do agente subjacente já gera eventos compatíveis com Respostas API, a implementação é simples:

    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. Adicione o seguinte código ao seu bloco de notas para testar a sua ResponsesAgent classe:

    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. Combine todo o código do agente em um único arquivo para que você possa registrá-lo e implantá-lo.

  • Consolide todo o código do agente em uma única célula do notebook.
  • Na parte superior da célula, adicione o %%writefile quickstart_agent.py comando magic para salvar seu agente em um arquivo.
  • Na parte inferior da célula, chame mlflow.models.set_model() com o seu objeto de agente. Isso informa ao MLflow qual objeto de agente usar ao servir previsões. Esta etapa configura efetivamente o ponto de entrada para o nosso código de agente.

A célula do bloco de notas deve ter o seguinte aspeto:

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

Registrar o agente

Inicie sessão com o seu agente e registe-o no Catálogo Unity. Isso empacota seu agente e suas dependências em um único artefato para implantação.

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
)

Implantar o agente

Implante seu agente registrado em um ponto de extremidade de serviço:

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
)

Depois que o endpoint do agente for iniciado, poderá conversar com ele usando o AI Playground ou partilhá-lo com as partes interessadas para obter feedback.

Próximos passos

Escolha para onde ir em seguida com base nos seus objetivos:

Meça e melhore a qualidade do seu agente: consulte Guia de início rápido da Avaliação do agente.

Crie agentes mais avançados: crie um agente que execute o RAG usando dados não estruturados, lide com conversas de várias voltas e use a Avaliação do Agente para medir a qualidade. Consulte Tutorial: Criar, avaliar e implantar um agente de recuperação.

Saiba como criar agentes usando outras estruturas: Saiba como criar agentes usando bibliotecas populares como LangGraph, Python puro e OpenAI. Consulte Agentes de IA de autor no código