Vytváření agentů AI v kódu

V tomto článku se dozvíte, jak v Pythonu vytvořit agenta AI s využitím architektury agenta Mosaic AI a oblíbených knihoven pro vytváření agentů, jako jsou LangGraph, PyFunc a OpenAI.

Požadavky

Databricks při vývoji agentů doporučuje nainstalovat nejnovější verzi klienta Pythonu MLflow.

Pokud chcete vytvářet a nasazovat agenty pomocí přístupu v tomto článku, musíte mít následující minimální verze balíčků:

databricks-agents verze 0.16.0 a vyšší
mlflow verze 2.20.2 a vyšší
Python 3.10 nebo novější K splnění tohoto požadavku můžete použít bezserverové výpočetní prostředky nebo Databricks Runtime 13.3 LTS a vyšší.

%pip install -U -qqqq databricks-agents>=0.16.0 mlflow>=2.20.2

Databricks také při vytváření agentů doporučuje nainstalovat integrační balíčky mostu Databricks AI. Tyto integrační balíčky (například databricks-langchain, databricks-openai) poskytují sdílenou vrstvu rozhraní API pro interakci s funkcemi AI Databricks, jako jsou Databricks AI/BI Genie a Vector Search, napříč architekturami pro vytváření agentů a sadami SDK.

LangChain/LangGraph

%pip install -U -qqqq databricks-langchain

OpenAI

%pip install -U -qqqq databricks-openai

Agenti čistě v Pythonu

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

Použijte `ChatAgent` k vytváření agentů

Databricks doporučuje k vytváření agentů na produkční úrovni používat rozhraní ChatAgent MLflow. Tato specifikace schématu chatu je určená pro scénáře agentů a je podobná schématu OpenAI ChatCompletion, ale není přísně kompatibilní. ChatAgent také přidává funkcionalitu pro agenty s podporou více interakcí a volání nástrojů.

Vytváření vašeho agenta pomocí ChatAgent poskytuje následující výhody:

Pokročilé možnosti agenta
- Streamování výstupu: Povolte interaktivní uživatelské prostředí streamováním výstupu v menších blocích.
- Úplná historie volání nástrojů: Vrátí více zpráv, včetně mezilehlých zpráv volání nástrojů, pro lepší kvalitu a správu konverzace.
- podpora potvrzení aktivace nástroje
- Podpora systému s více agenty
Zjednodušený vývoj, nasazení a monitorování
- Integrace funkcí Databricks nezávislá na frameworku: Napište svého agenta v libovolném frameworku dle vašeho výběru a získejte okamžitou kompatibilitu s AI Playground, vyhodnocením agentů a monitorováním agentů.
- Rozhraní pro psaní s typy: Psaní kódu agenta pomocí typovaných tříd Pythonu, což vám umožní využít automatické dokončování v integrovaných vývojových prostředích (IDE) a poznámkových blocích.
- automatické odvozování podpisů: MLflow automaticky odvodí ChatAgent podpisy při protokolování agenta, zjednoduší registraci a nasazení. Viz Zjištění podpisu modelu při protokolování.
- tabulky odvození rozšířené brány AI: Pro nasazené agenty jsou automaticky povolené tabulky odvozování brány AI, které poskytují přístup k podrobným metadatům protokolu požadavků.

Informace o vytvoření ChatAgent najdete v příkladech v následující části a dokumentaci k MLflow – co je rozhraní ChatAgent.

příklady `ChatAgent`

Následující poznámkové bloky ukazují, jak vytvořit streamované a nesestreamované ChatAgents pomocí oblíbených knihoven OpenAI, LangGraph a AutoGen.

Informace o tom, jak rozšířit možnosti těchto agentů přidáním nástrojů, najdete v tématu nástroje agenta AI.

Příklad `ChatAgent` systému s více agenty

Informace o vytvoření systému s více agenty pomocí Genie najdete v tématu Použití Genie v systémech s více agenty.

Tvorba výstupu pro streamovací agenty

Streamovací agenti doručují odpovědi v souvislém proudu menších a postupných bloků. Výstup streamování umožňuje koncovým uživatelům číst výstup agenta při generování, což snižuje vnímanou latenci a zlepšuje celkové uživatelské prostředí pro konverzační agenty.

