Compartilhar 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.
  • Implante seu agente em um ponto de extremidade de serviço de modelo do Databricks.

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

Requisitos

Seu workspace deve ter os seguintes recursos habilitados:

Notebook de exemplo

Este notebook contém todo o código necessário para criar e implantar seu primeiro agente de IA. Importe o notebook para o workspace do Azure Databricks para execução.

Demonstração do agente de IA do Mosaico

Obter laptop

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 notebook do Databricks para definir um agente de chamada de ferramenta simples:

  1. Instale os pacotes necessários do Python:

    %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 conectar ao LLM hospedado pelo Databricks e acessar as ferramentas do Unity Catalog.
    • databricks-agents: utilizado para empacotar e desplegar o agente.
  2. Defina o agente. Este snippet de código faz o seguinte:

    • Conecta-se ao endpoint de serviço do modelo Databricks usando o cliente OpenAI.
    • Habilita o rastreamento do MLflow usando autolog(). Isso adiciona instrumentação para que você possa ver o que seu agente faz ao enviar uma consulta.
    • Adiciona a ferramenta system.ai.python_exec ao agente. Essa função interna do Catálogo do Unity permite que seu agente execute o código do Python.
    • Usa funções auxiliares do MLflow (output_to_responses_items_stream, create_function_call_output_item) para converter a saída de streaming do LLM em um formato compatível com Respostas da 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 exige a execução do código Python:

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

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

Implantar o agente

Agora que você tem um agente, pode empacotar e implantá-lo em um endpoint de serviço do Databricks. Comece a coletar comentários em um agente implantado compartilhando-o com outras pessoas e conversando com ele usando uma interface do usuário de chat interna.

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

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

  1. Para implementar a ResponsesAgent interface, defina os predict_stream() métodos (para respostas de streaming) e predict() (para solicitações que não são de streaming). Como a lógica do agente subjacente já gera eventos compatíveis com a API de Respostas, 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 bloco de anotações para testar 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 célula de notebook.
  • No topo da célula, adicione o comando mágico %%writefile quickstart_agent.py para salvar o agente em um arquivo.
  • Na parte inferior da célula, chame mlflow.models.set_model() com o objeto do agente. Isso informa ao MLflow qual objeto de agente usar ao servir previsões. Esta etapa configura o ponto de entrada do nosso código de agente de forma eficaz.

A célula do bloco de anotações deve ter a seguinte aparência:

%%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 em log

Faça login do seu agente e registre-o no Catálogo do 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 endpoint 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 ponto de extremidade do agente for iniciado, você poderá interagir com ele usando o AI Playground ou compartilhá-lo com os stakeholders para obter feedback.

Próximas etapas

Escolha para onde ir em seguida com base em suas metas:

Medir e melhorar a qualidade do agente: confira o início rápido da Avaliação do Agente.

Crie agentes mais avançados: crie um agente que executa o RAG usando dados não estruturados, lida com conversas de vários turnos e usa a Avaliação do Agente para medir a qualidade. Confira o Tutorial: Compilar, 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 do Autor no código