Hostowani agenci programu Foundry

Hostowani agenci w usłudze agenta Microsoft Foundry umożliwiają wdrażanie konteneryzowanych aplikacji agentów w infrastrukturze zarządzanej Microsoft. Platforma obsługuje skalowanie, trwałość stanu sesji, zabezpieczenia i zarządzanie cyklem życia, dzięki czemu można skoncentrować się na logice agenta. Hostowani agenci Microsoft Foundry są już ogólnie dostępni i obsługują agentów zbudowanych przy użyciu własnego kodu lub preferowanego frameworku agentów. W tym artykule opisano w szczególności integrację hostingu platformy Agent Framework.

Korzystając z integracji hostingu platformy Agent Framework, można uwidocznić Agent za pośrednictwem protokołu Foundry Responses lub Invocations z minimalnym kodem. Python obsługuje również hostowanie natywnego Workflow bezpośrednio, bez konwertowania go na agenta.

Uwaga / Notatka

Można również wdrożyć kod agenta utworzony przy użyciu innych frameworków w hostowanych agentach Foundry, korzystając z przepływów pracy Azure Developer CLI (azd). Aby uzyskać niezależne od platformy pojęcia i wskazówki dotyczące wdrażania, zobacz Co to są hostowani agenci? Pozostała część tego artykułu koncentruje się na integracji z platformą Agent Framework.

Kiedy używać hostowanych agentów

Wybierz hostowane przez Foundry agenty, jeśli chcesz:

  • Infrastruktura zarządzana — nie trzeba samodzielnie konfigurować kontenerów, serwerów internetowych ani reguł skalowania.
  • Wbudowane zarządzanie sesjami — platforma utrwala $HOME i przesyła pliki przez kolejki i okresy bezczynności.
  • Tożsamość dedykowanego agenta — każdy wdrożony agent uzyskuje własną tożsamość Entra na potrzeby bezpiecznego dostępu do modeli, narzędzi i usług podrzędnych.
  • Punkty końcowe zgodne z interfejsem OpenAI — klienci mogą korzystać z agenta przy użyciu dowolnego zestawu SDK zgodnego z interfejsem OpenAI za pośrednictwem protokołu Odpowiedzi.
  • W przypadku agentów audio w czasie rzeczywistym należy używać hostowanych agentów z Azure speech in Foundry Tools (Voice Live) na potrzeby wykrywania aktywności głosowej po stronie serwera, anulowania echa i redukcji szumu. Aby uzyskać szczegółowe informacje, zobacz Używanie usługi Voice Live z hostowanymi agentami.

Uwaga / Notatka

Integracja języka Python agent-framework-foundry-hosting jest w wersji przedpremierowej. Agenci hostowani Microsoft Foundry, usługa zarządzanego hostingu, są już ogólnie dostępni.

Wymagania wstępne

  • Subskrypcja platformy Azure
  • Azure Developer CLI (azd) z rozszerzeniem agenta sztucznej inteligencji:

Do testowania lokalnego potrzebne są również następujące elementy:

Zainstaluj pakiet NuGet hostingu:

dotnet add package Microsoft.Agents.AI.Foundry.Hosting --prerelease
  • Python 3.10 lub nowszy

Zainstaluj pakiet hostingu wersji wstępnej, klienta programu Foundry i pakiet uwierzytelniania Azure:

pip install --pre agent-framework-foundry agent-framework-foundry-hosting azure-identity

W Foundry platforma dostarcza kontekst użytkownika podmiotu wywołującego oraz kontekst wywołania; infrastruktura hostująca używa ich do izolowania stanu dla poszczególnych użytkowników i przekazywania kontekstu żądania do usług Foundry. W przypadku uruchomień lokalnych ten kontekst platformy nie jest dostępny, dlatego aplikacje muszą w razie potrzeby zapewnić własne mechanizmy tożsamości i zarządzania stanem.

Protokół odpowiedzi

Protokół Responses jest zalecanym punktem wyjścia dla większości agentów. Udostępnia punkt końcowy zgodny z interfejsem OpenAI /responses, a platforma automatycznie zarządza historią konwersacji, przesyłaniem strumieniowym oraz cyklem życia sesji.

W przypadku agentów hostowanych w Pythonie odpowiedź, która kończy się przedwcześnie, ma stan incomplete. Klienci strumieniowi otrzymują końcowe zdarzenie response.incomplete, podczas gdy klienci niestrumieniowi otrzymują status ustawione na incomplete. Przyczyna zakończenia content_filter odpowiada ustawieniu incomplete_details.reason na content_filter i length odpowiada max_output_tokens. Wszystkie wygenerowane dane wyjściowe lub zawartość odmowy pozostają dostępne w odpowiedzi.

using Azure.AI.AgentServer.Core;
using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Foundry.Hosting;