Chcete-li vytvořit streamovací ChatAgent, definujte predict_stream metodu, která vrací generátor produkující ChatAgentChunk objekty, přičemž každý obsahuje část odpovědi. Přečtěte si více o ideálním chování streamování ChatAgent v dokumentaci MLflow .

Následující kód ukazuje příklad funkce predict_stream, pro kompletní příklady streamovacích agentů viz příklady 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"]
      )

Vytváření modelů připravených k nasazení pro službu Databricks Model Serving

Databricks nasazuje ChatAgents v distribuovaném prostředí ve službě Databricks Model Serving, což znamená, že během vícefázového rozhovoru nemusí stejná replika služby zpracovávat všechny požadavky. Věnujte pozornost důsledkům, které mají vliv na správu stavu agenta.

Vyhněte se místnímu ukládání do mezipaměti: Když nasazujete ChatAgent, nepředpokládejte, že stejná replika bude zpracovávat všechny požadavky ve vícekrokovém dialogu. Rekonstruujte vnitřní stav pomocí schématu slovníku ChatAgentRequest pro každý krok.
vláknově bezpečný stav: Navrhněte stav agenta tak, aby byl vláknově bezpečný a bránil konfliktům v prostředí více vláken.
Inicializovat stav ve funkci predict: Inicializovat stav při každém zavolání predict funkce, ne během inicializace ChatAgent. Ukládání stavu na úrovni ChatAgent může způsobit únik informací mezi konverzacemi a způsobit konflikty, protože jedna replika ChatAgent může zpracovávat žádosti z více konverzací.

vlastní vstupy a výstupy

Některé scénáře mohou vyžadovat další vstupy agenta, jako jsou client_type a session_id, nebo výstupy, jako jsou odkazy na načtení zdroje, které by neměly být zahrnuty do historie chatu pro budoucí interakce.

V těchto scénářích ChatAgent MLflow nativně podporuje pole custom_inputs a custom_outputs.

Varování

Aplikace pro vyhodnocení agenta v současné době nepodporuje vykreslování tras pro agenty s dalšími vstupními poli.

V následujících příkladech se dozvíte, jak nastavit vlastní vstupy a výstupy pro agenty OpenAI/PyFunc a LangGraph.

OpenAI + PyFunc poznámkový blok pro agenta s vlastním schématem

Pořiďte si poznámkový blok

Poznámkový blok agenta vlastního schématu LangGraph

Pořiďte si poznámkový blok

Poskytnout `custom_inputs` v aplikaci AI Playground a ve aplikaci pro přezkum agentů

Pokud váš agent přijímá další vstupy pomocí pole custom_inputs, můžete tyto vstupy zadat ručně jak v aplikaci AI Playground, tak v aplikaci pro kontrolu agenta .

V aplikaci AI Playground nebo aplikaci Agent Review vyberte ikonu ozubeného kola .
Aktivujte custom_inputs.
Zadejte objekt JSON, který odpovídá definovanému schématu vstupu vašeho agenta.

Určení vlastních schémat retrieveru

Agenti umělé inteligence běžně používají retrievery k vyhledávání a dotazování nestrukturovaných dat z indexů vektorového vyhledávání. Například nástroje pro vyhledávání, viz Nestrukturované vyhledávací nástroje agenta AI.

Trasujte tyto retrievery ve vašem agentovi pomocí MLflow RETRIEVER spans pro aktivaci funkcí produktu Databricks, mezi které patří:

Automatické zobrazení odkazů na načtené zdrojové dokumenty v uživatelském rozhraní AI Playground
Automatické spouštění správnosti načítání a hodnocení relevance při vyhodnocení agenta

Poznámka

Databricks doporučuje používat nástroje pro načítání poskytované balíčky Databricks AI Bridge, jako jsou databricks_langchain.VectorSearchRetrieverTool a databricks_openai.VectorSearchRetrieverTool, protože již odpovídají schématu retrieveru MLflow. Viz Vyvíjejte místně nástroje pro vyhledávání vektorů s AI Bridge.

Pokud váš agent zahrnuje rozsahy retrieveru s vlastním schématem, zavolejte funkci mlflow.models.set_retriever_schema při definování agenta v kódu. To mapuje výstupní sloupce vašeho retrieveru na očekávaná pole 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"],
)

Poznámka

Sloupec doc_uri je zvlášť důležitý při vyhodnocování výkonu retrieveru. doc_uri je hlavním identifikátorem dokumentů vrácených retrieverem, což vám umožní porovnat je se sadami vyhodnocení základní pravdy. Podívejte se na vyhodnocovací sady.

