Foundry hostovaní agenti

Hostovaní agenti ve službě agenta Microsoft Foundry umožňují nasadit kontejnerizované aplikace agentů do Microsoft spravované infrastruktury. Platforma zpracovává škálování, trvalost stavu relace, zabezpečení a správu životního cyklu, abyste se mohli soustředit na logiku vašeho agenta. Microsoft Agenti hostované v Foundry jsou obecně k dispozici a podporují agenty vytvořené s vlastním kódem nebo upřednostňovanou architekturou agentů. Tento článek se zabývá konkrétně integrací hostování rozhraní Agent Framework.

Pomocí integrace hostování rozhraní Agent Framework můžete vystavit Agent prostřednictvím protokolu Foundry Responses nebo Invocations s minimálním kódem. Python také podporuje hostování nativní Workflow přímo, aniž byste ho převáděli na agenta.

Poznámka:

Kód agenta sestavený s jinými architekturami můžete nasadit také do agentů hostovaných ve Foundry pomocí pracovních postupů rozhraní příkazového řádkuazd pro vývojáře Azure. Obecné koncepty nezávislé na konkrétním frameworku a pokyny k nasazení najdete v článku Co jsou hostovaní agenti? Zbytek tohoto článku se zaměřuje na integraci s Agent Frameworkem.

Kdy použít hostované agenty

Zvolte agenty hostované v Foundry, pokud chcete:

  • Spravovaná infrastruktura – nemusíte konfigurovat kontejnery, webové servery ani pravidla škálování sami.
  • Integrovaná správa relací – platforma zachovává stav a nahrané soubory během přechodů mezi interakcemi a období nečinnosti.
  • Identita vyhrazeného agenta – každý nasazený agent získá vlastní identitu Entra pro zabezpečený přístup k modelům, nástrojům a podřízeným službám.
  • Koncové body kompatibilní s OpenAI – klienti můžou komunikovat s vaším agentem pomocí libovolné sady SDK kompatibilní s OpenAI prostřednictvím protokolu Odpovědi.
  • V případě zvukových agentů v reálném čase používejte hostované agenty se službou Azure Speech in Foundry Tools (Voice Live) pro detekci hlasových aktivit na straně serveru, zrušení ozvěny a snížení šumu. Podrobnosti najdete v tématu Použití služby Voice Live s hostovanými agenty.

Poznámka:

Integrace jazyka Python agent-framework-foundry-hosting je ve verzi Preview. Hostovaní agenti Microsoft Foundry, spravovaná hostingová služba, jsou nyní obecně dostupní.

Prerequisites

  • Předplatné Azure
  • rozhraní příkazového řádku pro vývojáře Azure (azd) s rozšířením agenta AI: azd ext install azure.ai.agents

Pro místní testování potřebujete také:

Nainstalujte hostující balíček NuGet:

dotnet add package Microsoft.Agents.AI.Foundry.Hosting --prerelease
  • Python 3.10 nebo novější

Nainstalujte předem připravený balíček pro hostování, klienta Foundry a Azure ověřovací balíček:

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

V Foundry poskytuje platforma kontext uživatele volajícího a kontext volání; hostitelská infrastruktura je používá k izolaci stavu na uživatele a předávání kontextu žádosti do služeb Foundry. Při místním spuštění aplikace nezískají kontext dané platformy, takže si v případě potřeby musí zajistit vlastní mechanismy pro identitu a řízení stavu.

Protokol odpovědí

Protokol odpovědi je doporučeným výchozím bodem pro většinu agentů. Zpřístupní koncový bod kompatibilní /responses s OpenAI a platforma automaticky spravuje historii konverzací, streamování a životní cyklus relací.

U agentů Pythonu s hostováním platí, že odpověď, která předčasně skončí, má stav incomplete. Streamovací klienti obdrží ukončovací událost response.incomplete, zatímco nestreamovací klienti obdrží status nastavené na incomplete. Důvod ukončení content_filter odpovídá hodnotě incomplete_details.reason nastavené na content_filter a length odpovídá max_output_tokens. Veškerý vygenerovaný výstup nebo obsah odmítnutí je nadále k dispozici v odpovědi.

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