var projectEndpoint = new Uri(Environment.GetEnvironmentVariable("FOUNDRY_PROJECT_ENDPOINT")
    ?? throw new InvalidOperationException("FOUNDRY_PROJECT_ENDPOINT is not set."));
var deployment = Environment.GetEnvironmentVariable("FOUNDRY_MODEL")
    ?? Environment.GetEnvironmentVariable("AZURE_AI_MODEL_DEPLOYMENT_NAME")
    ?? "gpt-4o";

AIAgent agent = new AIProjectClient(projectEndpoint, new DefaultAzureCredential())
    .AsAIAgent(
        model: deployment,
        instructions: "You are a helpful AI assistant.",
        name: "my-agent");

var builder = AgentHost.CreateBuilder(args);
builder.Services.AddFoundryResponses(agent);
builder.RegisterProtocol("responses", endpoints => endpoints.MapFoundryResponses());

var app = builder.Build();
app.Run();

Obiekt AgentHost.CreateBuilder tworzy wstępnie skonfigurowany serwer aplikacji w środowisku hostingu Foundry. AddFoundryResponses rejestruje agenta w procedurze obsługi protokołu Responses i MapFoundryResponses mapuje /responses punkt końcowy HTTP.

import os

from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from agent_framework_foundry_hosting import ResponsesHostServer
from azure.identity import DefaultAzureCredential

client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ.get("FOUNDRY_MODEL") or os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
    credential=DefaultAzureCredential(),
)

agent = Agent(
    client=client,
    instructions="You are a helpful AI assistant.",
)

server = ResponsesHostServer(agent)
server.run()

Element ResponsesHostServer opakowuje instancję agenta i uwidacznia ją za pośrednictwem protokołu Foundry Responses. Pole store obiektu wywołującego określa, czy zapisywane są odpowiedź zewnętrzna, sesja zarządzana przez hosta oraz stan zatwierdzenia. To history_source ustawienie niezależnie wybiera, kto dostarcza historię modelu:

history_source Zachowanie historii modelu
"agent_server" (ustawienie domyślne) Host rekonstruuje przechowywaną zewnętrzną transkrypcję odpowiedzi i wyłącza magazyn usług podrzędnych, aby zapobiec zduplikowaniu historii.
"service" Host wysyła tylko bieżące dane wejściowe i zapisuje we własnym zakresie identyfikator kontynuacji usługi przechowywania modelu. Zapisana konwersacja z dostawcą nie może rozgałęziać się od wcześniejszej odpowiedzi.
"agent" Host wysyła tylko bieżące dane wejściowe. Domyślne ustawienia magazynu agenta HistoryProvider lub usługi podrzędnej zarządzają historią.

Nie należy łączyć "agent_server" ani "service" z HistoryProvider z włączonym obciążeniem. Tryb domyślny odrzuca również stałe opcje kontynuacji podrzędnej, takie jak conversation_id, previous_response_idi conversation. Użyj history_source="agent" do niestandardowej implementacji SupportsAgentRun.

response_store Parametr konstruktora wybiera zaplecze dla trwałości odpowiedzi zewnętrznych. Starszy parametr konstruktora store jest przestarzałym aliasem parametru response_store; żaden z tych parametrów nie ustawia pola store wywołującego dla poszczególnych żądań. Żądanie z elementem store=false jest jednorazowe: nie zapisuje stanu zarządzanego przez hosta, wyłącza obsługiwaną pamięć podrzędną i nie może używać background=true.

Host jest właścicielem dostarczonego agenta i może dodać dostawców kontekstu specyficznych dla hostingu. Nie należy ponownie używać agenta z innym hostem ani wywoływać go bezpośrednio po zakończeniu budowy hosta.

Host Responses zachowuje natywne wywołania systemowe, zrzuty ekranu i mechanizmy kontroli bezpieczeństwa. Aplikacja musi wykonać żądane akcje i jawnie potwierdzić wszelkie kontrole bezpieczeństwa. Aby uzyskać pełny przepływ, zobacz Używanie komputera natywnego.

Wybierz instancję agenta lub fabrykę

Zarówno ResponsesHostServer, jak i InvocationsHostServer akceptują za pośrednictwem parametru agent albo instancję agenta, albo synchroniczny lub asynchroniczny obiekt wywoływalny niewymagający argumentów. Host używa ponownie instancji przez cały czas swojego istnienia. Wywołanie jest uruchamiane raz na żądanie, a zwrócony agent należy do tego żądania.

Użyj obiektu wywoływalnego, gdy zwykły agent zachowuje modyfikowalny stan poza AgentSession. Hosty utrwalają tylko obsługiwane magazyny sesji, punktów kontrolnych i zatwierdzania funkcji, a nie dowolne pola dla agenta o zakresie żądania.

Warning