parametrizovat kód agenta pro nasazení napříč prostředími

Kód agenta můžete parametrizovat tak, aby znovu používal stejný kód agenta v různých prostředích.

Parametry jsou páry klíč-hodnota, které definujete ve slovníku Pythonu nebo v souboru .yaml.

Pokud chcete kód nakonfigurovat, vytvořte ModelConfig pomocí slovníku Pythonu nebo souboru .yaml . ModelConfig je soubor parametrů ve formátu klíč-hodnota, který umožňuje flexibilní správu konfigurace. Můžete například použít slovník během vývoje a pak ho převést na soubor .yaml pro produkční nasazení a CI/CD.

Podrobnosti o ModelConfignajdete v dokumentaci MLflow.

Příklad ModelConfig je uvedený níže:

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

V kódu agenta můžete odkazovat na výchozí konfiguraci (vývoj) ze souboru .yaml nebo slovníku:

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

Poté při protokolování agenta zadejte hodnotu parametru model_config na log_model, abyste určili vlastní sadu parametrů k použití při načítání protokolovaného agenta. Viz dokumentaci k MLflow - ModelConfig.

Šíření chyb streamování

Mosaic AI přenáší všechny chyby, ke kterým dochází při streamování, přičemž poslední token je označen pod databricks_output.error. Je na volajícím klientovi, aby správně zvládl a zobrazil tuto chybu.

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

Ověřování jménem uživatele

Důležité

Tato funkce je v beta verzi.

Varování

I když pověřená autentizace uživatele je výkonným nástrojem pro vynucování zabezpečeného přístupu k citlivým datům, umožňuje uživatelům prostoru vytvářet agenty, kteří v Databricks jednají jménem jiných uživatelů. Proto je ve výchozím nastavení zakázaný během beta verze a musí ho povolit správce pracovního prostoru. Než tuto funkci povolíte, projděte si bezpečnostní aspekty ověřování jménem uživatele.

S ověřováním jménem uživatele můžou agenti nasazení prostřednictvím modelu Mosaic AI přistupovat k prostředkům Databricks pomocí identity koncového uživatele Databricks, který se na agenta dotazoval. To umožňuje přístup k citlivým informacím pro každého jednotlivého uživatele s detailním vynucováním řízení přístupu k datům v Unity Catalogu.

Ověřování jménem uživatele dále omezuje příchozí token uživatele prostřednictvím downscopingu a zajišťuje, aby byl token vystavený kódu agenta omezen pouze na přístup pouze ke konkrétním rozhraním API definovaným autorem agenta. To zlepšuje zabezpečení tím, že brání neoprávněným akcím a snižuje riziko zneužití tokenu.

Při vytváření agenta můžete dál používat existující sady SDK pro přístup k prostředkům Databricks, jako jsou indexy vektorového vyhledávání. Chcete-li jednat jménem uživatele a získat přístup k prostředkům, jsou zde dva kroky:

V kódu agenta aktualizujte volání sady SDK tak, aby označovala, že prostředky by měly být přístupné jménem koncového uživatele agenta.
V době protokolování agenta (před nasazením agenta) zadejte rozsahy rozhraní REST API koncového uživatele vyžadované vaším agentem. Další podrobnosti najdete v sekci Autentizace jménem uživatele

Kompletní příklad ověřování jménem uživatele najdete v příkladu od začátku do konce.

Následující zdroje jsou kompatibilní s ověřováním prostřednictvím jména uživatele pomocí agentů.

Prostředek Databricks	Kompatibilní klienti
Index vektorové vyhledávání	`databricks_langchain.VectorSearchRetrieverTool`, `databricks_openai.VectorSearchRetrieverTool` nebo `VectorSearchClient`
Koncový bod obsluhy modelu	`databricks.sdk.WorkspaceClient`
SQL Warehouse	`databricks.sdk.WorkspaceClient`
Spojení UC	`databricks.sdk.WorkspaceClient`
Tabulky UC a funkce UC	V současné době nepodporujeme přímé klienty pro přístup k tabulkám UC nebo funkcím UC s ověřováním jménem uživatele. Místo toho doporučujeme uživatelům používat Genie pro přístup ke strukturovaným datům s ověřováním jménem uživatele.
Genie Space	`databricks_langchain.GenieAgent` nebo `databricks_openai.GenieAgent`

