Nota
O acesso a esta página requer autorização. Pode tentar iniciar sessão ou alterar os diretórios.
O acesso a esta página requer autorização. Pode tentar alterar os diretórios.
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:
- Catálogo Unity
- Estrutura do Mosaic AI Agent
- Modelos de base (pagamento por token, taxa de transferência provisionada ou modelos externos). Ver Funcionalidades com disponibilidade regional limitada
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
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:
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.
-
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_execao 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.
Para implementar a
ResponsesAgentinterface, defina tanto o métodopredict_stream()(para respostas em streaming) como o métodopredict()(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)Adicione o seguinte código ao seu bloco de notas para testar a sua
ResponsesAgentclasse: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)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.pycomando 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