Hostowanie elementu WorkflowAgent, takiego jak workflow.as_agent(), za pośrednictwem agent= jest przestarzałe. Element WorkflowAgent zachowuje stan przepływu pracy w pamięci między przebiegami, więc jedno wystąpienie nigdy nie może obsługiwać żądań od różnych użytkowników lub konwersacji. Hostowanie przepływu pracy natywnie za pośrednictwem workflow= i fabryki obsługującej żądania.

Dopóki nie zmigrujesz hosta Responses, fabryka, która tworzy nowy przepływ pracy, funkcje wykonawcze, opakowanych agentów, klientów, dostawców i narzędzi dla każdego żądania, jest bezpieczną starszą formą. Zachowaj stabilną nazwę przepływu pracy i identyfikatory wykonawców, aby host mógł przywrócić punkty kontrolne. W przypadku hosta wywołań ta starsza wersja fabryki odpowiada tylko bezstanowym, jednoetapowym przepływom pracy, ponieważ hosting agenta nie przywraca punktów kontrolnych przepływu pracy.

Używaj również fabryki, gdy integracja niesie ze sobą tożsamość żądania lub jest właścicielem zasobów specyficznych dla żądania. Na przykład twórz połączenia MCP, Toolboxy, dostawców umiejętności, klientów wyszukiwania, dostawców pamięci oraz ich poświadczenia w fabryce, jeśli korzystają z bieżącego wywołania platformy lub kontekstu użytkownika. Ponowne użycie połączenia MCP obejmującego cały proces może zachować tożsamość żądania, które go otworzyło.

Host wprowadza i zamyka agentów utworzonych przez fabrykę dla każdego żądania. Agent zarządza klientami zarządzanymi kontekstowo i narzędziami MCP, ale twoja fabryka musi zamknąć dowolnego innego dostawcę, transport lub poświadczenia, które tworzy. Nie zamykaj współdzielonych obiektów, które aplikacja dostarczyła spoza fabryki.

Uruchom natywny przepływ pracy z Responses

Python może hostować wbudowany przepływ pracy bezpośrednio za pośrednictwem usługi workflow=. Natywny przepływ pracy wymaga callbacku parse_response, który mapuje bieżące żądanie Responses na typowane dane wejściowe początkowe lub pełną partię oczekujących odpowiedzi:

from pydantic import BaseModel

from agent_framework_foundry_hosting import (
    CheckpointStoreProvider,
    HostedResponseRequest,
    ResponsesHostServer,
    WorkflowTurn,
)


class Ticket(BaseModel):
    text: str


def build_workflow(request: HostedResponseRequest):
    return build_fresh_workflow()


async def parse_response(request: HostedResponseRequest) -> WorkflowTurn[Ticket]:
    items = await request.get_input_items()
    if any(item.get("type") in ("function_call_output", "mcp_approval_response") for item in items):
        return WorkflowTurn(responses=await request.get_workflow_responses())

    text = await request.get_input_text()
    return WorkflowTurn(input=Ticket.model_validate_json(text or ""))


server = ResponsesHostServer(
    workflow=build_workflow,
    parse_response=parse_response,
    checkpoint_store_provider=CheckpointStoreProvider(
        allowed_checkpoint_types=[f"{Ticket.__module__}:{Ticket.__qualname__}"],
    ),
)

Użyj synchronicznej lub asynchronicznej fabryki uwzględniającej kontekst żądania dla przepływów pracy, które mogą wstrzymywać, kontynuować lub odzyskiwać pracę w tle. Fabryka musi zwrócić nowo utworzony graf ze świeżymi modyfikowalnymi wykonawcami, agentami, klientami, dostawcami i narzędziami. Zachowaj stabilną nazwę przepływu pracy i identyfikatory wykonawców, aby host mógł przywrócić dokładny punkt kontrolny skojarzony z odpowiedzią zewnętrzną.

Zaufany użytkownik platformy i piaskownica usługi Foundry izolują natywny stan workflow. Host weryfikuje pełną oczekującą partię odpowiedzi, zanim zużyje jakiekolwiek uprawnienie do odpowiedzi. Nieaktualne, częściowe, zduplikowane, odtwarzane, odpowiedzi między użytkownikami i odpowiedzi między piaskownicami kończą się niepowodzeniem przed wykonaniem przepływu pracy. Żądanie z elementem store=false nie zapisuje stanu przepływu pracy i nie może zwrócić pauzy, którą można wznowić.

W przypadku starszego przepływu pracy, który akceptuje list[Message], użyj polecenia response_input_messages(request), aby przekonwertować tylko bieżącą turę Responses. Nie ładuje wcześniejszej historii nadrzędnej ani nie dekoduje oczekujących odpowiedzi przepływu pracy. Hosting agent=workflow.as_agent() pozostaje dostępny w bieżącej wersji beta, ale emituje ostrzeżenie o wycofaniu. Aby zapoznać się z pełnymi przykładami, zobacz przykłady przepływów pracy Native Responses.

Zachowywać stan i obsługiwać długotrwałe rozmowy