Vytváří AgentHost.CreateBuilder server aplikace předkonfigurovaný pro hostitelské prostředí Foundry. AddFoundryResponses zaregistruje vašeho agenta pomocí obslužné rutiny protokolu Odpovědi a MapFoundryResponses mapuje /responses koncový bod 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()

Zabalí ResponsesHostServer agenta a zpřístupní ho prostřednictvím protokolu Foundry Responses. Pole volajícího store určuje, zda se uloží externí odpověď, relace spravovaná hostitelem a stav schválení. Nastavení history_source nezávisle vybere, kdo dodává historii modelů:

history_source Chování historie modelu
"agent_server" (výchozí) Hostitel rekonstruuje uložený externí záznam odpovědí a zakáže ukládání v následné službě, aby se předešlo duplicitní historii.
"service" Hostitel odešle pouze aktuální vstup a soukromě uloží ID pokračování služby modelu. Uložená konverzace poskytovatele se nemůže větvit od dřívější odpovědi.
"agent" Hostitel odešle pouze aktuální vstup. Historii spravuje výchozí nastavení úložiště agenta HistoryProvider nebo podřízené služby.

Nekombinujte "agent_server" ani "service" s HistoryProvider podporujícím zatížení. Výchozí režim také odmítá pevně dané možnosti následného pokračování, jako jsou conversation_id, previous_response_id a conversation. Použijte history_source="agent" pro vlastní implementaci SupportsAgentRun.

Parametr konstruktoru response_store určuje backend pro perzistenci outer Responses. Starší parametr konstruktoru store je zastaralý alias pro response_store; ani jeden z parametrů nenastavuje pole store volajícího pro jednotlivé požadavky. Požadavek s store=false je jednorázový: neukládá stav spravovaný hostitelem, zakáže podporované následné úložiště a nemůže používat background=true.

Hostitel vlastní dodaného agenta a může přidat zprostředkovatele kontextu specifického pro hostování. Nepoužívejte agenta znovu s jiným hostem ani ho nevyvolávejte přímo po vytvoření hosta.

Hostitel Responses zachovává nativní systémová volání, snímky obrazovky a bezpečnostní kontroly. Vaše aplikace musí provést požadované akce a explicitně potvrdit všechny bezpečnostní kontroly. Úplný tok najdete v tématu Použití nativního počítače.

Volba instance agenta nebo továrny

ResponsesHostServer i InvocationsHostServer přijímají prostřednictvím parametru agent buď instanci agenta, nebo synchronní či asynchronní volatelný objekt bez argumentů. Hostitel znovu používá instanci po celou dobu její existence. Volání se spustí jednou na požadavek a vrácený agent patří do této žádosti.

Použijte callable, když běžný agent zachovává proměnlivý stav mimo AgentSession. Hostitelé uchovávají pouze podporovaná úložiště relací, checkpointů a schválení funkcí, ne libovolná pole v agentovi s rozsahem požadavku.

Warning

Hostování WorkflowAgent, například workflow.as_agent(), prostřednictvím agent= je zastaralé. WorkflowAgent udržuje stav pracovního postupu v paměti mezi běhy, takže jedna instance nesmí nikdy obsluhovat žádosti od různých uživatelů nebo z různých konverzací. Hostujte pracovní postup nativně prostřednictvím workflow= a továrny zohledňující požadavky.

Dokud nemigrujete hostitele Responses, továrna, která pro každý požadavek vytvoří nový pracovní postup, exekutory, zabalené agenty, klienty, poskytovatele a nástroje, je bezpečnou starší variantou. Udržujte název workflow a ID exekutorů stabilní, aby hostitel mohl obnovit kontrolní body. Pro hostitele Invocations host je tato starší forma factory vhodná pouze pro bezstavové jednokrokové pracovní postupy, protože hostování agenta neobnovuje kontrolní body pracovního postupu.