Při inicializaci nástrojů jménem klientského uživatele můžete buď zabalit inicializaci nástroje do bloku try-except, nebo povolit, aby chyby byly uživatelům zpřístupněny. Díky zpracování chyb může agent stále reagovat na maximum, i když koncový uživatel nemá přístup ke všem potřebným nástrojům. Pokud se však rozhodnete, že během inicializace nástrojů nezpracujete chyby, agent vyvolá chybu, pokud uživatel nemá žádné požadované prostředky.

Nakonfigurujte sady SDK

Následující fragmenty kódu ukazují, jak nakonfigurovat přístup koncového uživatele k různým prostředkům Databricks pomocí různých sad SDK.

Nástroj pro vyhledávání vektorů

from databricks.sdk import WorkspaceClient
from databricks.sdk.credentials_provider import ModelServingUserCredentials
from databricks_langchain import VectorSearchRetrieverTool

# Configure a Databricks SDK WorkspaceClient to use on behalf of end
# user authentication
user_client = WorkspaceClient(credentials_strategy = ModelServingUserCredentials())

vector_search_tools = []
# Exclude exception handling if you want the agent to fail
# when users lack access to all required Databricks resources
try:
    tool = VectorSearchRetrieverTool(
       index_name="<index_name>",
       description="...",
       tool_name="...",
       workspace_client=user_client # Specify the user authenticated client
     )
     vector_search_tools.append(tool)