ResponsesHostServer i InvocationsHostServer domyślnie konfigurują trwałe magazyny sesji. AgentSessionStoreProvider dostarcza FoundryAgentSessionStore; sesje odpowiedzi używają logicznego magazynu agent_sessions, podczas gdy sesje wywołań używają oddzielnego magazynu invocation_sessions. Te magazyny używają Foundry State Store w środowisku hostowanym oraz magazynu opartego na plikach z zestawu SDK podczas uruchamiania lokalnie.

W przypadku agentów przepływu pracy Responses element CheckpointStoreProvider zapewnia element FoundryCheckpointStore. Przepływy pracy Odpowiedzi natywne i Wywołania używają tego samego dostawcy dla dokładnie tych samych punktów kontrolnych wznowienia. FunctionApprovalStoreProvider udostępnia FoundryFunctionApprovalStore dla oczekujących zatwierdzeń narzędzi agenta. Zamiast tego natywne żądania przepływu pracy i odpowiedzi na zatwierdzenie są powiązane z punktami kontrolnymi przepływu pracy.

Podczas działania w środowisku Foundry domyślna instancja Pythona przechowuje stan obszaru nazw na podstawie identyfikatora użytkownika platformy i identyfikatora sesji sandboxa Foundry. Wymagają one również identyfikatora wywołania platformy dla każdej operacji stanu. Identyfikator wywołania służy do autoryzacji i korelowania operacji; nie jest to identyfikator rozmowy i nie stanowi części klucza magazynowania.

W przypadku odpowiedzi skonfigurowany przez platformę element FOUNDRY_AGENT_SESSION_ID identyfikuje piaskownicę, a inny element agent_session_id dostarczony przez wywołującego jest odrzucany. W przypadku wywołań host weryfikuje kierowany agent_session_id parametr zapytania względem kontekstu żądania. Jeśli FOUNDRY_AGENT_SESSION_ID nie jest skonfigurowane, parametr zapytania musi być obecny, nie może być pusty i musi odpowiadać kontekstowi żądania. Brakujące, zduplikowane lub sprzeczne wartości są odrzucane zamiast użycia zapasowego identyfikatora SDK.

Te gwarancje dotyczą domyślnych sklepów hostowanych. Dostawcy niestandardowego magazynu danych muszą wdrożyć równoważną izolację użytkowników i piaskownicy, przechowywać wewnętrzny element AgentSession.session_id oddzielnie od kluczy wyszukiwania hosta oraz stosować zapisy warunkowe, aby nieaktualne żądania nie mogły nadpisywać nowszych migawek. Nowe klucze powinny używać operacji zapisu tylko do tworzenia, a nie bezwarunkowych operacji upsert. Zobacz przykładową implementację niestandardowego magazynu dla usługi Cosmos DB z operacjami zapisu i usuwania chronionymi za pomocą tagów ETag.

W przypadku history_source="agent" skonfigurowany magazyn sesji zachowuje stan dostawcy przenoszony przez AgentSession, w tym komunikaty z InMemoryHistoryProvider.

Oba hosty akceptują element StoreProvider[SessionStore] za pośrednictwem agent_session_store_provider. Stan sesji musi obsługiwać serializację AgentSession. Zarejestruj kodeki dla niestandardowych typów stanu za pomocą register_state_type(); przywrócony stan nie zachowuje tożsamości obiektu w Pythonie. W nowych domyślnych magazynach danych sesje wygasają po 30 dniach od ich ostatniego zapisu. Niestandardowi dostawcy sami określają okres przechowywania danych.

Domyślne magazyny danych z określonym zakresem nie odczytują starszych danych bez określonego zakresu: agent_sessions, invocation_sessions, punktów kontrolnych ani danych dotyczących zatwierdzania funkcji. Rozpocznij nową konwersację Responses zamiast używać starego previous_response_id lub ID konwersacji. Wywołanie rozpoczyna się od pustej sesji Agent Framework w magazynie z określonym zakresem.

Załadowane AgentSession rekordy używają warunków ETag. Jeśli kolejne żądanie najpierw przechodzi tę samą sesję, nieaktualny zapis kończy się niepowodzeniem zamiast zastępowania nowszego stanu. Ta kontrola nie zapewnia transakcji ani dokładnie jednokrotne wykonanie dla efektów ubocznych agenta lub narzędzia, dlatego aplikacje muszą nadal koordynować nakładające się żądania.

W przypadku pamięci specyficznej dla Responses przekaż StoreProvider do function_approval_store_provider lub ContextScopedStoreProvider do checkpoint_store_provider.

Zewnętrzna operacja w tle używa widocznego dla wywołującego elementu response.id do odpytywania. Domyślnie background_source="agent_server" powoduje, że wykonywanie odbywa się w tle na hoście. Ustaw background_source="provider" tylko z history_source="service" i klientem Responses obsługującym przechowywanie i wznawianie. Jeśli ResponsesServerOptions(resilient_background=True) jest również ustawione, host może wznowić odpytywanie dostawcy dopiero po zapisaniu prywatnego tokenu kontynuacji. Zapewnij, aby lokalne efekty uboczne narzędzia były idempotentne, ponieważ awaria przed zapisaniem następnego tokenu może spowodować ich ponowne wykonanie.