Tovární objekt použijte také, pokud integrace nese identitu požadavku nebo spravuje prostředky specifické pro daný požadavek. Například v rámci továrny vytvořte připojení MCP, sady nástrojů, poskytovatele dovedností, klienty služby Search, poskytovatele paměti a jejich přihlašovací údaje, pokud používají aktuální kontext volání platformy nebo uživatelský kontext. Opětovné použití připojení MCP pro celý proces může zachovat identitu žádosti, která ji otevřela.

Hostitel pro každou žádost zadá a ukončí agenty vytvořené továrnou. Agent spravuje kontextově spravované klienty a nástroje MCP, ale vaše továrna musí zavřít jakéhokoli jiného poskytovatele, přenos nebo přihlašovací údaje, které vytvoří. Nezavírejte sdílené objekty, které aplikace poskytla mimo továrnu.

Hostujte nativní workflow pomocí Responses

Python může provozovat sestavený pracovní postup přímo prostřednictvím workflow=. Nativní pracovní postup vyžaduje parse_response zpětné volání, které mapuje aktuální požadavek Responses na typovaný počáteční vstup nebo úplnou dávku čekajících odpovědí:

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__}"],
    ),
)

Pro pracovní postupy, které můžou pozastavit, pokračovat nebo obnovit práci na pozadí, použijte synchronní nebo asynchronní továrnu v kontextu požadavku. Factory musí vrátit nově sestavený graf s novými měnitelnými executory, agenty, klienty, poskytovateli a nástroji. Udržujte název workflow a ID executorů stabilní, aby hostitel mohl obnovit přesný kontrolní bod přidružený k vnější odpovědi.

Důvěryhodná platforma uživatele a sandbox Foundry izolují nativní stav pracovního postupu. Hostitel ověří úplnou čekající dávku odpovědí, než spotřebuje jakékoli oprávnění k odpovědím. Zastaralé, částečné, duplicitní, opakované, odpovědi mezi uživateli a mezi sandboxy selžou před spuštěním pracovního postupu. Požadavek s store=false neuloží stav pracovního postupu a nemůže vrátit obnovitelné pozastavení.

Pro starší pracovní postup, který přijímá list[Message], slouží response_input_messages(request) k převodu pouze aktuálního turn Responses. Nenačítá předchozí vnější historii ani nedekóduje čekající odpovědi workflow. Hostování agent=workflow.as_agent() zůstává dostupné během aktuální beta verze, ale vygeneruje upozornění na vyřazení. Kompletní příklady najdete v ukázkách nativních pracovních postupů Responses.

Zachování stavu a zpracování dlouhotrvajících konverzací

ResponsesHostServer a InvocationsHostServer ve výchozím nastavení nakonfigurujte trvalé úložiště relací. AgentSessionStoreProvider poskytuje FoundryAgentSessionStore; relace odpovědí používají logické úložiště agent_sessions, zatímco relace volání používají samostatné úložiště invocation_sessions. Tato úložiště používají Foundry State Store v hostovaném prostředí a souborové úložiště sady SDK při lokálním spuštění.

Pro agenty pracovního postupu Responses poskytuje CheckpointStoreProviderFoundryCheckpointStore. Pracovní postupy nativních odpovědí a invokací používají stejného poskytovatele pro přesné kontrolní body pro pokračování. FunctionApprovalStoreProvider poskytuje FoundryFunctionApprovalStore možnost schvalování nevyřízených žádostí o schválení nástrojů agenta. Integrované odpovědi na žádosti a schválení v pracovním postupu jsou místo toho vázány na kontrolní body pracovního postupu.

Při spuštění ve Foundry výchozí Python ukládá stav jmenného prostoru pod ID uživatele platformy a ID relace sandboxu Foundry. Vyžadují také ID volání platformy pro každou stavovou operaci. ID volání autorizuje a koreluje operaci; nejedná se o ID konverzace a není součástí klíče úložiště.