except Exception as e:
     _logger.debug("Skipping adding tool as user does not have permissions)

Klient vektorové vyhledávání

from databricks.vector_search.client import VectorSearchClient
from databricks.vector_search.utils import CredentialStrategy

# Configure a VectorSearch Client to use on behalf of end
# user authentication
user_authenticated_vsc = VectorSearchClient(credential_strategy=CredentialStrategy.MODEL_SERVING_USER_CREDENTIALS)
# Exclude exception handling if you want the agent to fail when
# users lack access to all required Databricks resources
try:
  vs_index = user_authenticated_vsc.get_index(endpoint_name="endpoint_name", index_name="index_name")
  ...
except Exception as e:
  _logger.debug("Skipping Vector Index because user does not have permissions)

Koncový bod obsluhy modelu

from databricks.sdk import WorkspaceClient
from databricks.sdk.credentials_provider import ModelServingUserCredentials

# Configure a Databricks SDK WorkspaceClient to use on behalf of end
# user authentication
user_client = WorkspaceClient(credentials_strategy=ModelServingUserCredentials())

# Exclude exception handling if you want the agent to fail
# when users lack access to all required Databricks resources
try:
    user_client.serving_endpoints.query("endpoint_name", input="")
except Exception as e:
    _logger.debug("Skipping Model Serving Endpoint due to no permissions")

Spojení UC

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import ExternalFunctionRequestHttpMethod

# Configure a Databricks SDK WorkspaceClient to use on behalf of end
# user authentication
user_client = WorkspaceClient(credentials_strategy=ModelServingUserCredentials())

user_client.serving_endpoints.http_request(
  conn="connection_name",
  method=ExternalFunctionRequestHttpMethod.POST,
  path="/api/v1/resource",
  json={"key": "value"},
  headers={"extra_header_key": "extra_header_value"},
)

Genie Spaces

from databricks.sdk import WorkspaceClient
from databricks.sdk.credentials_provider import ModelServingUserCredentials
from databricks_langchain.genie import GenieAgent

# Configure a Databricks SDK WorkspaceClient to use on behalf of end
# user authentication
user_client = WorkspaceClient(credentials_strategy=ModelServingUserCredentials())

genie_agent = GenieAgent(
      genie_space_id="<genie_space_id>",
      genie_agent_name="Genie",
      description="Genie_description",
      client=user_client, # Specify the user client here
  )

Inicializace agenta

ověřování jménem uživatele je kompatibilní s rozhraním ChatAgent. Při použití ověřování jménem uživatele se identita koncového uživatele pozná pouze v případě, že je váš nasazený agent dotazován, tj. v rámci funkcí predict a predict_stream rozhraní ChatAgent. V důsledku toho musíte z těchto metod provádět akce s přístupem uživatele k prostředkům jménem uživatele (např. seznam indexů vyhledávání vektorů, ke kterým má koncový uživatel přístup), a nikoliv v __init__ metodě implementace ChatAgent. Tím se zajistí izolace prostředků mezi vyvoláními.

from mlflow.pyfunc import ChatAgent


class LangGraphChatAgent(ChatAgent):
    def initialize_agent():
        user_client = WorkspaceClient(
            credentials_strategy=ModelServingUserCredentials()
        )
        system_authorized_client = WorkspaceClient()
        ### Use the clients above to access resources with either system or user authorization

    def predict(
        self,
        messages: list[ChatAgentMessage],
        context: Optional[ChatContext] = None,
        custom_inputs: Optional[dict[str, Any]] = None,
    ) -> ChatAgentResponse:
        agent = initialize_agent() # Initialize the Agent in Predict
        request = {"messages": self._convert_messages_to_dict(messages)}

        messages = []
        for event in self.agent.stream(request, stream_mode="updates"):
            for node_data in event.values():
                messages.extend(
                    ChatAgentMessage(**msg) for msg in node_data.get("messages", [])
                )
        return ChatAgentResponse(messages=messages)

Důležité informace o zabezpečení

Před povolením ověřování jménem uživatele pomocí agentů je potřeba zvážit několik bezpečnostních dopadů:

Přístup k citlivým prostředkům Databricks: Povolení ověřování jménem uživatele umožňuje agentům přístup k citlivým prostředkům Databricks. Přestože jsme implementovali obory rozhraní API, abychom omezili prostředky, ke kterým můžou vývojáři přistupovat, a zmírnit riziko zneužití tokenů, některá rizika stále zůstávají. Obor rozhraní API například serving.serving-endpoints uděluje agentům oprávnění ke spuštění koncového bodu obsluhy jménem uživatele. Samotný koncový bod obsluhy ale může mít přístup k dalším oborům rozhraní API, které původní agent nemá oprávnění používat.
Žádná podpora souhlasu koncového uživatele: V aktuální fázi beta verze nemůžou uživatelé agenta zobrazit rozsahy rozhraní REST API Databricks nebo s tím souhlasit. Uživatelé zodpovídají za to, že důvěřují uživatelům s oprávněními "Může spravovat" na obsluhovacím koncovém bodu, aby mohli provádět akce v Databricks jejich jménem.

Kompletní příklad

Následující poznámkový blok ukazuje, jak vytvořit agenta s využitím vektorového vyhledávání a autentizace jménem uživatele.

Ověřování jménem na základě vektorového vyhledávání

Pořiďte si poznámkový blok

Další kroky

Váš názor

Byla tato stránka užitečná?

Last updated on 2025-05-08

Vytváření agentů AI v kódu

Požadavky

LangChain/LangGraph

OpenAI

Agenti čistě v Pythonu

Použijte ChatAgent k vytváření agentů

příklady ChatAgent

Agent pro volání nástrojů LangGraph

Agent volání nástrojů OpenAI

Agent pro volání nástrojů AutoGen

Agent pro volání nástrojů API pro OpenAI Responses

Chatovací agent pouze pro OpenAI

DSPy agent pouze pro chat

Příklad ChatAgent systému s více agenty

Tvorba výstupu pro streamovací agenty

Vytváření modelů připravených k nasazení pro službu Databricks Model Serving

vlastní vstupy a výstupy

OpenAI + PyFunc poznámkový blok pro agenta s vlastním schématem

Poznámkový blok agenta vlastního schématu LangGraph

Poskytnout custom_inputs v aplikaci AI Playground a ve aplikaci pro přezkum agentů

Určení vlastních schémat retrieveru

parametrizovat kód agenta pro nasazení napříč prostředími

Šíření chyb streamování

Ověřování jménem uživatele

Nakonfigurujte sady SDK

Nástroj pro vyhledávání vektorů

Klient vektorové vyhledávání

Koncový bod obsluhy modelu

Spojení UC

Genie Spaces

Inicializace agenta

Důležité informace o zabezpečení

Kompletní příklad

Ověřování jménem na základě vektorového vyhledávání

Další kroky

Váš názor

Další materiály

Použijte `ChatAgent` k vytváření agentů

příklady `ChatAgent`

Příklad `ChatAgent` systému s více agenty

Poskytnout `custom_inputs` v aplikaci AI Playground a ve aplikaci pro přezkum agentů