Zaimportuj ResponsesServerOptions z azure.ai.agentserver.responses i przekaż go do ResponsesHostServer za pomocą parametru options. Dostępne długotrwałe opcje konwersacji zależą od typu agenta:

Zdolność Typ agenta Wymagania i zachowanie
Odzyskiwanie punktu kontrolnego przepływu pracy w tle Tylko przepływ pracy Ustaw wartość ResponsesServerOptions(resilient_background=True). Wyślij żądanie Responses za pomocą store=true i background=true. Po ponownym uruchomieniu host wznowi najnowszy trwały punkt kontrolny przepływu pracy lub odtwarza oryginalne dane wejściowe, jeśli nie istnieje żaden punkt kontrolny. Nie konfiguruj magazynu punktów kontrolnych w przepływie pracy, ponieważ host zarządza nim. Spraw, aby zewnętrzne efekty uboczne były idempotentne, ponieważ operacje wykonane po ostatnim trwałym punkcie kontrolnym mogą się powtórzyć.
Natywne odpowiedzi dostawcy w tle Niezwiązany z przepływem pracy Agent z klientem Responses przechowującym dane Ustaw history_source="service" i background_source="provider". Ustaw resilient_background=True, gdy zapisane tokeny kontynuacji dostawcy muszą zostać zachowane po ponownym uruchomieniu hosta.
Konwersacje sterowane Tymczasowo niedostępne Nie ustawiaj wartości steerable_conversations=True. Host zgłasza RuntimeError podczas tworzenia, do czasu aż SDK serwera agenta będzie bezpiecznie obsługiwać odrzucone zmiany sterowania.

Aby zapoznać się z pełnymi implementacjami, zobacz przykłady niestandardowego przechowywania, podstawowej historii odpowiedzi i przetwarzania w tle oraz odpornego, długotrwałego przepływu pracy.

Odczytywanie plików z hostowanego sandboxa

Traktuj trwały zasób $HOME hostowanej piaskownicy jako zasób kierowany przez żądanie, a nie jako ogólną granicę systemu plików. Akceptuj tylko pliki, które aplikacja jawnie przesyła do dedykowanego katalogu, zweryfikuj bieżący identyfikator sandboxa i odrzucaj ścieżki bezwzględne, sekwencje przechodzenia po katalogach, linki, pliki inne niż zwykłe oraz zbyt dużą lub nieprawidłową zawartość.

W przypadku protokołu Responses należy kierować żądanie do hostowanej sesji za pomocą pola treści żądania agent_session_id. Selektor ciągu zapytania jest przeznaczony dla wywołań. Pliki przesyłane w sesji i pliki interpretera kodu Toolbox są oddzielnymi zasobami; przesłany plik sandboxa nie jest automatycznie montowany w kontenerze Toolbox. Zobacz przykład plików sesji, aby zapoznać się z ograniczonymi odczytami UTF-8 oraz wskazówkami dotyczącymi lokalnego i hostowanego przekazywania.

Opcje żądania sterowania

Host mapuje natywne pola generowania odpowiedzi do opcji uruchamiania Agent Framework. Na przykład max_output_tokens staje się max_tokens, a parallel_tool_calls staje się .allow_multiple_tool_calls Spłaszczone wartości z extra_body nadpisują przetłumaczone wartości natywne.

Użyj synchronicznej lub asynchronicznej funkcji hook prepare_options(request, options), aby usunąć lub zastąpić opcje modelu wywołującego, zanim zostanie uruchomiony zwykły agent. Hook nie może ustawiać pól tożsamości, pamięci masowej, kontynuacji ani transportu kontrolowanych przez hosta. W przypadku niestandardowej implementacji SupportsAgentRun, która nie może przyjmować opcji modelu w czasie wykonywania, ustaw "warn" na unsupported_options (wartość domyślna), "ignore" lub "error".

Gdy narzędzie MCP hostowane w usłudze Foundry wymaga zgody użytkownika, ResponsesHostServer zwraca niekompletną odpowiedź z elementem wyjściowym oauth_consent_request . Przedstaw użytkownikowi element consent_link, a następnie kontynuuj, używając identyfikatora niekompletnej odpowiedzi jako previous_response_id, po wyrażeniu zgody przez użytkownika. Host zachowuje sesję agenta dla tej próby i uwidacznia tylko bezwzględne łącza zgody HTTPS.

Jeśli host zna oczekiwane źródła autoryzacji, ogranicz linki zgody za pomocą polecenia allowed_oauth_consent_origins:

server = ResponsesHostServer(
    agent,
    allowed_oauth_consent_origins=[
        "https://logic-region.consent.azure-apihub.net",
        "https://auth.partner.example",
    ],
)

Pominięcie listy dozwolonych powoduje zachowanie weryfikacji bezwzględnej protokołu HTTPS bez ograniczania źródła docelowego. Podanie pustej listy powoduje odrzucenie wszystkich linków zgody. Skonfiguruj wyłącznie dokładne źródła HTTPS; wpisy ze ścieżką, zapytaniem lub fragmentem są odrzucane.

Protokół wywołań

Protokół Wywołania zapewnia pełną kontrolę nad żądaniem HTTP i odpowiedzią. Użyj go, gdy potrzebujesz niestandardowych ładunków, przetwarzania niekonwersacyjnego lub protokołów przesyłania strumieniowego, które nie są zgodne z OpenAI.

Za pomocą protokołu Invocations w języku C# implementujesz niestandardowy InvocationHandler, aby przetwarzać przychodzące żądania.

using Azure.AI.AgentServer.Core;
using Azure.AI.AgentServer.Invocations;
using Microsoft.Agents.AI;

var builder = AgentHost.CreateBuilder(args);

builder.Services.AddSingleton<AIAgent, MyAgent>();
builder.Services.AddInvocationsServer();
builder.Services.AddScoped<InvocationHandler, MyInvocationHandler>();

builder.RegisterProtocol("invocations", endpoints => endpoints.MapInvocationsServer());

var app = builder.Build();
app.Run();

Metoda AddInvocationsServer rejestruje usługi protokołu Invocations. Zaimplementowana jest implementacja InvocationHandler w celu zdefiniowania sposobu przetwarzania poszczególnych żądań przez agenta.

Dla uproszczonej konfiguracji użyj InvocationsHostServer z pakietu agent_framework_foundry_hosting. Automatycznie opakowuje agenta w sposób podobny do ResponsesHostServer i zajmuje się zarządzaniem sesjami:

import os

from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from agent_framework_foundry_hosting import InvocationsHostServer
from azure.identity import DefaultAzureCredential

client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ.get("FOUNDRY_MODEL") or os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
    credential=DefaultAzureCredential(),
)

agent = Agent(
    client=client,
    instructions="You are a friendly assistant. Keep your answers brief.",
    default_options={"store": False},
)

server = InvocationsHostServer(agent)
server.run()

InvocationsHostServer akceptuje to samo wystąpienie lub formularze fabryki w zakresie żądania opisane dla hosta Odpowiedzi. Przywraca zserializowane sesje ze skonfigurowanego repozytorium, dzięki czemu rozpoczęte konwersacje mogą być kontynuowane po ponownym uruchomieniu hosta. Aby uzyskać informacje o sposobie przechowywania, okresie przechowywania i dostosowywaniu, zobacz Utrwalanie stanu i obsługa długotrwałych rozmów.

Gdy są hostowane, Invocations używa zweryfikowanego zakresu żądania opisanego w sekcji Utrwalanie stanu i obsługa długotrwałych konwersacji. Traktuj AgentSession.session_id jako jedną wartość nieprzezroczystą; nie parsuj jej ani nie polegaj na jej wewnętrznej reprezentacji. Lokalne uruchomienia zachowują dotychczasowy sposób działania pamięci masowej dla jednego użytkownika.

Hostuj natywny przepływ pracy za pomocą Invocations

Przekaż workflow= i jawnie określone wywołanie zwrotne parse_request do obsługi natywnego przepływu pracy. Wywołanie zwrotne zarządza schematem JSON aplikacji i zwraca element WorkflowTurn z typizowanymi danymi wejściowymi lub pełną oczekującą partią odpowiedzi:

from pydantic import BaseModel
from starlette.requests import Request

from agent_framework_foundry_hosting import (
    CheckpointStoreProvider,
    InvocationsHostServer,
    WorkflowTurn,
)


class Ticket(BaseModel):
    ticket_id: str
    question: str


class TicketDecision(BaseModel):
    approved: bool


def build_workflow(_request: Request):
    return build_fresh_workflow()


async def parse_request(request: Request) -> WorkflowTurn[Ticket]:
    payload = await request.json()
    stream = payload.get("stream", False)

    if "responses" in payload:
        decisions = {
            request_id: TicketDecision.model_validate(value)
            for request_id, value in payload["responses"].items()
        }
        return WorkflowTurn(responses=decisions, stream=stream)

    ticket = Ticket.model_validate(payload)
    return WorkflowTurn(input=ticket, stream=stream)


server = InvocationsHostServer(
    workflow=build_workflow,
    parse_request=parse_request,
    checkpoint_store_provider=CheckpointStoreProvider(
        allowed_checkpoint_types=[
            f"{Ticket.__module__}:{Ticket.__qualname__}",
            f"{TicketDecision.__module__}:{TicketDecision.__qualname__}",
        ],
    ),
)