U odpovědí sandbox identifikuje hodnota FOUNDRY_AGENT_SESSION_ID nakonfigurovaná platformou a jiná hodnota agent_session_id, zadaná volajícím, je odmítnuta. Při vyvolání hostitel ověřuje směrovaný parametr dotazu agent_session_id vůči kontextu požadavku. Pokud FOUNDRY_AGENT_SESSION_ID není nakonfigurovaný, musí být parametr dotazu přítomný, neprázdný a musí odpovídat kontextu požadavku. Chybějící, duplicitní nebo konfliktní hodnoty se místo použití náhradního ID sady SDK zamítnou.

Tyto záruky platí pro výchozí hostované obchody. Vlastní poskytovatelé úložiště musí implementovat ekvivalentní izolaci uživatele a sandboxu, zachovat vnitřní AgentSession.session_id odděleně od klíčů vyhledávání hostitelů a používat podmíněné zápisy, aby zastaralé požadavky nemohly přepsat novější snímky. Nové klíče by měly místo nepodmíněných upsertů používat zápisy pouze při vytváření. Podívejte se na ukázku vlastního úložiště pro implementaci Cosmos DB s zápisy a odstraněními chráněnými značkou ETag.

S history_source="agent" nakonfigurované úložiště relací uchovává stav poskytovatele přenášený pomocí AgentSession, včetně zpráv z InMemoryHistoryProvider.

Oba hostitelé přijímají StoreProvider[SessionStore] prostřednictvím agent_session_store_provider. Stav relace musí podporovat AgentSession serializaci. Zaregistrujte kodeky pro vlastní typy stavů. register_state_type()Obnovený stav nezachová Python identitu objektu. Platnost nových výchozích úložišť vyprší 30 dnů po posledním zápisu. Vlastní poskytovatelé řídí vlastní uchovávání informací.

Výchozí úložiště s vymezeným oborem nečtou starší neskopovaná agent_sessionsdata, invocation_sessionskontrolní bod ani data schválení funkcí. Začněte novou konverzaci ve službě Responses místo opětovného použití starého previous_response_id nebo ID konverzace. Vyvolání začíná prázdnou relací agenta Framework v omezeném úložišti.

Načtené záznamy AgentSession používají podmínky ETag. Pokud jiný požadavek nejprve posune stav téže relace, zastaralý zápis selže, místo aby přepsal novější stav. Tato kontrola neposkytuje transakce ani spuštění právě jednou pro vedlejší účinky na straně agenta nebo nástroje, takže aplikace musí i nadále koordinovat překrývající se požadavky.

Pro úložiště specifické pro odpovědi předejte StoreProvider do function_approval_store_provider nebo ContextScopedStoreProvider do checkpoint_store_provider.

Vnější úloha na pozadí používá pro dotazování objekt response.id, který je viditelný pro volajícího. Výchozí možnost background_source="agent_server" zachovává spouštění na pozadí v hostiteli. Nastavte background_source="provider" pouze s history_source="service" a ukládajícím, obnovitelným klientem Responses. Pokud je také nastaveno ResponsesServerOptions(resilient_background=True), hostitel může obnovit dotazování poskytovatele teprve po uložení soukromého tokenu pro pokračování. Zajistěte, aby lokální vedlejší efekty nástroje byly idempotentní, protože pád před uložením dalšího tokenu může způsobit jejich opětovné provedení.

Importujte ResponsesServerOptions z azure.ai.agentserver.responsesparametru a předejte ho ResponsesHostServer prostřednictvím parametru options . Dostupné možnosti dlouhotrvající konverzace závisí na typu agenta:

