Programar agentes de IA en código

2025-06-12

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 superior
mlflow 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 ajusta fácilmente los agentes existentes para la compatibilidad de Databricks.

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.

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.
LaChatAgent clase requiere que implemente un predict método para controlar las solicitudes que no son de streaming.

predict debe aceptar:
- messages: list[ChatAgentMessage], que es una lista de ChatAgentMessage, cada uno de los elementos con un rol (como "usuario" o "asistente"), la instrucción y un ID.
- (Opcional) context: Optional[ChatContext] y custom_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 un ChatAgentResponse.
```
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 de list[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 un ChatAgentResponse.

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

Agente de invocación de herramientas de LangGraph

Obtener el cuaderno

OpenAI

AutoGen

Agente de invocación de herramientas de AutoGen

Obtener el cuaderno

DSPy

Agente de solo chat de DSPy

Obtener el cuaderno

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 ChatAgenten 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 esquema ChatAgentRequest 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ón predict, no durante la inicialización de ChatAgent. Almacenar el estado en el nivel ChatAgent podría filtrar información entre conversaciones y provocar conflictos porque una sola réplica ChatAgent 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

Obtener el cuaderno

Cuaderno del agente de esquema personalizado de LangGraph

Obtener el cuaderno

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."
    }
  }
}

Compartir a través de

Programar agentes de IA en código

Requisitos

LangChain/LangGraph

OpenAI

Agentes puros de Python

Uso de ChatAgent para crear agentes

¿Qué ocurre si ya tengo un agente?

ejemplos de ChatAgent

LangGraph

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

Ejemplo de varios agentes

Agentes de salida con flujos

Crear ChatAgent listos para la implementación para el servicio de modelos de Databricks

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

Especificar esquemas personalizados del recuperador

Parametrizar el código de agente para la implementación entre entornos

Propagación de errores de streaming

Pasos siguientes

Comentarios

Recursos adicionales

Uso de `ChatAgent` para crear agentes

ejemplos de `ChatAgent`

Crear `ChatAgent` listos para la implementación para el servicio de modelos de Databricks

Proporcione `custom_inputs` en AI Playground y la aplicación de revisión de agentes