Uwzględnij każdy niestandardowy typ aplikacji zapisywany przez przepływ pracy na liście allowed_checkpoint_types dostawcy punktów kontrolnych.

Hostowane przepływy pracy wymagają fabryki obsługującej żądania, która zwraca nowy utworzony graf ze stabilnymi identyfikatorami przepływu pracy i wykonawcy. Przepływ pracy z bezpośrednim kompilowaniem jest dostępny tylko dla lokalnego, jednorazowego uruchomienia, które się nie zatrzymuje.

Odpowiedzi przepływu pracy bez przesyłania strumieniowego używają formatu application/json z listą zdarzeń output. Przesyłanie strumieniowe emituje zdarzenia output i request_info w ramkach, po których done następuje tylko po zapisaniu dokładnego kursora przepływu pracy. Traktuj przesyłane strumieniowo dane wyjściowe jako tymczasowe do done. Natywne przepływy pracy nie obsługują legacy_wire_format=True.

Host weryfikuje odpowiedzi względem dokładnego oczekującego punktu kontrolnego w kontekście zaufanego użytkownika i piaskownicy. Jeśli przepływ ma wiele oczekujących żądań, odpowiedz na pełną partię w jednej turze. Informacje o działającym analizatorze, typowanym przepływie pracy zgłoszeń, liście dozwolonych typów punktów kontrolnych oraz przykładach JSON/SSE można znaleźć w przykładowym przepływie pracy wywołań natywnych.

Dostosowywanie żądań i odpowiedzi dla wywołań

Domyślnie POST /invocations akceptuje obiekt JSON z ciągiem message, obiektem opcjonalnym options i opcjonalną wartością logiczną stream . Aby zaakceptować ładunek specyficzny dla aplikacji, przekaż synchroniczne lub asynchroniczne parse_request wywołanie zwrotne, które zwraca wartość InvocationRun(messages, options, stream). Użyj prepare_options, aby filtrować lub zastępować kopię opcji generowania obiektu wywołującego, zanim agent zostanie uruchomiony.

Host weryfikuje dane wyjściowe haka i odrzuca mechanizmy sterowania tożsamością platformy, pamięcią masową, kontynuacją i wykonywaniem agenta. W przypadku agentów, którzy nie akceptują opcji środowiska uruchomieniowego, ustaw wartość unsupported_options"warn" (wartość domyślną), "ignore"lub "error". Zobacz przykład analizatora wywołań, aby uzyskać pełną implementację.

W przypadku powodzenia bez przesyłania strumieniowego zwracane są dane JSON w postaci {"response": "..."}. Transmisja strumieniowa wykorzystuje zdarzenia wysyłane przez serwer: jedną lub więcej ramek event: delta, po których następuje event: done w przypadku powodzenia lub event: error w przypadku niepowodzenia. Strumień może emitować delty przed wystąpieniem błędu, więc klienci muszą traktować done, a nie deltę, jako pomyślne zakończenie. Host emituje done tylko po zakończeniu strumienia odpowiedzi i utrwala AgentSessionelement . Jest to session_id identyfikator trasy piaskownicy platformy, a nie serializowany AgentSession.session_id.

Ustaw legacy_wire_format=True tylko podczas migrowania istniejących klientów, którzy wymagają poprzedniej odpowiedzi w postaci zwykłego tekstu i nieprzetworzonego strumienia fragmentów tekstu. Ten tryb zgodności jest przestarzały i nie konwertuje niepowodzeń na pomyślny tekst. Host przetwarza żądania z tej samej sesji sekwencyjnie tylko w obrębie jednego procesu; konflikt operacji compare-and-swap między procesami może nadal wystąpić po działaniu zewnętrznego narzędzia.

Protokół Invocations nie wznawia uruchomień przepływu pracy, które oczekują lub zostały przerwane. Użyj niestandardowego wzorca obsługi w poniższej sekcji, jeśli potrzebujesz innego zachowania kontynuacji przepływu pracy.

Aby uzyskać pełną kontrolę nad obsługą żądań, użyj InvocationAgentServerHost bezpośrednio z pakietu azure.ai.agentserver.invocations i zaimplementuj własną procedurę obsługi wywołań:

import os
from collections.abc import AsyncGenerator

from agent_framework import Agent, AgentSession
from agent_framework.foundry import FoundryChatClient
from azure.ai.agentserver.invocations import InvocationAgentServerHost
from azure.identity import DefaultAzureCredential
from starlette.requests import Request
from starlette.responses import JSONResponse, Response, StreamingResponse

_sessions: dict[str, AgentSession] = {}

client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ.get("FOUNDRY_MODEL") or os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
    credential=DefaultAzureCredential(),
)

agent = Agent(
    client=client,
    instructions="You are a friendly assistant. Keep your answers brief.",
    default_options={"store": False},
)

app = InvocationAgentServerHost()


