Serwer agenta

Serwer agenta to biblioteka, która zamienia Twój kod agenta w usługę. Umieszcza pętlę agenta w serwerze HTTP, definiuje API, które klienci wywołują, aby uruchomić agenta, zarządza połączeniami klientów i określa, co się dzieje, gdy uruchomienie zostanie przerwane. Serwer agenta działa w środowisku uruchomieniowym agenta. Aby dowiedzieć się, jak warstwy się ze sobą łączą, zobacz Wdrażanie agentów w usłudze Azure Databricks.

Dla nowych agentów Databricks zaleca serwer agentów oferowany przez AgentKit, który opisano na tej stronie. Aby porównać to z serwerami agentów starszej generacji, zobacz Porównanie serwerów agentów w usłudze Azure Databricks.

Requirements

  • Python 3.10 i nowsze.
  • Pakiet databricks-agentbricks, który zawiera AgentKit. Projekty, które tworzysz za pomocą agentbricks init, deklarują to jako zależność.

AgentKit

AgentKit to biblioteka Agent Bricks do tworzenia agentów działających na Agent Runtime. Zapewnia serwer agenta obsługujący twojego agenta, a także klientów i pomocników do pamięci agenta, sesji agenta, śledzenia i uwierzytelniania. Projekty, które tworzysz za pomocą Agent Bricks CLI , domyślnie używają AgentKit.

Serwer agenta

AgentKit udostępnia swój serwer agentów poprzez klasę DurableAgentServer . Serwer agenta opakowuje pętlę agenta serwerem HTTP, który obsługuje interfejs API wywołania, śledzi każde uruchomienie i odzyskuje uruchomienia przerwane przez awarię lub restart. Zapewnia:

  • Jedno API dla każdego trybu żądań: wywołania synchroniczne, strumieniowe i w tle, plus ponowne połączenie strumienia, wszystko obsługiwane przez tego samego handlera.
  • Wywołania idempotentne: ID wywołania generowany przez klienta zapewnia, że ponowione żądanie nie uruchamia zduplikowanego przebiegu.
  • Sesje uporządkowane: Inwokacje w tej samej sesji są uruchamiane pojedynczo, w kolejności.
  • Trwały stan działania: Po wdrożeniu status działania, zdarzenia i wyniki przetrwają restarty pracowników.
  • Odzyskiwanie po awarii: Serwer wykrywa przerwane uruchomienia i rozpoczyna ponowną próbę.
  • Autoryzacja użytkownika żądania: Narzędzia mogą działać z uprawnieniami użytkownika, który wysłał żądanie.
  • Niestandardowe endpointy: Serwer agenta to aplikacja FastAPI, więc możesz dodawać własne ścieżki.

Co zawiera AgentKit

AgentKit jest biblioteką databricks_agentkit Python. Eksportuje następujące:

Export Opis
DurableAgentServer, InvocationContext Serwer agenta i kontekst przekazywany do Twoich programów obsługi wywołania i odzyskiwania.
AgentKitClient Klient pamięci agenta i magazynów sesji agenta. Tworzy i pobiera store’y oraz udostępnia pamięci i sesje store’ów jako obiekty Memory, SessionStore, SessionItem, Session, MemorySearchResult i MemoryStore.
configure_tracing, start_trace Ustaw śledzenie MLflow dla agenta i rozpocznij śledzenie wokół jednostki pracy.
workspace_client, workspace_headers Stwórz uwierzytelniony zestaw SDK Databricks WorkspaceClient lub pobierz nagłówki uwierzytelniające dla bezpośrednich wywołań HTTP ze środowiska agenta.
list_ai_gateway_model_services Wymień usługi modelowe, które agent może wywołać przez Unity Gateway.

Biblioteka zawiera także pomocniki frameworka w databricks_agentkit.langgraph i databricks_agentkit.openai, których generowane szablony używają do łączenia każdego frameworka z magazynem sesji. Informacje o interfejsach API pamięci i sesji zawiera sekcja Pamięć agenta oraz Sesje agenta.

Zarejestruj agenta

Gdy tworzysz projekt z agentbricks init, CLI robi to za ciebie. Wygenerowany runtime/main.py tworzy serwer i rejestruje handlery wywołania i odzyskiwania szablonu, więc edytuje się kod agenta tylko w agent/. Postępuj zgodnie z krokami w tej sekcji, aby użyć istniejącego agenta lub napisać własnego handlera.

