Delen via

AI-agents ontwerpen in code

Op deze pagina ziet u hoe u een AI-agent in Python maakt met behulp van Mosaic AI Agent Framework en populaire bibliotheken voor het ontwerpen van agents, zoals LangGraph, PyFunc en OpenAI.

Eisen

Aanbeveling

Databricks raadt aan de nieuwste versie van de MLflow Python-client te installeren bij het ontwikkelen van agents.

Als u agents wilt ontwerpen en implementeren met behulp van de methode op deze pagina, installeert u het volgende:

  • databricks-agents 0.16.0 of hoger
  • mlflow 2.20.2 of hoger
  • Python 3.10 of hoger.
    • U kunt serverloze berekeningen of Databricks Runtime 13.3 LTS of hoger gebruiken om aan deze vereiste te voldoen.
%pip install -U -qqqq databricks-agents>=0.16.0 mlflow>=2.20.2

Databricks raadt ook aan om Databricks AI Bridge te installeren integratiepakketten bij het ontwerpen van agents. Deze integratiepakketten (zoals databricks-langchain, databricks-openai) bieden een gedeelde laag API's om te communiceren met Databricks AI-functies, zoals Databricks AI/BI Genie en Vector Search, in frameworks en SDK's voor agentcreatie.

LangChain/LangGraph

%pip install -U -qqqq databricks-langchain

OpenAI

%pip install -U -qqqq databricks-openai

Zuivere Python-agenten

%pip install -U -qqqq databricks-ai-bridge

ChatAgent gebruiken om agents te schrijven

Databricks raadt de MLflow-interface ChatAgent aan om agents op productieniveau te maken. Deze specificatie van het chatschema is vergelijkbaar met, maar is niet strikt compatibel met het OpenAI-schema ChatCompletion .

ChatAgent verpakt eenvoudig bestaande agents voor Databricks-compatibiliteit.

ChatAgent biedt de volgende voordelen:

  • Geavanceerde agentmogelijkheden

    • Ondersteuning voor meerdere agents
    • streaming-uitvoer: interactieve gebruikerservaringen inschakelen door uitvoer te streamen in kleinere segmenten.
    • Uitgebreide berichtgeschiedenis voor het aanroepen van hulpprogramma's: Meerdere berichten retourneren, inclusief tussenliggende berichten voor bellen via hulpprogramma's, voor verbeterde kwaliteit en gespreksbeheer.
    • Bevestigingsondersteuning bij oproep van tools

  • Gestroomlijnde ontwikkeling, implementatie en bewaking

    • Creëer uw agent met behulp van een framework: gebruik de ChatAgent interface om een bestaande agent te omhullen en te profiteren van de standaardcompatibiliteit met AI Playground, Agent Evaluation en Agent Monitoring.
    • Getypte ontwerpinterfaces: Agentcode schrijven met behulp van getypte Python-klassen en profiteren van automatisch aanvullen van IDE en notebook.
    • Automatische handtekeningdeductie: MLflow bepaalt automatisch ChatAgent handtekeningen bij het registreren van de agent, waardoor registratie en implementatie eenvoudiger worden. Zie Modelhandtekening vaststellen tijdens het loggen.
    • door AI Gateway verbeterde deductietabellen: AI Gateway-deductietabellen worden automatisch ingeschakeld voor geïmplementeerde agents en bieden toegang tot gedetailleerde metagegevens van aanvraaglogboeken.

Om te leren hoe je een ChatAgent maakt, zie de voorbeelden in de volgende sectie en de MLflow-documentatie - Wat is de ChatAgent-interface.

Wat gebeurt er als ik al een agent heb?

Als u al een agent hebt gebouwd met LangChain, LangGraph of een vergelijkbaar framework, hoeft u uw agent niet opnieuw te schrijven om deze te gebruiken in Databricks. In plaats daarvan kunt u uw bestaande agent verpakken met de MLflow-interface ChatAgent :

  1. Schrijf een Python-wrapperklasse die overkomt van mlflow.pyfunc.ChatAgent.

    Bewaar in de wrapper-klasse uw bestaande agent als kenmerk self.agent = your_existing_agent.

  2. Voor deChatAgent klasse moet u een predict methode implementeren om niet-streamingaanvragen te verwerken.

    predict moet het volgende accepteren:

    • messages: list[ChatAgentMessage], dat is een lijst met ChatAgentMessage elk met een rol (zoals 'gebruiker' of 'assistent'), de prompt en een id.

    • (Optioneel) context: Optional[ChatContext] en custom_inputs: Optional[dict] voor extra gegevens.

    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 moet een ChatAgentResponse teruggeven.

    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."
    )
  ]
)

  3. Omzetten tussen formaten

    Converteer de binnenkomende berichten van predict in list[ChatAgentMessage] naar de invoerindeling die uw agent verwacht.

    Nadat uw agent een antwoord heeft gegenereerd, converteert u de uitvoer naar een of meer ChatAgentMessage objecten en verpakt u deze in een ChatAgentResponse.

