Nota
El acceso a esta página requiere autorización. Puede intentar iniciar sesión o cambiar directorios.
El acceso a esta página requiere autorización. Puede intentar cambiar los directorios.
En esta página se muestra cómo crear un agente de INTELIGENCIA ARTIFICIAL en Python mediante Mosaic AI Agent Framework y bibliotecas populares de creación de agentes como LangGraph, PyFunc y OpenAI.
Requisitos
Sugerencia
Databricks recomienda instalar la versión más reciente del cliente de Python de MLflow al desarrollar agentes.
Para crear e implementar agentes mediante el enfoque de esta página, instale lo siguiente:
databricks-agents
0.16.0 o superiormlflow
2.20.2 o superior- Python 3.10 o superior.
- Para cumplir este requisito, puede utilizar informática sin servidor o Databricks Runtime 13.3 LTS o superior.
%pip install -U -qqqq databricks-agents>=0.16.0 mlflow>=2.20.2
Databricks también recomienda instalar Databricks AI Bridge paquetes de integración al crear agentes. Estos paquetes de integración (como databricks-langchain
, databricks-openai
) proporcionan una capa compartida de API para interactuar con las características de inteligencia artificial de Databricks, como Databricks AI/BI Genie y Vector Search, entre marcos de creación de agentes y SDK.
LangChain/LangGraph
%pip install -U -qqqq databricks-langchain
OpenAI
%pip install -U -qqqq databricks-openai
Agentes puros de Python
%pip install -U -qqqq databricks-ai-bridge
Uso de ChatAgent
para crear agentes
Databricks recomienda la interfaz de MLflow ChatAgent
para crear agentes de nivel de producción. Esta especificación de esquema de chat es similar a, pero no es estrictamente compatible con, el esquema openAI ChatCompletion
.
ChatAgent
proporciona las siguientes ventajas:
Funcionalidades avanzadas del agente
- Compatibilidad con varios agentes
- Transmisión de salida: Permita experiencias de usuario interactivas transmitiendo la salida en trozos más pequeños.
- Historial completo de mensajes de llamada a herramientas: devuelve varios mensajes, incluidos los mensajes intermedios de llamada a herramientas, para mejorar la calidad y la administración de conversaciones.
- Compatibilidad con la confirmación de llamadas a herramientas
Desarrollo, implementación y supervisión simplificados
- Elabore su agente utilizando cualquier plataforma: encapsule cualquier agente existente mediante la interfaz
ChatAgent
para tener una compatibilidad de serie con AI Playground, Agent Evaluation y Agent Monitoring. - Interfaces de creación con tipo: escriba código de agente mediante clases de Python con tipo, que se benefician del IDE y la acción de autocompletar para los cuadernos.
- inferencia automática de firmas: MLflow deduce automáticamente
ChatAgent
firmas al registrar el agente, lo que simplifica el registro y la implementación. Consulte Deducción de la firma del modelo durante el registro. - tablas de inferencia mejoradas por AI Gateway: las tablas de inferencia de AI Gateway se habilitan automáticamente para los agentes implementados, lo que proporciona acceso a metadatos detallados del registro de solicitudes.
- Elabore su agente utilizando cualquier plataforma: encapsule cualquier agente existente mediante la interfaz
Para obtener información sobre cómo crear un ChatAgent
, consulte los ejemplos en la sección siguiente y la documentación de MLflow - ¿Qué es la interfaz de ChatAgent?
¿Qué ocurre si ya tengo un agente?
Si ya tiene un agente compilado con LangChain, LangGraph o un marco similar, no es necesario volver a escribir el agente para usarlo en Databricks. En su lugar, encapsule el agente existente con la interfaz MLflow ChatAgent
:
Escriba una clase contenedora de Python que herede de
mlflow.pyfunc.ChatAgent
.Dentro de la clase contenedora, mantenga el agente existente como atributo
self.agent = your_existing_agent
.La
ChatAgent
clase requiere que implemente unpredict
método para controlar las solicitudes que no son de streaming.predict
debe aceptar:messages: list[ChatAgentMessage]
, que es una lista deChatAgentMessage
, cada uno de los elementos con un rol (como "usuario" o "asistente"), la instrucción y un ID.(Opcional)
context: Optional[ChatContext]
ycustom_inputs: Optional[dict]
para datos adicionales.
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
debe devolver unChatAgentResponse
.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." ) ] )
Conversión entre formatos
En
predict
, convierta los mensajes entrantes delist[ChatAgentMessage]
al formato de entrada que espera su agente.Después de que el agente genere una respuesta, conviértalo en uno o varios objetos
ChatAgentMessage
y encapsúlelos en unChatAgentResponse
.
Sugerencia
Automatizar la conversión de la salida de LangChain
Si estás encapsulando un agente de LangChain, puedes usar mlflow.langchain.output_parsers.ChatAgentOutputParser
para convertir automáticamente los resultados de LangChain en el esquema de MLflow ChatAgentMessage
y ChatAgentResponse
.
Consulte la siguiente plantilla simplificada para convertir el agente:
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())))
Para obtener ejemplos completos, consulte los cuadernos de la sección siguiente.
ejemplos de ChatAgent
En los cuadernos siguientes se muestra cómo crear ChatAgents
con flujos o sin flujos mediante las bibliotecas populares de OpenAI, LangGraph y AutoGen.
LangGraph
Si estás encapsulando un agente de LangChain, puedes usar mlflow.langchain.output_parsers.ChatAgentOutputParser
para convertir automáticamente los resultados de LangChain en el esquema de MLflow ChatAgentMessage
y ChatAgentResponse
.
Agente de invocación de herramientas de LangGraph
OpenAI
Agente de llamada a herramientas de OpenAI
Agente de llamada a herramientas de La API de respuestas de OpenAI
Agente de solo chat de OpenAI
AutoGen
Agente de invocación de herramientas de AutoGen
DSPy
Agente de solo chat de DSPy
Para obtener información sobre cómo expandir las funcionalidades de estos agentes mediante la adición de herramientas, consulte herramientas del agente de IA.
Ejemplo de varios agentes
Para obtener información sobre cómo crear un sistema multiagente mediante Genie, consulte Uso de Genie en sistemas multiagente.
Agentes de salida con flujos
Los agentes de streaming proporcionan respuestas en un flujo continuo de fragmentos incrementales más pequeños. El streaming reduce la latencia percibida y mejora la experiencia del usuario para los agentes conversacionales.
Para crear un streaming ChatAgent
, defina un predict_stream
método que devuelva un generador que produce ChatAgentChunk
objetos, cada uno ChatAgentChunk
contiene una parte de la respuesta. Obtenga más información sobre el comportamiento de streaming ideal ChatAgent
en los documentos de MLflow.
En el código siguiente se muestra una función de ejemplo predict_stream
, para ver ejemplos completos de agentes de streaming, vea Ejemplos de 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"]
)
Crear ChatAgent
listos para la implementación para el servicio de modelos de Databricks
Databricks implementa ChatAgent
en un entorno distribuido en Databricks Model Serving, lo que significa que, en una conversación de varios turnos, es posible que no todas las solicitudes sean gestionadas por la misma réplica de servicio. Preste atención a las siguientes implicaciones para administrar el estado del agente:
Evitar el almacenamiento en caché local: al implementar un
ChatAgent
, no supongamos que la misma réplica controlará todas las solicitudes en una conversación multiturno. Reconstruya el estado interno mediante un esquemaChatAgentRequest
de diccionario para cada turno.estado seguro para subprocesos: diseñe el estado del agente para que sea seguro para subprocesos, lo que impide conflictos en entornos multiproceso.
Inicializar el estado en la función
predict
: Inicializar el estado cada vez que se llama a la funciónpredict
, no durante la inicialización deChatAgent
. Almacenar el estado en el nivelChatAgent
podría filtrar información entre conversaciones y provocar conflictos porque una sola réplicaChatAgent
podría controlar las solicitudes de varias conversaciones.
entradas y salidas personalizadas
Algunos escenarios pueden requerir entradas adicionales de agente, como client_type
y session_id
, o salidas como enlaces a fuentes de recuperación que no deben incluirse en el historial de chat para futuras interacciones.
En estos escenarios, MLflow ChatAgent
admite de forma nativa los campos custom_inputs
y custom_outputs
.
Advertencia
La aplicación de revisión de la evaluación del agente actualmente no admite la representación de trazados para agentes con campos de entrada adicionales.
Consulte los cuadernos siguientes para obtener información sobre cómo establecer entradas y salidas personalizadas.
OpenAI + Cuaderno del agente de esquema personalizado de PyFunc
Cuaderno del agente de esquema personalizado de LangGraph
Proporcione custom_inputs
en AI Playground y la aplicación de revisión de agentes
Si su agente acepta entradas adicionales mediante el campo custom_inputs
, puede proporcionar manualmente estas entradas tanto en AI Playground como en la aplicación de revisión del agente.
En el AI Playground o la aplicación Agent Review, seleccione el icono de engranaje
Habilite custom_inputs.
Proporcione un objeto JSON que coincida con el esquema de entrada definido del agente.
Especificar esquemas personalizados del recuperador
Los agentes de inteligencia artificial suelen usar recuperadores para buscar y consultar datos no estructurados de índices de búsqueda vectorial. Para obtener herramientas recuperadoras de ejemplo, consulte Herramientas para construir y rastrear recuperadores para datos no estructurados.
Realice un seguimiento de estos recuperadores dentro del agente con intervalos de RECUPERACIÓN de MLflow para habilitar las características del producto de Databricks, entre las que se incluyen:
- Mostrar automáticamente vínculos a documentos de origen recuperados en la interfaz de usuario de AI Playground
- Ejecución automática de la recuperación del fundamento y evaluadores de relevancia en la evaluación del agente
Nota
Databricks recomienda usar las herramientas del recuperador proporcionadas por paquetes de Databricks AI Bridge como databricks_langchain.VectorSearchRetrieverTool
y databricks_openai.VectorSearchRetrieverTool
porque ya se ajustan al esquema del recuperador de MLflow. Consulte Desarrollo local de herramientas de recuperación de vectores de búsqueda con AI Bridge.
Si el agente incluye intervalos del recuperador con un esquema personalizado, llame a mlflow.models.set_retriever_schema al definir el agente en el código. Esto asigna las columnas de salida del recuperador a los campos esperados de 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"],
)
Nota
La columna doc_uri
es especialmente importante al evaluar el rendimiento del recuperador. doc_uri
es el identificador principal de los documentos devueltos por el recuperador, lo que le permite compararlos con los conjuntos de evaluación de la verdad fundamental. Consulte Conjuntos de evaluación (MLflow 2).
Parametrizar el código de agente para la implementación entre entornos
Puede parametrizar el código del agente para reutilizar el mismo código de agente en distintos entornos.
Los parámetros son pares clave-valor que se definen en un diccionario de Python o en un archivo .yaml
.
Para configurar el código, cree un ModelConfig
mediante un diccionario de Python o un archivo .yaml
. ModelConfig
es un conjunto de parámetros clave-valor que permite la administración flexible de la configuración. Por ejemplo, puede usar un diccionario durante el desarrollo y, a continuación, convertirlo en un archivo .yaml
para la implementación de producción y CI/CD.
Para obtener más información sobre ModelConfig
, consulte la documentación de MLflow.
A continuación se muestra un ejemplo 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
En el código del agente, puede hacer referencia a una configuración predeterminada (desarrollo) desde el archivo o diccionario .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')
A continuación, al registrar el agente, especifique el parámetro model_config
estableciéndolo en log_model
para definir un conjunto personalizado de parámetros que se usará al cargar el agente registrado. Consulte la Documentación de MLflow: ModelConfig.
Propagación de errores de streaming
Propagación de errores de Mosaic AI encontrados durante el streaming con el último token bajo databricks_output.error
. Depende del cliente que realiza la llamada manejar y mostrar adecuadamente este error.
{
"delta": …,
"databricks_output": {
"trace": {...},
"error": {
"error_code": BAD_REQUEST,
"message": "TimeoutException: Tool XYZ failed to execute."
}
}
}