Stwórz a DurableAgentServer i zarejestruj asynchroniczny handler wywołań za pomocą @app.invoke. Program obsługi otrzymuje input żądania oraz kontekst wywołania i zwraca wynik serializowalny do formatu JSON. Publikuj postęp jako zdarzenia z context.emit.

from databricks_agentkit import DurableAgentServer, InvocationContext

app = DurableAgentServer()


@app.invoke
async def invoke(input, context: InvocationContext) -> dict:
    await context.emit({"type": "status", "message": "Looking that up"})
    answer = await run_my_agent(input, session_id=context.session_id)
    return {"answer": answer}

Możesz zarejestrować tylko jeden program obsługi wywołania i serwer nie uruchamia się bez niego. Moduł obsługi obsługuje każdy tryb żądań: klient decyduje, czy czekać na wynik, przesyłać strumieniowo zdarzenia, czy uruchomić w tle.

Aby uruchomić serwer lokalnie, uruchom go za pomocą agentbricks dev. Projekty, które tworzysz agentbricks init, obejmują punkt wejściowy, który uruchamia serwer za pomocą Uvicorn, oraz app.yaml plik, który uruchamia ten sam punkt wejściowy po wdrożeniu.

Kontekst wywołania

Drugi argument handlera to InvocationContext:

Atrybut Opis
invocation_id ID, które klient przesłał dla tego wywołania.
session_id Sesja, do której należy wywołanie, lub None, jeśli klient nie wysłał sesji.
attempt Numer próby. Pierwsza próba to 1.
is_recovery True gdy moduł obsługi odzyskiwania wykonuje próbę zastąpienia.
emit(event) Przechowuje zdarzenie JSON, dostarcza je klientom streamingowym i zwraca pozycję zdarzenia w strumieniu.
request_auth Rozwiązywacz poświadczeń użytkownika żądania, gdy agent wymaga autoryzacji użytkownika żądania. W przeciwnym razie None.

API wywołań

DurableAgentServer obsługuje API wywołań pod adresem /api/invocations:

  • POST /api/invocations rozpoczyna wywołanie. Domyślnie żądanie czeka na wynik. Ustaw stream, aby otrzymywać zdarzenia jako Server-Sent Events (SSE), lub background, aby natychmiast zwrócić adres URL stanu.
  • GET /api/invocations/<id> zwraca status wywołania, a po jego zakończeniu także jego wyjście.
  • GET /api/invocations/<id>/events?after=<event-id> przesyła zapisane zdarzenia, dzięki czemu klient może ponownie połączyć się po zerwaniu połączenia.

Informacje o polach żądania, przykładach i formatach odpowiedzi można znaleźć w artykule Wysyłanie zapytań do agentów wdrożonych w Azure Databricks.

Idempotentność

Klienci wysyłają UUID id przy każdym wywołaniu. Serwer traktuje ID jako klucz idempotencji, zachowując jednocześnie rekord wywołania: ponowne wysłanie tego samego żądania zwraca istniejące wywołanie zamiast uruchamiać agenta ponownie. Ponowne użycie ID dla innego żądania powoduje błąd 409 .

Sessions

Klienci mogą wysyłać session_id, aby zgrupować wywołania w jedną rozmowę. Serwer przechowuje identyfikator sesji oddzielnie od input, przekazuje go twojemu handlerowi jako context.session_id i uruchamia wywołania współdzielące identyfikator sesji pojedynczo, w kolejności. Serwer nie wnioskuje sesji na podstawie identyfikatora wywołania ani danych wejściowych. Bez identyfikatora sesji wywołanie jest bezsesyjne.

Stan uruchomienia

DurableAgentServer przechowuje żądanie, status, sygnały heartbeat, zdarzenia i wynik każdego wywołania w Runtime Store.

  • Rozwój lokalny: agentbricks dev korzysta z magazynu Runtime Store w procesie. Interfejs API wywołań zachowuje się tak samo, ale stan uruchomienia jest tracony, gdy proces się zatrzyma, a serwer nie uruchamia ponownie przerwanej pracy.
  • Wdrożeni agenci: agentbricks deploy tworzy dedykowaną bazę danych dla magazynu Runtime Store każdego wdrożenia w projekcie Lakebase zarządzanym przez Azure Databricks i ponownie wykorzystuje ją podczas ponownego wdrożenia. Nie możesz używać własnego projektu Lakebase do Runtime Store i nie tworzysz go samodzielnie ani go nie powiązujesz. Wyniki i zdarzenia przetrwają restarty pracowników, a każda instancja agenta może obsłużyć żądania statusu i ponownego połączenia. agentbricks deployments delete usuwa Runtime Store wraz z wdrożeniem.