Aanbeveling

LangChain-uitvoer automatisch converteren

Als u een LangChain-agent verpakt, kunt u de uitvoer van LangChain met mlflow.langchain.output_parsers.ChatAgentOutputParser automatisch converteren naar het MLflow ChatAgentMessage en ChatAgentResponse schema.

Hier volgt een vereenvoudigde sjabloon voor het converteren van uw agent:

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

Zie de notebooks in de volgende sectie voor volledige voorbeelden.

voorbeelden van ChatAgent

De volgende notebooks laten zien hoe je streaming en niet-streaming ChatAgents kunt maken met behulp van de populaire bibliotheken OpenAI, LangGraph en AutoGen.

LangGraph

Als u een LangChain-agent verpakt, kunt u de uitvoer van LangChain met mlflow.langchain.output_parsers.ChatAgentOutputParser automatisch converteren naar het MLflow ChatAgentMessage en ChatAgentResponse schema.

LangGraph-agent voor het aanroepen van hulpprogramma's

Haal notitieblok op

OpenAI

OpenAI-agent voor het aanroepen van hulpprogramma's

Haal notitieblok op

OpenAI-antwoorden-API-toolaanroepagent

Haal notitieblok op

OpenAI-agent voor alleen-chat

Haal notitieblok op

AutoGen

AutoGen tool-oproepende agent

Haal notitieblok op

DSPy

DSPy-agent voor alleen-chat

Haal notitieblok op

Voor het uitbreiden van de mogelijkheden van deze agents door tools toe te voegen, zie AI agent tools.

Voorbeeld van meerdere agents

Zie Genie gebruiken in systemen met meerdere agentsvoor meer informatie over het maken van een multiagentsysteem met behulp van Genie.

Agents voor streaming-uitvoer

Streamingagents leveren antwoorden in een continue stroom van kleinere, incrementele delen. Streaming vermindert de waargenomen latentie en verbetert de gebruikerservaring voor gespreksagenten.

Als u een streaming ChatAgentwilt maken, definieert u een predict_stream methode die een generator retourneert die objecten oplevert ChatAgentChunk : elk ChatAgentChunk bevat een deel van het antwoord. Lees meer over het ideale streaminggedrag van ChatAgent in de MLflow-documenten.

In de volgende code ziet u een voorbeeld van de predict_stream-functie; zie voor volledige voorbeelden van streamingagents, ChatAgent-voorbeelden.

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"]
      )

Schrijf implementatieklare ChatAgents voor Databricks Model Serving

Databricks implementeert ChatAgents in een gedistribueerde omgeving op Databricks Model Serving, wat betekent dat tijdens een gesprek met meerdere beurten dezelfde serving-replica mogelijk niet alle aanvragen verwerkt. Houd rekening met de volgende implicaties voor het beheren van de agentstatus:

  • Vermijd lokale caching: bij het implementeren van een ChatAgent, moet er niet van worden uitgegaan dat dezelfde replica alle aanvragen in een gesprek met meerdere wendingen zal afhandelen. De interne status reconstrueren met behulp van een woordenlijst ChatAgentRequest schema voor elke beurt.

  • Thread-veilige status: Ontwerp de agentstatus zodat deze thread-veilig is en conflicten in omgevingen met meerdere threads worden voorkomen.

  • de status initialiseren in de predict functie: initialiseer de status telkens wanneer de predict functie wordt aangeroepen, niet tijdens ChatAgent initialisatie. Het opslaan van de status op ChatAgent niveau kan informatie lekken tussen gesprekken en conflicten veroorzaken omdat één ChatAgent replica aanvragen van meerdere gesprekken kan verwerken.

Aangepaste invoer en uitvoer

Voor sommige scenario's zijn mogelijk extra agentinvoer vereist, zoals client_type en session_id, of uitvoer zoals bronkoppelingen ophalen die niet moeten worden opgenomen in de chatgeschiedenis voor toekomstige interacties.

Voor deze scenario's biedt MLflow ChatAgent systeemeigen ondersteuning voor de velden custom_inputs en custom_outputs.