Schopnost Typ agenta Požadavky a chování
Obnovení kontrolního bodu pracovního postupu na pozadí Pouze pracovní postup Nastavte ResponsesServerOptions(resilient_background=True). Odešlete žádost o odpovědi s store=true a background=true. Po restartu host obnoví nejnovější kontrolní bod trvalého pracovního postupu nebo znovu přehraje původní vstup, pokud žádný kontrolní bod neexistuje. Nekonfigurujte úložiště kontrolních bodů v pracovním postupu, protože ho hostitel spravuje. Zajistěte, aby externí vedlejší efekty byly idempotentní, protože operace po posledním perzistentním kontrolním bodu se mohou opakovat.
Nativní odpovědi na pozadí pro poskytovatele Bez pracovního postupu Agent s klientem ukládání odpovědí Nastavte history_source="service" a background_source="provider". Nastavte resilient_background=True, pokud uložené pokračovací tokeny poskytovatele musí přetrvat restartování hostitele.
Řiditelné konverzace Dočasně nedostupné Nenastavujte steerable_conversations=True. Hostitel vyvolá RuntimeError při vytváření, dokud Server Agent SDK nedokáže bezpečně zpracovávat odmítnuté požadavky na řízení.

Kompletní implementace najdete v ukázkách vlastního úložiště, základní historie odpovědí a zpracování na pozadí a odolného dlouhotrvajícího pracovního postupu.

Čtení souborů z hostovaného sandboxu

Považujte trvalý $HOME hostovaného sandboxu za prostředek směrovaný podle požadavků, nikoli za obecnou hranici souborového systému. Přijímejte pouze soubory, které vaše aplikace výslovně nahrává do vyhrazeného adresáře, ověřujte aktuální identitu sandboxu a odmítejte absolutní cesty, sekvence pro přechod adresáři, odkazy, soubory, které nejsou běžnými soubory, a příliš velký nebo neplatný obsah.

U protokolu Responses směřujte požadavek na hostovanou relaci pomocí pole agent_session_id v těle požadavku. Selektor řetězce dotazu je určen pro vyvolání. Nahrávání relací a soubory interpretu kódu sady nástrojů jsou samostatné prostředky; nahraný soubor sandboxu se automaticky nepřipevní do kontejneru nástrojů. Projděte si ukázku souborů relací pro vázané čtení UTF-8 a pokyny pro místní a hostované nahrávání.

Možnosti řízení požadavků

Hostitel mapuje nativní pole generování odpovědí na možnosti spuštění rozhraní Agent Framework. Například max_output_tokens se stane max_tokens, a parallel_tool_calls se stane allow_multiple_tool_calls. Zploštěné hodnoty z extra_body přepisují přeložené nativní hodnoty.

Pomocí synchronního nebo asynchronního prepare_options(request, options) háku odeberte nebo nahraďte možnosti modelu volajícího před spuštěním běžného agenta. Háček nemůže nastavit pole identity řízené hostitelem, úložištěm, pokračováním nebo přenosem. Pro vlastní SupportsAgentRun implementaci, která nemůže přijmout možnosti modelu modulu runtime, nastavte unsupported_options na "warn" (výchozí) "ignore"nebo "error".

Pokud nástroj MCP hostovaný foundry vyžaduje souhlas uživatele, ResponsesHostServer vrátí neúplnou odpověď s výstupní položkou oauth_consent_request . Předložte uživateli jeho consent_link a poté pokračujte s ID neúplné odpovědi jako previous_response_id poté, co uživatel udělí souhlas. Hostitel uchovává relaci agenta pro tento opakovaný pokus a zpřístupňuje pouze absolutní odkazy HTTPS pro udělení souhlasu.

Pokud váš hostitel zná očekávaný původ autorizace, omezte odkazy na souhlas s allowed_oauth_consent_origins:

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

Vynechání seznamu povolených položek zachová absolutní ověření HTTPS bez omezení cílového původu. Poskytnutí prázdného seznamu odmítne každý odkaz na vyjádření souhlasu. Konfigurujte pouze přesné zdroje HTTPS; položky obsahující cestu, dotaz nebo fragment budou odmítnuty.

Protokol vyvolání

Protokol Vyvolání poskytuje úplnou kontrolu nad požadavkem a odpovědí HTTP. Použijte ho, pokud potřebujete vlastní datové části, nekoverzační zpracování nebo streamovací protokoly, které nejsou kompatibilní s OpenAI.