Runtime Store przechowuje stan wykonania serwera. Jest oddzielona od magazynów sesji i pamięci, których agent używa do przechowywania historii rozmów i pamięci długotrwałej.

Odzyskiwanie po awarii

Aby odzyskać przebiegi, których wykonanie przerwała awaria lub ponowne uruchomienie procesu roboczego, zarejestruj procedurę odzyskiwania za pomocą @app.recover. Gdy wdrożony serwer wykryje, że sygnały heartbeat danego uruchomienia ustały, rozpoczyna próbę zastąpienia na dostępnym workerze i wywołuje procedurę obsługi odzyskiwania z oryginalnymi danymi wejściowymi.

@app.recover
async def recover(input, context: InvocationContext) -> dict:
    # Resume from the agent's last checkpoint in the session store,
    # or replay the input if that's safe for your agent.
    return await resume_my_agent(input, session_id=context.session_id)

Jeśli nie zarejestrujesz programu obsługi odzyskiwania, automatyczne odzyskiwanie jest wyłączone, a serwer zgłasza ostrzeżenie po uruchomieniu.

Proces odzyskiwania przebiega następująco:

  • Kiedy rozpoczyna się odzyskiwanie: Każda uruchomiona próba wysyła sygnał kontrolny co kilka sekund. Jeśli sygnały heartbeat ustaną, na przykład dlatego, że proces roboczy ulega awarii, restartuje się lub zostaje wymieniony podczas ponownego wdrożenia, serwer wykrywa nieaktualne wykonanie w ciągu kilku sekund i uruchamia próbę zastąpienia.
  • Gdy recovery się nie uruchamia: Jeśli twój handler zgłosi wyjątek, wywołanie się nie udaje i serwer nie próbuje go powtórzyć. Odzyskiwanie obejmuje zatrzymanych workerów, a nie błędy w kodzie agenta.
  • Liczba prób: Serwer nie ogranicza liczby prób odzyskania. Każda próba wymiany zwiększa context.attempt o jeden. Aby zatrzymać się po pewnej liczbie prób, sprawdź context.attempt w procedurze obsługi odzyskiwania i zgłoś błąd.
  • Odzyskiwanie ręczne: Nie możesz wywołać odzyskiwania ręcznie. Ponowne wysłanie żądania z tym samym identyfikatorem wywołania zwraca istniejące wywołanie zamiast rozpoczynać nową próbę.

Recovery może uruchomić Twój kod agenta więcej niż raz dla tego samego wywołania. Przerwana próba mogła już wykonać wywołania do systemów zewnętrznych przed rozpoczęciem próby zastępczej, więc spraw, by te wywołania były idempotentne.

Zażądaj autoryzacji użytkownika

Domyślnie narzędzia Twojego agenta działają z uprawnieniami jednostki usługi aplikacji. Aby uruchomić narzędzie z uprawnieniami użytkownika, który wysłał żądanie, zadeklaruj autoryzację użytkownika w agent.toml:

  • Dla narzędzia zarządzanego ustaw auth = "user" we wpisie narzędzia. Polecenia agentbricks tools add dla serwerów MCP, sandboxów i agentów Genie zapisują auth = "user" domyślnie. Przekaż --auth app, aby zamiast tego użyć tożsamości aplikacji.

  • Dla narzędzia, które piszesz w kodzie, zadeklaruj wymaganie oraz zakresy API, których Agent Bricks nie potrafi wywnioskować:

    [auth.user]
required = true
additional_api_scopes = ["sql"]

Gdy agent wymaga autoryzacji użytkownika, DurableAgentServer odczytuje dane uwierzytelniające użytkownika z zaufanych nagłówków żądań aplikacji Databricks i przechowuje je w pamięci tylko na aktywną próbę. Runtime Store nie przechowuje poświadczenia. W swoim handlerze pobierz klienta workspace dla użytkownika z:context.request_auth

@app.invoke
async def invoke(input, context: InvocationContext) -> dict:
    user_client = context.request_auth.client_for("user")
    me = user_client.current_user.me()
    return {"answer": f"Hello, {me.user_name}"}

client_for("app") zwraca klienta, który korzysta z jednostki usługi aplikacji. Resolver zamyka się po zakończeniu próby, więc wywołuj go wewnątrz handlera zamiast przechowywać klienta. Gdy uruchamiasz agenta lokalnie z agentbricks dev, client_for("user") używa lokalnych danych uwierzytelniających.