@app.invoke_handler
async def handle_invoke(request: Request):
    """Handle streaming multi-turn chat."""
    data = await request.json()
    session_id = request.state.session_id
    stream = data.get("stream", False)
    user_message = data.get("message", None)

    if user_message is None:
        return Response(content="Missing 'message' in request", status_code=400)

    session = _sessions.setdefault(session_id, AgentSession(session_id=session_id))

    if stream:

        async def stream_response() -> AsyncGenerator[str]:
            async for update in agent.run(user_message, session=session, stream=True):
                yield update.text

        return StreamingResponse(
            stream_response(),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
        )

    response = await agent.run([user_message], session=session, stream=stream)
    return JSONResponse({"response": response.text})


if __name__ == "__main__":
    app.run()

Warning

Magazyn sesji w pamięci w przykładowym niestandardowym programie obsługi jest tracony po ponownym uruchomieniu. Użyj magazynu trwałego (na przykład Cosmos DB) w środowisku produkcyjnym.

Kompletny przykład wdrożenia Invocations znajdziesz w przykładzie Telegrama hostowanym w Foundry. Umieszcza usługę API Management przed webhookiem hostowanego agenta i używa tożsamości zarządzanych, usługi Key Vault oraz usługi Cosmos DB do przechowywania trwałej historii konwersacji.

Uwaga / Notatka

Obsługa języka Go dla agentów hostowanych w usłudze Foundry będzie wkrótce dostępna. Aktualny status znajdziesz w repozytorium Agent Framework dla Go.

Tip

Zapoznaj się z przykładami Python lub przykładami języka C#, aby zapoznać się z przykładami projektu hostowanego agenta. Możesz też użyć polecenia azd ai agent init do utworzenia szkieletu projektu agenta hostowanego od podstaw. Zapoznaj się z tym przewodnikiem Szybki start , aby uzyskać instrukcje krok po kroku.

Uruchamianie lokalnie

Interfejs wiersza polecenia dewelopera Azure (azd) zapewnia najprostszy sposób uruchamiania i testowania hostowanego agenta lokalnie.

Inicjowanie projektu

Utwórz nowy folder i zainicjuj go na podstawie pliku azure.yaml przykładu. Użyj wdrożenia kontenera w tym przewodniku:

mkdir my-hosted-agent && cd my-hosted-agent
azd ai agent init -t <path-to-azure.yaml> --src . --deploy-mode container

Tip

Zastąp <path-to-azure.yaml> lokalną ścieżką lub obsługiwanym adresem URL GitHub do dokumentu projektu azure.yaml przykładu.

Ustawianie zmiennych środowiskowych

export FOUNDRY_PROJECT_ENDPOINT="https://<account>.services.ai.azure.com/api/projects/<project>"
export FOUNDRY_MODEL="<your-model-deployment>"

Uruchom hosta agenta

azd ai agent run

Agent host rozpoczyna działanie na http://localhost:8088.

Wywoływanie agenta

azd ai agent invoke --local "Hello!"

Możesz też użyć curl:

curl -X POST http://localhost:8088/responses \
  -H "Content-Type: application/json" \
  -d '{"input": "Hello!"}'

Lub w programie PowerShell:

(Invoke-WebRequest -Uri http://localhost:8088/responses -Method POST -ContentType "application/json" -Body '{"input": "Hello!"}').Content

Wdrażanie do platformy Foundry

Po zweryfikowaniu agenta lokalnie wdróż go w usłudze Microsoft Foundry:

  1. Aprowizuj zasoby (jeśli nie masz jeszcze projektu Foundry):

    azd provision

    Tworzy grupę zasobów z instancją Foundry, projektem, implementacją modelu, Application Insights i rejestrem kontenerów.

  2. Wdróż agenta:

    azd deploy

    Spakietuje to Twojego agenta jako obraz kontenera, wypchnie go do Azure Container Registry i wdroży w usłudze Foundry Agent Service.

Infrastruktura hostingu rozwiązania Foundry automatycznie wprowadza następujące zmienne środowiskowe do kontenera agenta w czasie wykonywania:

Variable Opis
FOUNDRY_PROJECT_ENDPOINT Adres URL punktu końcowego dla projektu Foundry.
AZURE_AI_MODEL_DEPLOYMENT_NAME Nazwa wdrożenia modelu zarządzanego przez usługę azd skonfigurowana podczas azd ai agent init. Kod w Pythonie może preferować lokalnie FOUNDRY_MODEL i użyć tej hostowanej wartości jako rezerwowej.
APPLICATIONINSIGHTS_CONNECTION_STRING Usługa Application Insights parametry połączenia na potrzeby telemetrii.

Po wdrożeniu agent jest dostępny za pośrednictwem dedykowanego punktu końcowego rozwiązania Foundry i można go również przetestować w portalu Foundry.

Następne kroki

Pojęcia dotyczące hostowanych agentów

  • Ostatnia aktualizacja o