Při použití protokolu Invocations v jazyce C# implementujete vlastní InvocationHandler, abyste zpracovali příchozí požadavky.

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 zaregistruje služby protokolu Invocations. Implementujete InvocationHandler , abyste definovali, jak agent zpracovává jednotlivé požadavky.

Pro odlehčenou instalaci použijte InvocationsHostServer z agent_framework_foundry_hosting balíčku. Zabalí vašeho agenta podobně jako ResponsesHostServer a automaticky spravuje relaci.

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 přijímá stejné formuláře objektu pro vytváření s oborem požadavků, které jsou popsány pro hostitele Odpovědi. Obnoví serializované relace z nakonfigurovaného úložiště, aby dokončené konverzace mohly pokračovat po restartování hostitele. Informace o chování úložiště, uchovávání a přizpůsobení najdete v tématu Zachování stavu a zpracování dlouhotrvajících konverzací.

Při hostování používá Invocations ověřený rozsah požadavku popsaný v části Zachování stavu a zpracování dlouhotrvajících konverzací. Považujte AgentSession.session_id za jednu neprůhlednou hodnotu; neanalyzujte jeho vnitřní reprezentaci ani na ní nezáviste. Místní běhy zachovávají své stávající chování úložiště pro jednoho uživatele.

Hostujte nativní workflow pomocí Invocations

Předání workflow= a explicitního parse_request callbacku pro hostování nativního workflow Zpětné volání spravuje schéma JSON aplikace a vrátí WorkflowTurn buď typovaný vstup, nebo celou dávku čekajících odpovědí:

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__}",
        ],
    ),
)

Zahrňte každý vlastní typ aplikace, který pracovní postup ukládá do seznamu poskytovatele checkpointů allowed_checkpoint_types.

Hostované pracovní postupy vyžadují továrnu zohledňující kontext požadavku, která vrací nově sestavený graf se stabilními ID pracovních postupů a exekutorů. Pracovní postup přímého sestavení je k dispozici pouze pro místní jednorázové spuštění, které se nepozastaví.

Odpovědi pracovního postupu bez streamování používají application/json se seznamem událostí output. Streamování vysílá události output a request_info v rámci streamu, následované done až po uložení přesného kurzoru workflow. Považujte streamovaný výstup za prozatímní, dokud done. Nativní pracovní postupy nepodporují legacy_wire_format=True.

Hostitel ověřuje odpovědi vůči přesnému čekajícímu checkpointu v oboru důvěryhodného uživatele a sandboxu. Pokud má pracovní postup více čekajících žádostí, odpovězte na celou dávku najednou. Ukázky spustitelného parseru, pracovního postupu typed ticket, seznamu povolených typů checkpointů a příkladů JSON/SSE najdete v ukázce native Invocations workflow sample.

Přizpůsobení požadavků a odpovědí na vyvolání

Ve výchozím nastavení POST /invocations přijímá objekt JSON s řetězcem message, volitelným options objektem a volitelnou logickou stream hodnotou. Chcete-li přijmout payload specifický pro aplikaci, předejte synchronní nebo asynchronní callback parse_request, který vrací InvocationRun(messages, options, stream). Pomocí prepare_options lze před spuštěním agenta filtrovat nebo nahradit kopii možností generování pro volajícího.

Host ověří výstup hooku a odmítne mechanismy řízení identity platformy, úložiště, pokračování a spouštění agenta. U agentů, kteří nepřijímají možnosti modulu runtime, nastavte unsupported_options na "warn" (výchozí) "ignore"nebo "error". Úplnou implementaci najdete v ukázce analyzátoru invokací.

Při úspěšném požadavku bez streamování se vrátí JSON ve formátu {"response": "..."}. Streamování používá události odeslané serverem: jeden nebo více event: delta snímků následovaných úspěchem event: done nebo event: error selháním. Stream může vysílat delty před chybou, takže klienti musí za úspěšné dokončení považovat done, nikoli deltu. Hostitel vygeneruje done pouze po dokončení streamu odpovědi a zachování AgentSession. Jeho session_id je ID trasy sandboxu platformy, nikoli serializované AgentSession.session_id.