Podczas wdrażania element agentbricks deploy żąda zakresów użytkownika aplikacji Databricks, których potrzebują twoje narzędzia. Aby dodać brakujące zakresy do istniejącej aplikacji, przekaż --allow-user-scope-update. Zobacz Konfigurowanie autoryzacji w aplikacji usługi Databricks.

Wywołania request-user wykorzystują te same API synchroniczne, strumieniowe, działające w tle i ponownego łączenia. Ponieważ serwer nie przechowuje danych uwierzytelniających użytkownika, nie może wznowić przerwanego żądania użytkownika. Próba wymiany kończy się błędem MCP_USER_AUTH_RECOVERY_UNSUPPORTED zanim twoje procedury obsługi zostaną uruchomione.

Dodaj niestandardowe endpointy

DurableAgentServer jest aplikacją FastAPI. Dodawaj trasy w interfejsie API wywołań tak samo, jak dodajesz je do każdej aplikacji FastAPI:

@app.get("/status")
async def status() -> dict:
    return {"ready": True}

Szablony framework

agentbricks init generuje dwa katalogi:

  • Plik agent/ zawiera Twój kod frameworka: model, prompty i narzędzia.
  • runtime/ zawiera adapter łączący framework z DurableAgentServer, oraz punkt wejściowy, który rejestruje handlery wywołania i odzyskiwania adaptera.

Adapter tłumaczy każde wywołanie na wywołanie do pętli agenta frameworka i tłumaczy wyjście frameworka na zdarzenia i wynik. Oba szablony rejestrują procedurę obsługi odzyskiwania. Szablon LangGraph wznawia się od ostatniego punktu kontrolnego w magazynie sesji, a szablon SDK OpenAI Agents odtwarza żądanie w tej samej sesji. Aby przenieść istniejącego agenta, dodaj adapter i punkt wejściowy DurableAgentServer, a następnie ustaw server = "agentbricks" w sekcji [agent] pliku agent.toml.

Porównaj serwery agentów na platformie Azure Databricks

Azure Databricks oferuje trzy serwery agentów: serwer agentów AgentKit oraz dwa starsze serwery agentów, których używają wcześniejsze szablony aplikacji.

Serwer agenta Pakiet API klienta Trwałe wykonywanie Używane przez
AgentKit (DurableAgentServer, zalecane) databricks_agentkit, w pakiecie databricks-agentbricks API wywołań w /api/invocations: synchroniczne, streamingowe i działające w tle, z ponownym połączeniem strumienia Runtime Store zapewniający agentbricks deploy oraz odzyskiwanie po awarii przez obsługę odzyskiwania Projekty, które tworzysz za pomocą CLI Agent Bricks
LongRunningAgentServer (starsza wersja) databricks_ai_bridge.long_running, w pakiecie databricks-ai-bridge[agent-server] OpenAI Responses API w /responses, z uruchomieniami w tle i wznowieniem strumienia Stan uruchomienia w bazie danych Lakebase, którą sam konfigurujesz. Po awarii nowa próba kontynuuje przebieg z dziennika zdarzeń przerwanej próby. i agent-openai-advancedagent-langgraph-advanced
MLflow AgentServer (starsza wersja) mlflow.genai.agent_server, w pakiecie mlflow OpenAI Responses API at /responses: uruchomienia synchroniczne i strumieniowe None Podstawowe szablony aplikacji, takie jak agent-openai-agents-sdk

LongRunningAgentServer rozszerza MLflow AgentServer, a oba obsługują agentów implementujących interfejs MLflow ResponsesAgent . Aby wdrożyć i utrzymać agenta, który korzysta z jednego z nich, zobacz Uruchamianie agentów w aplikacjach Databricks przy użyciu starszego serwera agenta. Aby wysłać zapytanie do agenta na dowolnym z tych serwerów, zobacz Wysyłanie zapytań do agentów wdrożonych w usłudze Azure Databricks.

Limitations

  • Nie możesz zmienić serwera agenta dla istniejącego wdrożenia. Aby przełączać się między DurableAgentServer a własnym serwerem, stwórz nowy projekt z opcją agentbricks init --server, której chcesz, i wdroż go pod nową nazwą.
  • Zmiana pola server w agent.toml nie przekształca istniejącego kodu serwera do postaci DurableAgentServer.
  • Autoryzacja użytkownika żądającego wymaga server = "agentbricks".

Dodatkowe zasoby

  • Ostatnia aktualizacja o