Waarschuwing

De agentevaluatie-app ondersteunt momenteel geen weergave van traceringen voor agents met extra invoervelden.

Zie de volgende notebooks voor meer informatie over het instellen van aangepaste invoer en uitvoer.

OpenAI + PyFunc aangepast schema-agent-notitieboek

Haal notitieblok op

Aangepast schema-agentnotitieboek van LangGraph

Haal notitieblok op

Bied custom_inputs in de AI-programmeeromgeving en agentbeoordelingsapp

Als uw agent aanvullende invoer accepteert met behulp van het veld custom_inputs, kunt u deze invoer handmatig opgeven in zowel de AI Playground als de agent-beoordelingsapp .

  1. Selecteer in de AI Playground of de Agent Review App het tandwielpictogram.

  2. Schakel custom_inputs in.

  3. Geef een JSON-object op dat overeenkomt met het gedefinieerde invoerschema van uw agent.

    Voer aangepaste invoer in de AI-speelplaats in.

Aangepaste retrieverschema's opgeven

AI-agents gebruiken vaak retrievers om ongestructureerde gegevens te zoeken en op te vragen uit vectorzoekindexen. Zie bijvoorbeeld hulpmiddelen voor het ophalen van retriever-hulpmiddelen bouwen en traceren voor ongestructureerde gegevens.

Traceer deze retrievers binnen uw agent met MLflow RETRIEVER-overspanningen om Databricks-productfuncties in te schakelen, waaronder:

  • Automatisch koppelingen weergeven naar opgehaalde brondocumenten in de gebruikersinterface van AI Playground
  • Het automatisch uitvoeren van ophaal- en beoordelingsprocessen voor onderbouwing en relevantie bij agentenevaluatie.

Notitie

Databricks raadt het gebruik van retriever-hulpprogramma's aan die worden geleverd door Databricks AI Bridge-pakketten, zoals databricks_langchain.VectorSearchRetrieverTool en databricks_openai.VectorSearchRetrieverTool, omdat ze al voldoen aan het MLflow-retrieverschema. Zie Hulpprogramma's voor vectorzoekopdrachten lokaal ontwikkelen met AI Bridge.

Als uw agent retriever-spans bevat met een aangepast schema, roept u mlflow.models.set_retriever_schema aan wanneer u uw agent in de code definieert. Hiermee worden de uitvoerkolommen van de retriever toegewezen aan de verwachte velden van 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"],
)

Notitie

De doc_uri kolom is vooral belangrijk bij het evalueren van de prestaties van de retriever. doc_uri is de belangrijkste identificatie voor documenten die door de retriever worden geretourneerd, zodat u ze kunt vergelijken met grondwaarheidsevaluatiesets. Zie Evaluatiesets (MLflow 2).

Parametriseer agentcode voor implementatie in verschillende omgevingen

U kunt agentcode parametrizeren om dezelfde agentcode opnieuw te gebruiken in verschillende omgevingen.

Parameters zijn sleutel-waardeparen die u definieert in een Python-woordenlijst of een .yaml-bestand.

Als u de code wilt configureren, maakt u een ModelConfig met behulp van een Python-woordenlijst of een .yaml -bestand. ModelConfig is een set sleutelwaardeparameters waarmee flexibel configuratiebeheer mogelijk is. U kunt bijvoorbeeld een woordenlijst gebruiken tijdens de ontwikkeling en deze vervolgens converteren naar een .yaml-bestand voor productie-implementatie en CI/CD.

Zie de ModelConfigvoor meer informatie over .

Hieronder ziet u een voorbeeld 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

In uw agentcode kunt u verwijzen naar een standaardconfiguratie (ontwikkeling) vanuit het .yaml bestand of woordenlijst:

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

Bij het registreren van de agent specificeert u vervolgens de model_config parameter naar log_model om een aangepaste set parameters op te geven die moeten worden gebruikt bij het laden van de geregistreerde agent. Zie MLflow-documentatie - ModelConfig.

Verspreiding van streamingfouten

Mosaic AI verspreidt eventuele fouten die zijn opgetreden tijdens het streamen met de laatste token onder code databricks_output.error. Het is aan de aanroepende client om deze fout correct te verwerken en weer te geven.

{
  "delta": …,
  "databricks_output": {
    "trace": {...},
    "error": {
      "error_code": BAD_REQUEST,
      "message": "TimeoutException: Tool XYZ failed to execute."
    }
  }
}

Volgende stappen