Nastavte legacy_wire_format=True pouze při migraci stávajících klientů, kteří vyžadují předchozí odpověď v prostém textu a nezpracovaný proud textových bloků. Tento režim kompatibility je zastaralý a nepřevádí chyby na úspěšný text. Hostitel serializuje požadavky v rámci stejné relace pouze v jednom procesu; ke konfliktu compare-and-swap napříč procesy může stále dojít po zásazích externích nástrojů.

Protokol Invocations neobnoví běhy pracovního postupu, které čekají na dokončení nebo byly přerušeny. Pokud potřebujete jiné chování pokračování pracovního postupu, použijte vlastní vzor obslužné rutiny v následující části.

Pro úplnou kontrolu nad zpracováním požadavků použijte InvocationAgentServerHost přímo z balíčku a implementujte vlastní azure.ai.agentserver.invocations rutinu vyvolání:

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

Při restartování dojde ke ztrátě úložiště relací v paměti v příkladu vlastní obslužné rutiny. V produkčním prostředí používejte trvalé úložiště (například Cosmos DB).

Kompletní nasazení Invocations najdete v ukázce Telegramu hostované ve Foundry. Umístí službu API Management před webhook hostovaného agenta a používá spravované identity, Key Vault a Cosmos DB pro trvalou historii konverzací.

Poznámka:

Podpora Go pro agenty hostované v Foundry bude brzy k dispozici. Nejnovější stav najdete v úložišti Agent Framework Go .

Tip

Příklady hostovaného projektu agenta najdete v Python ukázkách nebo ukázkách jazyka C#. Nebo pomocí azd ai agent init příkazu vygenerujte nový projekt hostovaného agenta úplně od začátku. Podrobné pokyny najdete v této příručce pro rychlý start .

Lokální spuštění

Rozhraní příkazového řádku pro vývojáře Azure (azd) poskytuje nejjednodušší způsob, jak spustit a otestovat hostovaného agenta místně.

Inicializace projektu

Vytvořte novou složku a inicializujte ji ze souboru azure.yaml ukázky. V tomto názorném postupu použijte nasazení kontejnerů:

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

Tip

Nahraďte <path-to-azure.yaml> místní cestou nebo podporovanou adresou URL GitHub do dokumentu ukázkového projektu azure.yaml.

Nastavení proměnných prostředí

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

Spusťte hostitele agenta

azd ai agent run

Hostitel agenta se spustí na http://localhost:8088.

Vyvolání agenta

azd ai agent invoke --local "Hello!"

Nebo použijte curl:

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

Nebo v PowerShellu:

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

Nasazení do Foundry

Jakmile agenta ověříte místně, nasaďte ho do služby Microsoft Foundry:

  1. Poskytnutí zdrojů (pokud ještě projekt Foundry nemáte):

    azd provision

    Tím se vytvoří skupina prostředků s instancí Foundry, projektem, nasazením modelu, Application Insights a registrem kontejneru.

  2. Nasazení agenta:

    azd deploy

    Tím se váš agent zabalí jako image kontejneru, odešle ji do Azure Container Registry a nasadí ji do služby Foundry Agent Service.

Infrastruktura hostování Foundry automaticky vloží následující proměnné prostředí do kontejneru agenta za běhu:

Proměnné Description
FOUNDRY_PROJECT_ENDPOINT Adresa URL koncového bodu pro projekt Foundry.
AZURE_AI_MODEL_DEPLOYMENT_NAME Název nasazení modelu spravovaného službou azd nakonfigurovaný během azd ai agent init. Python kód může upřednostnit FOUNDRY_MODEL lokálně a jako záložní použít tuto hostovanou hodnotu.
APPLICATIONINSIGHTS_CONNECTION_STRING Řetězec připojení Application Insights pro telemetrii.

Po nasazení je váš agent přístupný prostřednictvím vyhrazeného koncového bodu Foundry a dá se také otestovat z portálu Foundry.

Další kroky

Koncepty hostovaných agentů

  • Poslední aktualizace v