Trwałe rozszerzenie

Rozszerzenie Durable Extension for Microsoft Agent Framework zapewnia trwałe wykonywanie agentom, orkiestracji wielu agentów i przepływom pracy platformy agentów Microsoft. Służy do utrwalania sesji agentów, orkiestracji punktów kontrolnych i postępu przepływu pracy, odzyskiwania po awariach i skalowania pracy na hostach rozproszonych bez zmiany podstawowej logiki agenta.

Rozszerzenie obsługuje dwa modele hostingu w języku C# i Python:

  • Azure Functions dla zarządzanego, bezserwerowego hostingu za pomocą modelu programowania Azure Functions.
  • Bring-your-own-compute/ self-hosted do uruchamiania trwałych agentów i przepływów pracy we własnym procesie roboczym, usłudze, kontenerze, środowisku Kubernetes lub istniejącej infrastrukturze aplikacji.

Note

W tym przewodniku omówiono hosting rozszerzenia Durable Extension w języku C# i Python. Aby uzyskać informacje na temat hostingu języka Go, zobacz Hosting A2A i repozytorium Agent Framework Go.

Overview

Agenci durable agents łączą model programowania Platformy agentów z infrastrukturą Durable Task, taką jak trwały harmonogram zadań, aby utworzyć agentów, którzy:

  • Utrwalanie stanu automatycznie między żądaniami i wykonywaniem procesów roboczych
  • Wznawianie po awariach bez utraty kontekstu konwersacji lub powtarzania ukończonej pracy
  • Skalowanie między rozproszonymi, bezstanowymi procesami roboczymi na podstawie zapotrzebowania
  • Organizowanie przepływów pracy z wieloma agentami przy użyciu niezawodnych gwarancji wykonania
  • Przepływy pracy programu Checkpoint Agent Framework utworzone za pomocą modelu przepływu pracy opartego na grafach
  • Wstrzymywanie danych wejściowych lub zewnętrznych bez korzystania z tokenów obliczeniowych lub tokenów modelu podczas oczekiwania
  • Wznawianie dostarczania przesyłanego strumieniowo po skonfigurowaniu za pomocą brokera strumienia, takiego jak Redis; potwierdzanie trwałego ukończenia oddzielnie
  • Zarządzanie cyklem życia sesji za pomocą oczyszczania czasu wygaśnięcia sesji i monitorowania opartego na pulpicie nawigacyjnym

Kiedy należy używać agentów trwałych

Wybierz agentów trwałych, gdy potrzebujesz:

  • Stan konwersacji trwałej: sesje agenta przetrwają awarie procesu, ponowne uruchomienia i zdarzenia skalowania w poziomie
  • Złożone aranżacje: Koordynuj wielu agentów przy użyciu deterministycznych, niezawodnych przepływów pracy, które mogą być uruchamiane przez dni lub tygodnie
  • Orkiestracja sterowana zdarzeniami: integracja z wyzwalaczami, kolejkami, elementami webhook, czasomierzami lub istniejącymi zdarzeniami aplikacji
  • Stan konwersacji automatycznej: historia konwersacji agenta jest automatycznie zarządzana i utrwalana bez konieczności jawnej obsługi stanu w kodzie
  • Durable Przepływy pracy platformy agentów: Tworzenie przepływów pracy struktury agentów opartych na grafach Microsoft, dzięki czemu każdy krok można sprawdzić i wznowić
  • Długotrwałe sesje: automatyczne usuwanie sesji bezczynnych sesji przy użyciu czyszczenia czasu wygaśnięcia (TTL) sesji bezczynności
  • Odpowiedzi w czasie rzeczywistym: dostarczaj dane wyjściowe tokenu przyrostowego za pomocą ponownego połączenia opartego na brokerze, sprawdzając trwały wynik oddzielnie

Takie podejście hostingu różni się od hostingu agenta opartego na usłudze zarządzanej (na przykład usługi agenta usługi Foundry), który zapewnia w pełni zarządzaną infrastrukturę bez konieczności wdrażania hostów procesów roboczych ani zarządzania nimi. Agenci trwałe są idealnym rozwiązaniem, gdy potrzebujesz elastyczności wdrożenia opartego na kodzie w połączeniu z trwałym zarządzaniem stanem.

Wybieranie modelu hostingu

Model hostingu Wybierz ją, gdy potrzebujesz
Azure Functions Zarządzany, bezserwerowy model hostingu; wbudowane skalowanie w poziomie i skalowanie do zera; Azure Functions wyzwalacze i powiązania; Punkty końcowe HTTP generowane przez model programowania usługi Functions; wyzwalacz serwera MCP; oraz minimalne zarządzanie infrastrukturą hosta.
Bring-your-own-compute/ self-hosted Większa kontrola nad procesem hosta, środowiskiem wdrażania, cyklem życia środowiska uruchomieniowego, infrastrukturą, siecią, uwierzytelnianiem lub integracją z istniejącą aplikacją lub usługą. Ten model jest używany w przypadku kontenerów, platformy Kubernetes, długotrwałych procesów roboczych, aplikacji konsolowych, usług niestandardowych lub środowisk hostingowych innych niż funkcje.

W przypadku hostowania w ramach planu hostingu Flex Consumption usługi Azure Functions agenci mogą skalować do tysięcy wystąpień lub do zera, gdy nie są używane, co pozwala płacić tylko za potrzebne zasoby obliczeniowe. W scenariuszach samodzielnie host kontroluje okres istnienia procesu, skalowanie, sieć i wdrażanie.

Rozpoczęcie pracy

Użyj poleceń instalacyjnych dla modelu hostingu poniżej. Zachowaj opcje wersji wstępnej wyświetlane dla pakietów w wersji zapoznawczej. Aby uzyskać pełne aplikacje, zobacz źródło i przykłady.

Dokumentacja gałęzi źródłowej może opisywać możliwości, które nie są jeszcze dostępne w opublikowanych pakietach. Zastosuj kroki konfiguracji i migracji tylko dla funkcji uwzględnionych w zainstalowanej wersji.

Important

Przed uaktualnieniem istniejącego wdrożenia przejrzyj migrację wdrożenia i schematu. Nie należy wskazywać uaktualnionego procesu roboczego w starych historiach przepływu pracy bez sprawdzania zgodności odtwarzania.

W projekcie .NET wybierz zestaw pakietów dla modelu hostingu.

W przypadku hostowania Azure Functions dodaj pakiet integracji Azure Functions i pakiety procesów roboczych usługi Functions.

dotnet add package Microsoft.Agents.AI.Foundry --prerelease
dotnet add package Microsoft.Agents.AI.Hosting.AzureFunctions --prerelease

Note

Użyj rozszerzenia Procesu roboczego usługi Functions, rozszerzenia procesu roboczego DurableTask i rozszerzenia procesu roboczego AzureManaged zadeklarowanych przez zależności rozszerzenia agenta. Usuń odwołania do pakietów bezpośrednich, które obniżają te zależności. Zapoznaj się z odwołaniami .NET zależności. Uaktualnienia schematu równoległego w wersji 2 również wymagają filtrowania elementów roboczych; zobacz wdrażanie i migracja schematu.

W przypadku hostingu typu bring-your-own-compute dodaj podstawowy pakiet integracji rozszerzenia Durable Task oraz pakiety robocze/klienckie narzędzia Durable Task Scheduler używane przez hosta:

dotnet add package Microsoft.Agents.AI.Foundry --prerelease
dotnet add package Microsoft.Agents.AI.DurableTask --prerelease
dotnet add package Microsoft.DurableTask.Client.AzureManaged
dotnet add package Microsoft.DurableTask.Worker.AzureManaged
dotnet add package Microsoft.Extensions.Hosting

W projekcie Python wybierz pakiet dla modelu hostingu. Python hosty wymagają jawnej strategii wdrażania.

W przypadku hostingu Azure Functions użyj Python 3.13 lub nowszej i zainstaluj pakiet integracji Azure Functions. Następujące polecenia instalują również klienta Azure OpenAI używanego w tym przewodniku.

pip install agent-framework-azurefunctions agent-framework-openai azure-identity --pre

W przypadku hostingu typu bring-your-own-compute użyj Python 3.10 lub nowszej wersji i zainstaluj pakiet integracji rozszerzenia Durable Task.

pip install agent-framework-durabletask agent-framework-openai azure-identity --pre

W przypadku nowego izolowanego centrum zadań z uaktualnionymi procesami roboczymi i klientami ustaw przed DURABLE_AGENTS_DEPLOYMENT_MODE=isolated_v2 uruchomieniem jednego z hostów. W przypadku istniejącego centrum wybierz strategię migracji zamiast stosować to ustawienie w sposób ślepy. Ustawienie potwierdza izolację; nie wykrywa innych procesów roboczych ani nie jest zgodny ze starymi historiami.

hosting Azure Functions

Rozszerzenie Durable umożliwia wdrażanie i hostowanie agentów Microsoft Agent Framework w Azure Functions z wbudowanymi punktami końcowymi HTTP i wywołaniami opartymi na orkiestracji. Usługa Azure Functions zapewnia oparte na zdarzeniach model cenowy z płatnością za użycie oraz automatyczne skalowanie i minimalne zarządzanie infrastrukturą.

Podczas konfigurowania trwałego agenta w Azure Functions rozszerzenie automatycznie tworzy punkty końcowe HTTP dla agenta i zarządza podstawową infrastrukturą do przechowywania stanu konwersacji, obsługi żądań współbieżnych i koordynowania przepływów pracy z wieloma agentami. Integracja Azure Functions hostingu zapewnia również wygody specyficzne dla funkcji, takie jak wygenerowane interfejsy API REST do wysyłania komunikatów, sprawdzania stanu i zarządzania sesjami, a także wyzwalacze, takie jak wyzwalacz serwera MCP dla agentów hostingu jako serwery MCP bez kleju wyzwalacza zapisu.

using System;
using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Hosting.AzureFunctions;
using Microsoft.Azure.Functions.Worker.Builder;
using Microsoft.Extensions.Hosting;

var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT");
var deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT") ?? "gpt-4o-mini";

// Create an AI agent following the standard Microsoft Agent Framework pattern
AIAgent agent = new AIProjectClient(new Uri(endpoint), new DefaultAzureCredential())
    .AsAIAgent(
        model: deploymentName,
        instructions: "You are good at telling jokes.",
        name: "Joker");

// Configure the function app to host the agent with durable thread management
// This automatically creates HTTP endpoints and manages state persistence
using IHost app = FunctionsApplication
    .CreateBuilder(args)
    .ConfigureFunctionsWebApplication()
    .ConfigureDurableAgents(options =>
        options.AddAIAgent(agent)
    )
    .Build();
app.Run();

Warning

DefaultAzureCredential jest wygodne do programowania, ale wymaga starannego rozważenia w środowisku produkcyjnym. W środowisku produkcyjnym rozważ użycie określonego poświadczenia (np. ManagedIdentityCredential), aby uniknąć problemów z opóźnieniami, niezamierzonego sondowania poświadczeń i potencjalnych zagrożeń bezpieczeństwa wynikających z mechanizmów awaryjnych.

import os
from agent_framework_azurefunctions import AgentFunctionApp
from agent_framework.openai import OpenAIChatCompletionClient
from azure.identity import DefaultAzureCredential

endpoint = os.getenv("AZURE_OPENAI_ENDPOINT")
deployment_name = os.getenv("AZURE_OPENAI_DEPLOYMENT", "gpt-4o-mini")
api_version = os.getenv("AZURE_OPENAI_API_VERSION")

# Create an AI agent following the standard Microsoft Agent Framework pattern
agent = OpenAIChatCompletionClient(
    azure_endpoint=endpoint,
    model=deployment_name,
    api_version=api_version,
    credential=DefaultAzureCredential()
).as_agent(
    instructions="You are good at telling jokes.",
    name="Joker"
)

# Configure the function app to host the agent with durable thread management
# This automatically creates HTTP endpoints and manages state persistence
app = AgentFunctionApp(agents=[agent])

Bring-your-own-compute/self-hosted hosting

Użyj hostingu bring-your-own-compute, jeśli chcesz korzystać z możliwości rozszerzenia Durable Bez korzystania z modelu programowania Azure Functions. W tym modelu proces uruchamia proces roboczy durable task, rejestruje trwałych agentów lub przepływów pracy i nawiązuje połączenie z zapleczem harmonogramu zadań durable. Kod klienta może działać w tym samym procesie lub w oddzielnej usłudze.

Samoobsługowe procesy robocze korzystają z tych samych podstawowych funkcji rozszerzenia Durable Extension, co Azure Functions hostingu: punktów kontrolnych i wznowienia, orkiestracji agentów deterministycznych, trwałych przepływów pracy platformy Agent Framework, czekania w pętli człowieka, niezawodnego przesyłania strumieniowego, oczyszczania bezczynności sesji, widoczności pulpitu nawigacyjnego i rozproszonego wykonywania w wystąpieniach procesów roboczych bezstanowych. Host jest odpowiedzialny za udostępnianie własnych interfejsów API, zarządzania cyklem życia, sieci, uwierzytelniania i modelu wdrażania.

Skonfiguruj hosta przy użyciu podstawowego pakietu integracji rozszerzenia Durable Task. Użyj ConfigureDurableAgents dla trwałych agentów i ConfigureDurableWorkflows dla przepływów pracy platformy agentów Microsoft opartych na grafach.

string connectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING")
    ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None";

IHost host = Host.CreateDefaultBuilder(args)
    .ConfigureServices(services =>
    {
        services.ConfigureDurableAgents(
            options => options.AddAIAgent(agent),
            workerBuilder: builder => builder.UseDurableTaskScheduler(connectionString),
            clientBuilder: builder => builder.UseDurableTaskScheduler(connectionString));
    })
    .Build();

await host.StartAsync();

Zobacz przykłady konsoli .NET Durable Agents i .NET przykładowe konsole Durable Workflows aby zapoznać się z przykładami hostowanymi samodzielnie.

Użyj pakietu integracyjnego Durable Task, aby uruchomić proces roboczy, który rejestruje agentów i nasłuchuje żądań. Kod klienta może łączyć się z tym samym centrum zadań narzędzia Durable Task Scheduler z innego procesu.

from agent_framework_durabletask import DurableAIAgentWorker
from durabletask.azuremanaged.worker import DurableTaskSchedulerWorker

worker = DurableTaskSchedulerWorker(
    host_address="http://localhost:8080",
    secure_channel=False,
    taskhub="default",
)

agent_worker = DurableAIAgentWorker(worker)
agent_worker.add_agent(agent)

worker.start()

Zobacz Python Durable Task samples for worker-client examples, including single-agent hosting, multi-agent routing, reliable streaming, orchestration chaining, concurrency, conditionals, and human-in-the-loop patterns.

Przepływy pracy platformy Durable Agent Framework

Trwałość nie jest ograniczona do trwałych aranżacji. Microsoft przepływy pracy platformy Agent Framework utworzone przy użyciu modelu przepływu pracy opartego na grafach mogą być również trwałe. Wykonywanie przepływu pracy punktów kontrolnych rozszerzenia Durable Extension, więc ukończone kroki wykonawcze i agenta nie są powtarzane po ponownym uruchomieniu lub niepowodzeniu procesu.

Używaj trwałych aranżacji, gdy chcesz koordynować imperatywne z rozgałęzianiami opartymi na kodzie, czasomierzami, działaniami i zdarzeniami zewnętrznymi. Używaj trwałych przepływów pracy platformy Agent Framework, gdy chcesz deklaratywnego grafu funkcji wykonawczych i agentów z typowanym routingiem, fan-out/fan-in, krawędziami warunkowymi, zdarzeniami przepływu pracy, stanem udostępnionym, pod przepływami pracy lub portami żądań pętli human-in-the-loop.

Note

Przepływy pracy platformy Durable Agent Framework różnią się od magazynu punktów kontrolnych w standardowych przepływach pracy. Magazyn punktów kontrolnych ułatwia wznawianie przebiegu przepływu pracy w środowisku uruchomieniowym platformy Agent Framework. Rozszerzenie Durable uruchamia przepływ pracy w infrastrukturze durable task, więc postęp przepływu pracy jest punktami kontrolnymi i odzyskiwanym przez rozproszone trwałe procesy robocze. Aby uzyskać standardowe punkty kontrolne przepływu pracy, zobacz Punkty kontrolne i wznawianie.

Rejestrowanie przepływów pracy opartych na grafach za pomocą ConfigureDurableWorkflows dla własnych aplikacji lub ConfigureDurableWorkflows w konstruktorze aplikacji usługi Functions na potrzeby hostingu Azure Functions.

Zobacz przykłady .NET Durable Workflows Azure Functions przykłady i .NET przykłady konsoli durable workflows.

Trwałe przykłady przepływów pracy są dostępne dla hostowania Azure Functions, w tym stanu współużytkowanego, wykonywania równoległego przepływu pracy i przepływów pracy pętli human-in-the-loop. Python obsługuje również przepływy pracy oparte na grafach w autonomicznym roboczym narzędziu Durable Task.

Zobacz przykłady Python Azure Functions dotyczące agenta trwałego, orkiestracji, serwera MCP i przepływu pracy.

Aby zapoznać się z własnym wykonywaniem grafu, zobacz przykład autonomicznego przepływu pracy Python.

Samples

Język Model hostingu Trwałe agenty i aranżacje Przepływy pracy oparte na grafach
C# Azure Functions przykłady agentów .NET przykłady przepływu pracy .NET
C# Self-hosted przykłady agenta konsoli .NET Przykłady przepływu pracy konsoli .NET
Python Azure Functions przykłady usługi Python Functions przykłady przepływu pracy usługi Python Functions
Python Self-hosted przykłady Python procesów roboczych i klientów Python autonomiczny przepływ pracy

Te przykłady pokazują obsługę, a nie identyczne kontrakty interfejsu API w różnych językach. Funkcje generują punkty końcowe HTTP; Aplikacje hostowane samodzielnie implementują i zabezpieczają własne interfejsy API.

Wątki agenta stanowego z historią konwersacji

Agenci utrzymują trwałe sesje, które przetrwają w wielu interakcjach. Każda sesja ma unikatowy identyfikator sesji i przechowuje stan konwersacji w trwałym magazynie zarządzanym przez infrastrukturę durable task, taką jak harmonogram zadań Durable.

Ten wzorzec umożliwia ciągłość konwersacji, w której stan agenta jest zachowywany przez awarie i ponowne uruchamianie procesu, dzięki czemu pełna historia konwersacji będzie utrzymywana w wątkach użytkownika. Trwały magazyn gwarantuje, że nawet jeśli proces hosta zostanie uruchomiony ponownie lub wznowi pracę w innym wystąpieniu procesu roboczego, konwersacja będzie bezproblemowo kontynuowana od miejsca, w którym została przerwana.

Skonfiguruj oczyszczanie sesji, przechowywanie historii i przechowywanie wyników celowo. Te zasady mają różne skutki; zobacz cykl życia stanu i pojemność.

Poniższy przykład Azure Functions pokazuje wiele żądań HTTP do tej samej sesji. Obie integracje języka zwracają x-ms-session-id i akceptują kanoniczny sessionId parametr zapytania. Traktuj zwrócony identyfikator jako nieprzezroczystym: w przykładzie pokazano adres jednostki .NET schema-v2, a formatowanie sesji Python zależy od strategii wdrażania. W aplikacjach hostowanych samodzielnie użyj interfejsów API klienta Durable Task z własnego procesu lub usługi.

# First interaction - start a new session
curl -i -X POST https://your-function-app.azurewebsites.net/api/agents/Joker/run \
  -H "Content-Type: text/plain" \
  -d "Tell me a joke about pirates"

# Response includes session ID in x-ms-session-id header and joke as plain text
# HTTP/1.1 200 OK
# Content-Type: text/plain
# x-ms-session-id: @dafx-v2-joker@263fa373-fa01-4705-abf2-5a114c2bb87d
#
# Why don't pirates shower before they walk the plank? Because they'll just wash up on shore later!

# Second interaction - reuse the exact session ID returned by the first request
curl -X POST "https://your-function-app.azurewebsites.net/api/agents/Joker/run?sessionId=@dafx-v2-joker@263fa373-fa01-4705-abf2-5a114c2bb87d" \
  -H "Content-Type: text/plain" \
  -d "Tell me another one about the same topic"

# Agent remembers the pirate context from the first message and responds with plain text
# What's a pirate's favorite letter? You'd think it's R, but it's actually the C!

Stan agenta jest utrzymywany w trwałej pamięci, co umożliwia rozproszone wykonywanie w wielu instancjach. Każda instancja może wznowić wykonywanie agenta po przerwach lub awariach, zapewniając ciągłe działanie.

Użyj poleceń sessionId i waitForResponse w nowych klientach HTTP agenta. thread_id jest przestarzałym aliasem żądania, a nie nagłówkiem odpowiedzi do przechwycenia. Aby uzyskać Python odpowiedzi JSON, przeczytaj sessionId i correlationId. Zobacz Przewodnik migracji interfejsu API HTTP , aby uzyskać aliasy i obsługę konfliktów.

Potwierdzanie trwałego ukończenia

W przypadku wygenerowanych .NET punktów końcowych agenta usługi Functions zinterpretuj uzupełnianie w następujący sposób:

Kod statusu HTTP Meaning
202 Accepted Wysyłanie jest akceptowane. Wykonanie nie zostało potwierdzone pomyślnie.
200 OK Dostępny jest zatwierdzony wynik zakończony powodzeniem.
500 Internal Server Error Zwracany jest obsługiwany zatwierdzony błąd terminalu. Rozróżnianie tego od zwykłych błędów hosta lub transportu.
410 Gone Ukończenie jest znane, ale jego wynik jest niedostępny. Zachowany wynik może być succeeded lub failed; niekoniecznie jest to nieudane wykonanie lub tylko wygaśnięcie wyniku.

Odczyt x-ms-durable-outcome , gdy ukończenie nie ma dostępnego wyniku. Nie traktuj 410 jako oczekujących, wywnioskuj wynik z historii konwersacji lub automatycznie przesyłaj ponownie nową tożsamość. Wywołania JSON używają ciągów status i liczb ; statusCodewcześniejsze status_code pole jest tymczasowym aliasem zgodności.

Python Functions rozróżnia również dostępne wyniki powodzenia, zachowane wyniki zakończone niepowodzeniem i niedostępne. Jego koperty błędów i opcje zestawu SDK nie są identyczne z .NET. Zobacz kontrakt ukończenia .NET i Python kontrakt HTTP.

Te stany opisują punkty końcowe agenta. Interfejs API zarządzania orkiestracją Durable Functions ma własny kontrakt odpowiedzi na stan i sondowania.

Niezawodne przesyłanie strumieniowe

Rozszerzenie Durable Obsługuje niezawodne przesyłanie strumieniowe dla aplikacji, które wymagają dostarczania tokenów w czasie rzeczywistym wraz z trwałym wykonywaniem agenta. Przesyłanie strumieniowe może być używane z rozszerzeniem podstawowym w obu modelach hostingu, ale hosty rozproszone wymagają niezawodnego brokera strumienia, takiego jak Redis, dzięki czemu klienci mogą wznowić odczytywanie zachowanych strumieni tokenów po ponownym połączeniu lub zmianach procesów roboczych, z zastrzeżeniem ustawień trwałości i przechowywania brokera.

Używaj niezawodnego przesyłania strumieniowego, gdy środowisko użytkownika zależy od odpowiedzi przyrostowych, ale obciążenie nadal wymaga trwałych semantyki wykonywania. Ukończenie strumienia nie potwierdza, że wynik i stan agenta były trwale utrwalane; sprawdź oddzielnie trwały wynik autorytatywne. Aby zapoznać się z przykładami możliwych do uruchomienia, zobacz przykłady Python Durable Task które obejmują niezawodne wzorce przesyłania strumieniowego.

Deterministyczne aranżacje wielu agentów

Rozszerzenie Durable obsługuje tworzenie deterministycznych przepływów pracy, które koordynują wielu agentów przy użyciu aranżacji durable task. W Azure Functions używają one orkiestracji Durable Functions; w hostach typu bring-your-own-compute są uruchamiane za pośrednictwem skonfigurowanego procesu roboczego rozszerzenia Durable Task i klienta.

Orkiestracje to oparte na kodzie przepływy pracy, które koordynują wiele operacji (takich jak wywołania agentów, zewnętrzne wywołania interfejsu API lub czasomierze) w niezawodny sposób. Deterministyczny oznacza, że kod orkiestracji jest wykonywany w taki sam sposób, gdy odtwarzany po awarii, co sprawia, że przepływy pracy są niezawodne i debugowane — podczas odtwarzania historii orkiestracji można zobaczyć dokładnie, co się stało w każdym kroku.

Orkiestracje są wykonywane niezawodnie, przetrzymując awarie między wywołaniami agentów, i zapewniają przewidywalne oraz powtarzalne procesy. To sprawia, że są idealne w przypadku złożonych scenariuszy obejmujących wiele agentów, w których potrzebna jest gwarantowana kolejność wykonywania i odporność na uszkodzenia.

Trwałe tworzenie punktów kontrolnych nie powoduje, że narzędzia ani zewnętrzne operacje zapisu są transakcyjne ani dokładnie jednokrotne. Utwórz działanie i efekty uboczne narzędzia idempotentne i zdefiniuj granice ponawiania. Niepowodzenie po zapisie zewnętrznym, ale przed trwałym zatwierdzeniem może spowodować ponowne uruchomienie operacji.

Sekwencyjne aranżacje

W sekwencyjnym schemacie wielu agentów wyspecjalizowani agenci wykonują działania w określonej kolejności, gdzie wyniki każdego agenta mogą mieć wpływ na wykonanie następnego agenta. Ten wzorzec obsługuje logikę warunkową i rozgałęzianie na podstawie odpowiedzi agenta.

W przypadku korzystania z agentów w orkiestracjach należy użyć context.GetAgent() API, aby uzyskać wystąpienie DurableAIAgent, które jest specjalną podklasą typu standardowego AIAgent, opakowującą jednego z zarejestrowanych agentów. Opakowanie DurableAIAgent zapewnia, że wywołania agentów są właściwie śledzone i kontrolowane przez trwały framework orkiestracji.

using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;
using Microsoft.Agents.AI.DurableTask;

[Function(nameof(SpamDetectionOrchestration))]
public static async Task<string> SpamDetectionOrchestration(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    Email email = context.GetInput<Email>();

    // Check if the email is spam
    DurableAIAgent spamDetectionAgent = context.GetAgent("SpamDetectionAgent");
    AgentSession spamSession = await spamDetectionAgent.CreateSessionAsync();

    AgentResponse<DetectionResult> spamDetectionResponse = await spamDetectionAgent.RunAsync<DetectionResult>(
        message: $"Analyze this email for spam: {email.EmailContent}",
        session: spamSession);
    DetectionResult result = spamDetectionResponse.Result;

    if (result.IsSpam)
    {
        return await context.CallActivityAsync<string>(nameof(HandleSpamEmail), result.Reason);
    }

    // Generate response for legitimate email
    DurableAIAgent emailAssistantAgent = context.GetAgent("EmailAssistantAgent");
    AgentSession emailSession = await emailAssistantAgent.CreateSessionAsync();

    AgentResponse<EmailResponse> emailAssistantResponse = await emailAssistantAgent.RunAsync<EmailResponse>(
        message: $"Draft a professional response to: {email.EmailContent}",
        session: emailSession);

    return await context.CallActivityAsync<string>(nameof(SendEmail), emailAssistantResponse.Result.Response);
}

W przypadku używania agentów w orkiestracjach należy użyć app.get_agent() metody, aby uzyskać trwałe wystąpienie agenta, które jest specjalną otoką wokół jednego z zarejestrowanych agentów. Otoczka trwałego agenta gwarantuje, że wywołania agentów są prawidłowo śledzone i sprawdzane przez trwałe środowisko orkiestracyjne.

from durabletask.task import OrchestrationContext
from agent_framework_azurefunctions import AgentFunctionApp
from pydantic import BaseModel

class SpamDetectionResult(BaseModel):
    is_spam: bool
    reason: str

class EmailResponse(BaseModel):
    response: str

app = AgentFunctionApp(agents=[spam_detection_agent, email_assistant_agent])

@app.orchestration_trigger(context_name="context")
def spam_detection_orchestration(context: OrchestrationContext, email: dict):
    # Check if the email is spam
    spam_agent = app.get_agent(context, "SpamDetectionAgent")
    spam_thread = spam_agent.create_session()

    spam_result_raw = yield spam_agent.run(
        messages=f"Analyze this email for spam: {email['content']}",
        session=spam_thread,
        options={"response_format": SpamDetectionResult},
    )
    spam_result = SpamDetectionResult.model_validate(spam_result_raw.value)

    if spam_result.is_spam:
        result = yield context.call_activity("handle_spam_email", input=spam_result.reason)
        return result

    # Generate response for legitimate email
    email_agent = app.get_agent(context, "EmailAssistantAgent")
    email_thread = email_agent.create_session()

    email_response_raw = yield email_agent.run(
        messages=f"Draft a professional response to: {email['content']}",
        session=email_thread,
        options={"response_format": EmailResponse},
    )
    email_response = EmailResponse.model_validate(email_response_raw.value)

    result = yield context.call_activity("send_email", input=email_response.response)
    return result

Orkiestracje koordynują pracę pomiędzy wieloma agentami, zarządzając błędami podczas wywołań agentów. Kontekst orkiestracji udostępnia metody pobierania i interakcji z hostowanymi agentami w ramach orkiestracji.

Aranżacje równoległe

W równoległym wzorcu wielu agentów, wykonuje się wielu agentów jednocześnie, a następnie po wykonaniu agreguje ich wyniki. Ten wzorzec jest przydatny do jednoczesnego zbierania różnych perspektyw lub przetwarzania niezależnych podzadań.

using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;
using Microsoft.Agents.AI.DurableTask;

[Function(nameof(ResearchOrchestration))]
public static async Task<string> ResearchOrchestration(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    string topic = context.GetInput<string>();

    // Execute multiple research agents in parallel
    DurableAIAgent technicalAgent = context.GetAgent("TechnicalResearchAgent");
    DurableAIAgent marketAgent = context.GetAgent("MarketResearchAgent");
    DurableAIAgent competitorAgent = context.GetAgent("CompetitorResearchAgent");

    // Start all agent runs concurrently
    Task<AgentResponse<TextResponse>> technicalTask = 
        technicalAgent.RunAsync<TextResponse>($"Research technical aspects of {topic}");
    Task<AgentResponse<TextResponse>> marketTask = 
        marketAgent.RunAsync<TextResponse>($"Research market trends for {topic}");
    Task<AgentResponse<TextResponse>> competitorTask = 
        competitorAgent.RunAsync<TextResponse>($"Research competitors in {topic}");

    // Wait for all tasks to complete
    await Task.WhenAll(technicalTask, marketTask, competitorTask);

    // Aggregate results
    string allResearch = string.Join("\n\n", 
        technicalTask.Result.Result.Text,
        marketTask.Result.Result.Text,
        competitorTask.Result.Result.Text);

    DurableAIAgent summaryAgent = context.GetAgent("SummaryAgent");
    AgentResponse<TextResponse> summaryResponse = 
        await summaryAgent.RunAsync<TextResponse>($"Summarize this research:\n{allResearch}");

    return summaryResponse.Result.Text;
}

from durabletask.task import OrchestrationContext, when_all
from agent_framework_azurefunctions import AgentFunctionApp

app = AgentFunctionApp(agents=[technical_agent, market_agent, competitor_agent, summary_agent])

@app.orchestration_trigger(context_name="context")
def research_orchestration(context: OrchestrationContext, topic: str):
    # Execute multiple research agents in parallel
    technical_agent = app.get_agent(context, "TechnicalResearchAgent")
    market_agent = app.get_agent(context, "MarketResearchAgent")
    competitor_agent = app.get_agent(context, "CompetitorResearchAgent")

    technical_task = technical_agent.run(messages=f"Research technical aspects of {topic}")
    market_task = market_agent.run(messages=f"Research market trends for {topic}")
    competitor_task = competitor_agent.run(messages=f"Research competitors in {topic}")

    # Wait for all tasks to complete
    results = yield when_all([technical_task, market_task, competitor_task])

    # Aggregate results
    all_research = "\n\n".join(r.text for r in results)

    summary_agent = app.get_agent(context, "SummaryAgent")
    summary = yield summary_agent.run(messages=f"Summarize this research:\n{all_research}")

    return summary.text

Wykonywanie równoległe jest śledzone przy użyciu listy zadań. Automatyczna punktacja kontrolna zapewnia, że ukończone zadania agenta nie są powtarzane ani tracone, jeśli wystąpi awaria podczas agregacji.

Orkiestracje pętli "człowiek w pętli"

Orkiestracje agentów deterministycznych mogą być wstrzymane, oczekując na dane wejściowe, zatwierdzenie lub przegląd, bez zużywania zasobów obliczeniowych. Trwałe wykonywanie umożliwia orkiestracjom czekać dniami lub nawet tygodniami, oczekując na odpowiedzi od ludzi. W połączeniu z hostingiem bezserwerowym wszystkie zasoby obliczeniowe są wyłączane w okresie oczekiwania, eliminując koszty przetwarzania do momentu, gdy użytkownik nie poda swojego wkładu.

using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;
using Microsoft.Agents.AI.DurableTask;

[Function(nameof(ContentApprovalWorkflow))]
public static async Task<string> ContentApprovalWorkflow(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    string topic = context.GetInput<string>();

    // Generate content using an agent
    DurableAIAgent contentAgent = context.GetAgent("ContentGenerationAgent");
    AgentResponse<GeneratedContent> contentResponse = 
        await contentAgent.RunAsync<GeneratedContent>($"Write an article about {topic}");
    GeneratedContent draftContent = contentResponse.Result;

    // Send for human review
    await context.CallActivityAsync(nameof(NotifyReviewer), draftContent);

    // Wait for approval with timeout
    HumanApprovalResponse approvalResponse;
    try
    {
        approvalResponse = await context.WaitForExternalEvent<HumanApprovalResponse>(
            eventName: "ApprovalDecision",
            timeout: TimeSpan.FromHours(24));
    }
    catch (OperationCanceledException)
    {
        // Timeout occurred - escalate for review
        return await context.CallActivityAsync<string>(nameof(EscalateForReview), draftContent);
    }

    if (approvalResponse.Approved)
    {
        return await context.CallActivityAsync<string>(nameof(PublishContent), draftContent);
    }

    return "Content rejected";
}

from durabletask.task import OrchestrationContext, when_any
from datetime import timedelta
from agent_framework_azurefunctions import AgentFunctionApp

app = AgentFunctionApp(agents=[content_agent])

@app.orchestration_trigger(context_name="context")
def content_approval_workflow(context: OrchestrationContext, topic: str):
    # Generate content using an agent
    content_agent = app.get_agent(context, "ContentGenerationAgent")
    draft_content = yield content_agent.run(
        messages=f"Write an article about {topic}"
    )

    # Send for human review
    yield context.call_activity("notify_reviewer", input=draft_content.text)

    # Wait for approval with timeout
    approval_task = context.wait_for_external_event("ApprovalDecision")
    timeout_task = context.create_timer(
        context.current_utc_datetime + timedelta(hours=24)
    )

    winner = yield when_any([approval_task, timeout_task])

    if winner == approval_task:
        timeout_task.cancel()
        approval_data = approval_task.result
        if approval_data.get("approved"):
            result = yield context.call_activity("publish_content", input=draft_content.text)
            return result
        return "Content rejected"

    # Timeout occurred - escalate for review
    result = yield context.call_activity("escalate_for_review", input=draft_content.text)
    return result

Deterministyczne orkiestracje agentów mogą czekać na zdarzenia zewnętrzne, trwale utrwalając swój stan podczas oczekiwania na informacje zwrotne od ludzi, przetrwanie niepowodzeń, ponowne uruchomienia i dłuższe okresy oczekiwania. Po nadejściu odpowiedzi człowieka orkiestracja zostanie automatycznie wznowiona z pełnym kontekstem konwersacji i stanem wykonywania bez zmian.

Dostarczanie danych wejściowych przez człowieka

Aby wysłać zatwierdzenie lub dane wejściowe do oczekującej aranżacji, zgłoś zdarzenie zewnętrzne do wystąpienia aranżacji przy użyciu zestawu SDK klienta rozszerzenia Durable Task lub punktów końcowych rozszerzenia Durable Azure Functions. Na przykład recenzent może zatwierdzić zawartość za pośrednictwem formularza internetowego, który wywołuje:

await client.RaiseEventAsync(instanceId, "ApprovalDecision", new HumanApprovalResponse 
{ 
    Approved = true,
    Feedback = "Looks great!"
});

approval_data = {
    "approved": True,
    "feedback": "Looks great!"
}
await client.raise_event(instance_id, "ApprovalDecision", approval_data)

Efektywność kosztowa

Przepływy pracy typu human-in-the-loop z trwałymi agentami są niezwykle ekonomiczne, gdy są hostowane w planie Flex Consumption usługi Azure Functions. W przypadku przepływu pracy czekającego 24 godziny na zatwierdzenie płacisz tylko za kilka sekund czasu wykonania (czas generowania zawartości, wysyłania powiadomień i przetwarzania odpowiedzi) — a nie za 24 godziny oczekiwania. W okresie oczekiwania nie są używane żadne zasoby obliczeniowe.

Możliwość obserwowania z trwałym harmonogramem zadań

Durable Task Scheduler (DTS) jest zalecanym trwałym zapleczem dla Twoich długotrwałych agentów, oferując najlepszą wydajność, w pełni zarządzaną infrastrukturę i wbudowaną funkcję monitorowania za pośrednictwem panelu UI. Azure Functions aplikacje mogą używać innych zapleczy magazynu (takich jak Azure Storage), ale usługa DTS jest zoptymalizowana specjalnie pod kątem trwałych obciążeń i zapewnia lepszą wydajność i możliwości monitorowania. Samoobsługowe procesy robocze używają również usług DTS do trwałego planowania, stanu i widoczności pulpitu nawigacyjnego.

Szczegółowe informacje o sesji agenta

  • Historia konwersacji: wyświetlanie pełnej historii czatów dla każdej sesji agenta, w tym wszystkich wiadomości, wywołań narzędzi i kontekstu konwersacji w dowolnym momencie w czasie
  • Chronometraż zadania: Monitorowanie czasu wykonywania określonych zadań i interakcji agenta

Zrzut ekranu przedstawiający pulpit nawigacyjny narzędzia Durable Task Scheduler przedstawiający historię czatów agenta z wątkami konwersacji i wiadomościami.

Szczegółowe informacje dotyczące orkiestracji

  • Wizualizacja wielu agentów: zobacz przepływ wykonywania podczas wywoływania wielu wyspecjalizowanych agentów z wizualną reprezentacją równoległych wykonań i rozgałęziania warunkowego
  • Historia wykonywania: uzyskiwanie dostępu do szczegółowych dzienników wykonywania
  • Monitorowanie w czasie rzeczywistym: Śledź aktywne orkiestracje, zlecenia robocze w kolejce i stany agentów w całym wdrożeniu.
  • Metryki wydajności: monitorowanie czasów odpowiedzi agenta, użycia tokenu i czasu trwania aranżacji

Zrzut ekranu dashboardu Durable Task Scheduler pokazujący wizualizację orkiestracji z wieloma interakcjami agentów oraz wykonaniem workflowów.

Możliwości debugowania

  • Wyświetlanie danych wyjściowych agenta strukturalnego i wyników wywołania narzędzia
  • Wywołania narzędzi śledzenia i ich wyniki
  • Monitorowanie zewnętrznej obsługi zdarzeń dla scenariuszy z udziałem człowieka w procesie

Pulpit nawigacyjny umożliwia zrozumienie dokładnie tego, co robią agenci, szybkie diagnozowanie problemów i optymalizowanie wydajności na podstawie rzeczywistych danych wykonywania.

Wdrażanie i migracja schematu

Stan agenta schema-v2 zmienia zarówno stan utrwalonego, jak i tożsamości wysyłania. Planowanie współistnienia procesu roboczego niezależnie od konwersji stanu.

Nowe sesje używają dafx-v2- przestrzeni nazw jednostki. Podczas wdrażania dwufazowego pozostaw ProcessLegacyEntities wyłączone, podczas gdy pracownicy w wersji 1 pozostają aktywni. Zatrzymaj wszystkich procesów roboczych w wersji 1 przed włączeniem ProcessLegacyEntities przetwarzania i uaktualniania kwalifikujących się starszych jednostek.

Autonomiczni pracownicy rejestrują filtry elementów roboczych. Wdrożenia funkcji równoległych wymagają również zgodnego procesu roboczego usługi Functions, rozszerzenia procesu roboczego DurableTask i pakietów rozszerzeń procesu roboczego AzureManaged, które obsługują filtrowanie po stronie hosta na każdym uczestniczącym hoście oraz workItemFilteringEnabled w konfiguracji dostawcy magazynu harmonogramu. Bez zgodnego filtrowania należy użyć trybu jednorazowego trybu offline zamiast współbieżnych hostów w wersji 1 i 2 w jednym koncentratoście. Postępuj zgodnie z przewodnikiem .NET uaktualniania, aby uzyskać pełną konfigurację.

Wybierz jawną strategię przed utworzeniem hosta:

  • two_phase: Użyj narzędzia Durable Task Scheduler z zgodnym filtrowaniem elementów roboczych zaplecza. Nowe nazwy wersji izolować wysyłanie, gdy kwalifikujące się starsze jednostki pozostają w starych pracownikach. Zatrzymaj starych procesów roboczych przed włączeniem przetwarzania nazw historycznych i uaktualnień jednostek w miejscu. Obsługa zgodnych filtrowanych procesów roboczych podczas współistnienia; filtry nazw nie są trwałymi rezerwacjami, jeśli ta flota się rozłączy.
  • isolated_v2: Użyj izolowanego centrum zadań z pasującymi uaktualnionymi procesami roboczymi i klientami. Ustawienie potwierdza izolację, ale nie sonduje innych procesów roboczych. Użyj nowych wystąpień przepływu pracy.

Nowe centrum nie jest bezwarunkowym wymaganiem dla ścieżki migracji jednostek dwufazowych. Zobacz Python migracji jednostek, aby uzyskać informacje o konfiguracji i wymaganiach wstępnych specyficznych dla hosta.

Warning

Migracja jednostek nie naprawia zgodności powtarzania przepływu pracy. Starsze historie przepływów pracy, w tym starsze historie w wersji 2, nie są obsługiwane przez uaktualnione środowisko uruchomieniowe. Jeśli muszą zakończyć, zachowaj je na swoich oryginalnych pracowników i centrum. Uruchom nowe wystąpienia po uaktualnieniu; nie polegaj tylko na znaczniku wersji przepływu pracy, aby udowodnić zgodność. Zobacz Python zgodność środowiska uruchomieniowego.

Cykl życia i pojemność stanu

Wybierz zasady dla każdej warstwy niezależnie:

Warstwa Co kontroluje
Czas wygaśnięcia sesji Usuwa całą jednostkę, w tym stan konwersacji i potwierdzenia ukończenia.
Przechowywanie historii trwałej Usuwa kwalifikowaną przechowywaną zawartość transkrypcji; stan chroniony może nadal przekraczać pojemność.
Kompaktowanie danych wejściowych modelu Zmienia kontekst wysyłany do modelu; niekoniecznie przycina trwałą historię.
Przechowywanie wyników Steruje dostępnością ładunku odpowiedzi niezależnie od zachowanego wyniku ukończenia.
Cykl życia ładunku obiektu blob Kontroluje odciążone obiekty blob ładunku zestawu SDK; usunięcie sesji lub przeczyszczenie aranżacji nie powoduje wyczyszczenia ich.

W .NET schema-v2 stan potwierdzenia nieskonfigurowany starszy 14-dniowy czas wygaśnięcia nie usuwa automatycznie jednostki. Skonfiguruj jawnie globalny lub dla agenta czas wygaśnięcia, jeśli chcesz wyczyścić sesję bezczynności. Przekazywanie timeToLive: null do AddAIAgent dziedziczy skonfigurowaną wartość domyślną globalną; nie zastępuje siedmiodniowej wartości domyślnej globalnej z wartością "nigdy nie wygasa". Ustaw DefaultTimeToLive = null wartość , aby wyłączyć globalny czas wygaśnięcia dla agentów bez dodatnich przesłonięć dla poszczególnych agentów.

.NET wartości domyślne na DurableAgentHistoryRetentionMode.KeepAll; Auto umożliwia przechowywanie transkrypcji na podstawie ciśnienia. ResultRetentionPeriod wartość domyślna to 60 sekund i jest oddzielona od czasu wygaśnięcia sesji. Python ma różne opcje przechowywania i budżetu, w tym retention="keep_all" domyślnie. Nie zakładaj tych samych opcji ani wartości domyślnych na hostach.

Python obsługuje opcjonalne ładowanie obiektów blob zarządzanych przez zestaw SDK w obu modelach hostingu. Skonfiguruj zarówno magazyny producentów, jak i konsumentów, poświadczenia i cykl życia obiektu blob. Odciążanie nie powoduje usunięcia limitów pojemności całej jednostki ani zastąpienia przechowywania; zachowaj obiekty blob, gdy są potrzebne informacje o stanie odwołania lub historii odtwarzania. Zobacz .NET stan i przechowywanie, przechowywanie Python i budżety oraz Python odciążanie obiektów blob usługi Functions.

Aby uzyskać wskazówki dotyczące infrastruktury, pojemności i operacji harmonogramu, zobacz Durable Task integration for Agent Framework (Integracja zadań trwałych dla struktury agentów).

Samouczek: tworzenie i uruchamianie trwałego agenta za pomocą Azure Functions

W tym samouczku pokazano, jak utworzyć i uruchomić trwałego agenta sztucznej inteligencji przy użyciu modelu hostingu Azure Functions rozszerzenia Durable Extension. Utworzysz aplikację usługi Azure Functions, która hostuje agenta stanowego z wbudowanymi punktami końcowymi HTTP i dowiesz się, jak monitorować ją przy użyciu pulpitu nawigacyjnego narzędzia Durable Task Scheduler. Aby zapoznać się z własnymi agentami, zobacz przykłady.

Prerequisites

Przed rozpoczęciem upewnij się, że masz następujące wymagania wstępne:

Note

Program Microsoft Agent Framework jest obsługiwany ze wszystkimi aktywnie obsługiwanymi wersjami platformy .NET. Na potrzeby tego przykładu zalecamy zestaw .NET 9 SDK lub nowszą wersję.

Ustaw DURABLE_AGENTS_DEPLOYMENT_MODE=isolated_v2 tylko po potwierdzeniu, że centrum samouczków jest izolowane, a wszyscy pracownicy i klienci korzystają z uaktualnionego środowiska uruchomieniowego.

Użyj nowego centrum zadań na potrzeby tego samouczka, niezależnie od istniejących aplikacji trwałych. W przypadku istniejącej aplikacji wykonaj zamiast tego kroki migracji wdrożenia i schematu .

Pobierz projekt Quickstart

Użyj Azure Developer CLI, aby zainicjować nowy projekt z szablonu quickstart dla trwałych agentów.

  1. Utwórz nowy katalog dla projektu i przejdź do niego:

    mkdir MyDurableAgent
cd MyDurableAgent

  1. Zainicjuj projekt na podstawie szablonu:

    azd init --template durable-agents-quickstart-dotnet

    Po wyświetleniu monitu o nazwę środowiska wprowadź nazwę, taką jak my-durable-agent.

Spowoduje to pobranie projektu szybkiego startu ze wszystkimi niezbędnymi plikami, w tym konfiguracją usługi Azure Functions, kodem agenta i infrastrukturą jako szablonami kodu.

  1. Utwórz nowy katalog dla projektu i przejdź do niego:

    mkdir MyDurableAgent
cd MyDurableAgent

  1. Zainicjuj projekt na podstawie szablonu:

    azd init --template durable-agents-quickstart-python

    Po wyświetleniu monitu o nazwę środowiska wprowadź nazwę, taką jak my-durable-agent.

  2. Tworzenie i aktywowanie środowiska wirtualnego:

    uv venv --python 3.13 .venv
source .venv/bin/activate

Note

Można również użyć python -m venv .venv z zainstalowanym interpreterem Python 3.13. Jeśli to polecenie zawiesza się na Windows, użyj powyższego uv venv polecenia. Przed zainstalowaniem pakietów sprawdź aktywny interpreter python --version za pomocą polecenia .

  1. Zainstaluj wymagane pakiety:

    Przed zainstalowaniem zaktualizuj szablon requirements.txt do pakietów usługi Functions w temacie Wprowadzenie, w tym agent-framework-openai pakiet klienta. Rozwiąż konfliktowe przypinania zależności w manifeście, a nie wymuszaj ich instalacji.

    uv pip install -r requirements.txt

Spowoduje to pobranie projektu szybkiego startu ze wszystkimi niezbędnymi plikami, w tym konfiguracją usługi Azure Functions, kodem agenta i infrastrukturą jako szablonami kodu. Przygotowuje również środowisko wirtualne z wymaganymi zależnościami.

Aprowizuj zasoby platformy Azure

Użyj interfejsu wiersza polecenia dla deweloperów platformy Azure, aby utworzyć wymagane zasoby platformy Azure dla swojego trwałego agenta.

  1. Skonfiguruj infrastrukturę:

    azd provision

    To polecenie tworzy:

    • Usługa Azure OpenAI z wdrożeniem gpt-4o-mini
    • Aplikacja usługi Azure Functions z planem hostingu Flex Consumption
    • Konto usługi Azure Storage dla środowiska uruchomieniowego usługi Azure Functions i magazynu trwałego
    • Trwałe wystąpienie harmonogramu zadań (plan zużycia) do zarządzania stanem agenta
    • Niezbędne konfiguracje sieci i tożsamości

  2. Po wyświetleniu monitu wybierz subskrypcję platformy Azure i wybierz lokalizację zasobów.

Proces przydzielania zasobów trwa kilka minut. Po zakończeniu usługa azd przechowuje informacje o utworzonym zasobie w środowisku.

Przejrzyj kod agenta

Teraz przyjrzyjmy się kodowi definiującemu trwałego agenta.

Otwórz plik , Program.cs aby wyświetlić konfigurację agenta:

using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Hosting.AzureFunctions;
using Microsoft.Azure.Functions.Worker.Builder;
using Microsoft.Extensions.Hosting;

var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") 
    ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT environment variable is not set");
var deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT") ?? "gpt-4o-mini";

// Create an AI agent following the standard Microsoft Agent Framework pattern
AIAgent agent = new AIProjectClient(new Uri(endpoint), new DefaultAzureCredential())
    .AsAIAgent(
        model: deploymentName,
        instructions: "You are a helpful assistant that can answer questions and provide information.",
        name: "MyDurableAgent");

using IHost app = FunctionsApplication
    .CreateBuilder(args)
    .ConfigureFunctionsWebApplication()
    .ConfigureDurableAgents(options => options.AddAIAgent(agent))
    .Build();
app.Run();

Ten kod:

  1. Pobiera konfigurację usługi Azure OpenAI ze zmiennych środowiskowych.
  2. Tworzy klienta usługi Azure OpenAI przy użyciu poświadczeń platformy Azure.
  3. Tworzy agenta sztucznej inteligencji z instrukcjami i nazwą.
  4. Konfiguruje aplikację usługi Azure Functions do hostowania agenta z trwałym zarządzaniem wątkami.

Otwórz plik , function_app.py aby wyświetlić konfigurację agenta:

import os
from agent_framework_azurefunctions import AgentFunctionApp
from agent_framework.openai import OpenAIChatCompletionClient
from azure.identity import DefaultAzureCredential

endpoint = os.getenv("AZURE_OPENAI_ENDPOINT")
if not endpoint:
    raise ValueError("AZURE_OPENAI_ENDPOINT is not set.")
deployment_name = os.getenv("AZURE_OPENAI_DEPLOYMENT", "gpt-4o-mini")
api_version = os.getenv("AZURE_OPENAI_API_VERSION")

# Create an AI agent following the standard Microsoft Agent Framework pattern
agent = OpenAIChatCompletionClient(
    azure_endpoint=endpoint,
    model=deployment_name,
    api_version=api_version,
    credential=DefaultAzureCredential()
).as_agent(
    instructions="You are a helpful assistant that can answer questions and provide information.",
    name="MyDurableAgent"
)

# Configure the function app to host the agent with durable thread management
app = AgentFunctionApp(agents=[agent])

Ten kod:

  • Pobiera konfigurację usługi Azure OpenAI ze zmiennych środowiskowych.
  • Tworzy klienta usługi Azure OpenAI przy użyciu poświadczeń platformy Azure.
  • Tworzy agenta sztucznej inteligencji z instrukcjami i nazwą.
  • Konfiguruje aplikację usługi Azure Functions do hostowania agenta z trwałym zarządzaniem wątkami.

Agent jest teraz gotowy do hostowania w usłudze Azure Functions. Rozszerzenie trwałego zadania automatycznie tworzy punkty dostępu HTTP na potrzeby współdziałania z agentem i zarządza stanem konwersacji podczas wielu żądań.

Konfigurowanie ustawień lokalnych

local.settings.json Utwórz plik dla lokalnego rozwoju na podstawie przykładowego pliku zawartego w projekcie.

  1. Skopiuj przykładowy plik ustawień:

    cp local.settings.sample.json local.settings.json

  1. Uzyskaj punkt końcowy usługi Azure OpenAI z zaaprowizowanych zasobów:

    azd env get-value AZURE_OPENAI_ENDPOINT

  2. Otwórz local.settings.json i zastąp <your-resource-name> wartość AZURE_OPENAI_ENDPOINT punktem końcowym z poprzedniego polecenia.

Twoje local.settings.json powinno wyglądać następująco:

{
  "IsEncrypted": false,
  "Values": {
    // ... other settings ...
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "AZURE_OPENAI_ENDPOINT": "https://your-openai-resource.openai.azure.com",
    "AZURE_OPENAI_DEPLOYMENT": "gpt-4o-mini",
    "DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None",
    "TASKHUB_NAME": "default"
  }
}

Note

Plik local.settings.json jest używany tylko do programowania lokalnego i nie jest wdrażany na platformie Azure. W przypadku wdrożeń produkcyjnych te ustawienia są automatycznie konfigurowane w aplikacji usługi Azure Functions przez szablony infrastruktury.

W przypadku Python dodaj "DURABLE_AGENTS_DEPLOYMENT_MODE": "isolated_v2" element tylko po Values potwierdzeniu, że centrum samouczków jest izolowane. Skonfiguruj równoważne ustawienie aplikacji podczas wdrażania; ustawienia lokalne nie są przekazywane automatycznie.

W przypadku hostingu opartego na harmonogramie użyj tego samego centrum zadań w programie host.json (za pośrednictwem %TASKHUB_NAME%), TASKHUB_NAME, DURABLE_TASK_SCHEDULER_CONNECTION_STRINGemulatora i pulpitu nawigacyjnego. W tym samouczku jest używany element default, który emulator tworzy automatycznie. Jeśli ten koncentrator zawiera już stare historie lub jest używany przez innych procesów roboczych, wybierz unikatową, nieużywaną nazwę centrum, ustaw nazwę emulatora DTS_TASK_HUB_NAMES i spójnie zaktualizuj ustawienia aplikacji i wybór pulpitu nawigacyjnego. Jedynie zdefiniowanie parametry połączenia harmonogramu nie wybiera dostawcy magazynu harmonogramu.

W host.jsonprogramie skonfiguruj dostawcę harmonogramu w obszarze extensions.durableTask, zachowując inne ustawienia hosta szablonu:

{
  "extensions": {
    "durableTask": {
      "hubName": "%TASKHUB_NAME%",
      "storageProvider": {
        "type": "AzureManaged",
        "connectionStringName": "DURABLE_TASK_SCHEDULER_CONNECTION_STRING"
      }
    }
  }
}

Użyj zależności dostawcy harmonogramu dla języka hostingu. Ta konfiguracja wybiera zaplecze dla nowego izolowanego wdrożenia samouczka; nie włącza filtrowania wymaganego do uaktualnienia równoległego.

Uruchamianie lokalnych zależności programistycznych

Aby uruchomić agentów trwałych lokalnie, należy uruchomić dwie usługi:

  • Azurite: emuluje usługi Azure Storage (używane przez usługę Azure Functions do zarządzania wyzwalaczami i stanem wewnętrznym).
  • Emulator harmonogramu zadań trwałych (DTS): zarządza stanem trwałym (historia konwersacji, stan orkiestracji) oraz planowaniem agentów

W tych krokach założono, że host usługi Functions korzysta z dostawcy magazynu Durable Task Scheduler. W wersji Python Functions przykłady są domyślnie Azure Storage. Jeśli zaczniesz od tych przykładów, wybierz i skonfiguruj dostawcę harmonogramu przed użyciem pulpitu nawigacyjnego usługi DTS; W przeciwnym razie Azurite przechowuje stan trwały, a pulpit nawigacyjny harmonogramu pozostaje pusty. Zobacz Durable Task Scheduler setup (Konfiguracja harmonogramu zadań trwałych).

Rozpocznij Azurite

Usługa Azurite emuluje usługi Azure Storage lokalnie. Usługa Azure Functions używa jej do zarządzania stanem wewnętrznym. Należy to uruchomić w nowym oknie terminalu i zachować jego działanie podczas opracowywania i testowania trwałego agenta.

  1. Otwórz nowe okno terminalu i pobierz obraz Docker Azurite.

    docker pull mcr.microsoft.com/azure-storage/azurite

  2. Uruchom Azurite w oknie terminalu:

    docker run -p 10000:10000 -p 10001:10001 -p 10002:10002 mcr.microsoft.com/azure-storage/azurite

    Usługa Azurite uruchamia się i nasłuchuje na domyślnych portach dla usług Blob (10000), Queue (10001) i Table (10002).

Pozostaw to okno terminalu otwarte podczas opracowywania i testowania trwałego agenta.

Wskazówka

Aby uzyskać więcej informacji na temat platformy Azurite, w tym alternatywnych metod instalacji, zobacz Use Azurite emulator for local Azure Storage development (Używanie emulatora Azurite do lokalnego programowania w usłudze Azure Storage).

Uruchamianie emulatora harmonogramu zadań durable

Emulator DTS zapewnia trwałe zaplecze systemowe do zarządzania stanem agenta i orkiestracjami. Przechowuje historię konwersacji i zapewnia, że stan agenta jest zachowywany po ponownym uruchomieniu systemu. Wyzwala również trwałe aranżacje i agentów. Należy to uruchomić w osobnym nowym oknie terminalu i utrzymać go w działaniu podczas tworzenia i testowania wytrzymałego agenta.

  1. Otwórz kolejne nowe okno terminalu i pobierz obraz Docker emulatora DTS.

    docker pull mcr.microsoft.com/dts/dts-emulator:latest

  2. Uruchom emulator DTS:

    docker run -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest

    To polecenie tworzy default centrum zadań i uruchamia emulator, pasując do ustawień tego samouczka. Polecenie uwidacznia:

    • Port 8080: punkt końcowy gRPC dla harmonogramu zadań trwałych (używany przez aplikację Functions)
    • Port 8082: pulpit nawigacyjny administracyjny

  3. Pulpit nawigacyjny będzie dostępny pod adresem http://localhost:8082.

Pozostaw to okno terminalu otwarte podczas opracowywania i testowania trwałego agenta.

Wskazówka

Aby dowiedzieć się więcej na temat emulatora usługi DTS, w tym sposobu konfigurowania wielu centrów zadań i uzyskiwania dostępu do pulpitu nawigacyjnego, zobacz Develop with Durable Task Scheduler (Programowanie za pomocą harmonogramu zadań trwałych).

Uruchom aplikację funkcji

Teraz możesz uruchomić aplikację usługi Azure Functions z trwałym agentem.

  1. W nowym oknie terminalu (zachowując zarówno Azurite, jak i emulator DTS uruchomiony w oddzielnych oknach), przejdź do katalogu projektu.

  2. Uruchom środowisko uruchomieniowe usługi Azure Functions:

    func start

  3. Powinny pojawić się wyniki wskazujące, że aplikacja funkcji działa, w tym punkty końcowe HTTP dla Twojego agenta.

    Functions:
     http-MyDurableAgent: [POST] http://localhost:7071/api/agents/MyDurableAgent/run
     dafx-v2-MyDurableAgent: entityTrigger

Ilustruje to rejestrację .NET schema-v2; Python nazwy wyzwalaczy zależą od strategii wdrażania. Te punkty końcowe automatycznie zarządzają stanem konwersacji, więc nie trzeba samodzielnie tworzyć obiektów sesji.

Przetestuj agenta lokalnie

Teraz możesz interagować z trwałym agentem przy użyciu żądań HTTP. Agent zachowuje stan rozmowy w trakcie wielu żądań, umożliwiając rozmowy wieloetapowe.

Rozpoczynanie nowej rozmowy

Utwórz nową sesję i wyślij pierwszą wiadomość:

curl -i -X POST http://localhost:7071/api/agents/MyDurableAgent/run \
  -H "Content-Type: text/plain" \
  -d "What are three popular programming languages?"

Przykładowa odpowiedź (x-ms-session-idnagłówek zawiera identyfikator sesji; w tym przykładzie pokazano identyfikator .NET schema-v2):

HTTP/1.1 200 OK
Content-Type: text/plain
x-ms-session-id: @dafx-v2-mydurableagent@263fa373-fa01-4705-abf2-5a114c2bb87d
Content-Length: 189

Three popular programming languages are Python, JavaScript, and Java. Python is known for its simplicity and readability, JavaScript powers web interactivity, and Java is widely used in enterprise applications.

Zapisz dokładny identyfikator sesji z nagłówka x-ms-session-id na potrzeby następnego żądania. Nie konstruuj ani nie zapisujej go ponownie. Python również zwraca ten nagłówek, ale jego format identyfikatora może się różnić.

Kontynuuj konwersację

Wyślij komunikat monitu do tej samej sesji, dołączając zwrócony identyfikator w parametrze zapytania kanonicznego sessionId :

Zastąp przykładowy identyfikator rzeczywistą wartością z nagłówka poprzedniej x-ms-session-id odpowiedzi. W razie potrzeby zakoduj wartość w adresie URL.

curl -X POST "http://localhost:7071/api/agents/MyDurableAgent/run?sessionId=@dafx-v2-mydurableagent@263fa373-fa01-4705-abf2-5a114c2bb87d" \
  -H "Content-Type: text/plain" \
  -d "Which one is best for beginners?"

Przykładowa odpowiedź:

Python is often considered the best choice for beginners among those three. Its clean syntax reads almost like English, making it easier to learn programming concepts without getting overwhelmed by complex syntax. It's also versatile and widely used in education.

Zwróć uwagę, że agent zapamiętuje kontekst z poprzedniego komunikatu (trzy języki programowania) bez konieczności ponownego określania ich. pl-PL: Ponieważ stan konwersacji jest trwale przechowywany przez Harmonogram Zadań Trwałych, ta historia się zachowuje, nawet jeśli ponownie uruchomisz aplikację funkcji lub konwersacja zostanie wznowiona przez inne wystąpienie.

Monitorowanie za pomocą pulpitu nawigacyjnego Durable Task Scheduler

Harmonogram zadań Durable Task Scheduler udostępnia wbudowany pulpit nawigacyjny do monitorowania i debugowania trwałych agentów. Pulpit zapewnia bezpośredni wgląd w operacje agentów, historię konwersacji i przepływ operacji.

Uzyskiwanie dostępu do pulpitu nawigacyjnego

  1. Otwórz pulpit nawigacyjny lokalnego emulatora usługi DTS w http://localhost:8082 przeglądarce internetowej.

  2. Wybierz wartość domyślną lub tę samą unikatową nazwę centrum skonfigurowaną w aplikacji i emulatorze, aby wyświetlić jego szczegóły.

  3. Wybierz ikonę ustawień w prawym górnym rogu, aby otworzyć ustawienia, i upewnij się, że w obszarze Funkcje w wersji zapoznawczej wybrano opcję Włącz strony agenta.

Eksploruj rozmowy agenta

  1. Na pulpicie nawigacyjnym przejdź do karty Agenci .

  2. Z listy wybierz trwały wątek agenta (np. mydurableagent - 263fa373-fa01-4705-abf2-5a114c2bb87d).

    Zobaczysz szczegółowy widok wątku agenta, w tym pełną historię konwersacji ze wszystkimi komunikatami i odpowiedziami.

    Zrzut ekranu pulpitu nawigacyjnego narzędzia Durable Task Scheduler pokazujący historię konwersacji wątku agenta.

Pulpit nawigacyjny zawiera widok osi czasu, który ułatwia zrozumienie przepływu konwersacji. Kluczowe informacje obejmują:

  • Znaczniki czasu i czas trwania każdej interakcji
  • Zawartość monitu i odpowiedzi
  • Liczba użytych tokenów

Wskazówka

Pulpit nawigacyjny usługi DTS zapewnia aktualizacje w czasie rzeczywistym, dzięki czemu możesz obserwować zachowanie agenta podczas interakcji z nim za pośrednictwem punktów końcowych HTTP.

Wdrażanie na platformie Azure

Po przetestowaniu agenta trwałego lokalnie wdróż go na platformie Azure.

  1. Wdróż aplikację:

    azd deploy

    To polecenie pakuje aplikację i wdraża ją w aplikacji usługi Azure Functions utworzonej podczas aprowizacji.

  2. Poczekaj na zakończenie wdrożenia. Dane wyjściowe będą potwierdzać, kiedy agent jest uruchomiony na platformie Azure.

Testowanie wdrożonego agenta

Po wdrożeniu przetestuj agenta uruchomionego na platformie Azure.

Pobierz klucz funkcji

W przypadku domyślnej autoryzacji na poziomie funkcji wygenerowanych punktów końcowych dołącz klucz usługi Functions w żądaniach do Azure. Klucz usługi Functions nie jest autoryzacją użytkownika aplikacji; zobacz zabezpieczenia produkcyjne i rozwiązywanie problemów.

API_KEY=`az functionapp function keys list --name $(azd env get-value AZURE_FUNCTION_NAME) --resource-group $(azd env get-value AZURE_RESOURCE_GROUP) --function-name http-MyDurableAgent --query default -o tsv`

Rozpoczynanie nowej konwersacji na platformie Azure

Utwórz nową sesję i wyślij pierwszą wiadomość do wdrożonego agenta:

curl -i -X POST "https://$(azd env get-value AZURE_FUNCTION_NAME).azurewebsites.net/api/agents/MyDurableAgent/run?code=$API_KEY" \
  -H "Content-Type: text/plain" \
  -d "What are three popular programming languages?"

Zapisz identyfikator sesji zwrócony w nagłówku x-ms-session-id odpowiedzi.

Kontynuuj konwersację na platformie Azure

Wyślij komunikat monitu w tej samej sesji. Użyj dokładnego identyfikatora zwróconego przez poprzednią odpowiedź:

SESSION_ID="<session-id>"
curl -X POST "https://$(azd env get-value AZURE_FUNCTION_NAME).azurewebsites.net/api/agents/MyDurableAgent/run?code=$API_KEY" \
  --url-query "sessionId=$SESSION_ID" \
  -H "Content-Type: text/plain" \
  -d "Which is easiest to learn?"

Agent utrzymuje kontekst rozmów na platformie Azure tak samo jak lokalnie, co pokazuje trwałość stanu agenta.

Monitorowanie wdrożonego agenta

Wdrożony agent można monitorować przy użyciu pulpitu nawigacyjnego narzędzia Durable Task Scheduler na platformie Azure.

  1. Pobierz nazwę wystąpienia narzędzia Durable Task Scheduler:

    azd env get-value DTS_NAME

  2. Otwórz witrynę Azure Portal i wyszukaj nazwę narzędzia Durable Task Scheduler z poprzedniego kroku.

  3. W bloku przeglądu zasobu Durable Task Scheduler wybierz centrum zadań skonfigurowane dla wdrożonej aplikacji.

  4. Wybierz pozycję Otwórz pulpit nawigacyjny w górnej części strony centrum zadań, aby otworzyć pulpit nawigacyjny monitorowania.

  5. Wyświetl konwersacje agenta tak samo jak w przypadku emulatora lokalnego.

Pulpit nawigacyjny hostowany na platformie Azure udostępnia te same funkcje debugowania i monitorowania co lokalny emulator, co umożliwia inspekcję historii konwersacji, wywołań narzędzi śledzenia i analizowanie wydajności w środowisku produkcyjnym.

Samouczek: organizowanie agentów trwałych za pomocą Azure Functions

W tym samouczku pokazano, jak zorganizować wielu trwałych agentów sztucznej inteligencji przy użyciu modelu hostingu Azure Functions oraz wzorca fan-out/fan-in. Rozszerzysz trwałego agenta z poprzedniego samouczka , aby utworzyć system z wieloma agentami, który przetwarza pytanie użytkownika, a następnie tłumaczy odpowiedź na wiele języków jednocześnie. Przykłady samodzielnej aranżacji można znaleźć w przykładach.

Zrozumienie wzorca orkiestracji

Orkiestracja, którą utworzysz, przebiega zgodnie z tym przepływem:

  1. Dane wejściowe użytkownika — pytanie lub wiadomość od użytkownika
  2. Główny agent — MyDurableAgent element z pierwszego samouczka przetwarza pytanie
  3. Fan-out — odpowiedź głównego agenta jest wysyłana współbieżnie do obu agentów tłumaczenia
  4. Agenci tłumaczenia — dwóch wyspecjalizowanych agentów tłumaczy odpowiedź (francuski i hiszpański)
  5. Fan-in — wyniki są agregowane w pojedynczą odpowiedź JSON z oryginalną odpowiedzią i tłumaczeniami

Ten wzorzec umożliwia współbieżne przetwarzanie, skracając całkowity czas odpowiedzi w porównaniu z tłumaczeniem sekwencyjnym.

Rejestrowanie agentów podczas uruchamiania

Aby prawidłowo używać agentów w trwałych orkiestracjach, zarejestruj je podczas uruchamiania aplikacji. Mogą być stosowane we wszystkich procesach orkiestracji.

Zaktualizuj element Program.cs , aby zarejestrować agentów tłumaczenia obok istniejącego MyDurableAgentelementu :

using System;
using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Hosting.AzureFunctions;
using Microsoft.Azure.Functions.Worker.Builder;
using Microsoft.Extensions.Hosting;

// Get the Azure OpenAI configuration
string endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT")
    ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
string deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT")
    ?? "gpt-4o-mini";

// Create the Microsoft Foundry client
AIProjectClient client = new(new Uri(endpoint), new DefaultAzureCredential());

// Create the main agent from the first tutorial
AIAgent mainAgent = client.AsAIAgent(
    model: deploymentName,
    instructions: "You are a helpful assistant that can answer questions and provide information.",
    name: "MyDurableAgent");

// Create translation agents
AIAgent frenchAgent = client.AsAIAgent(
    model: deploymentName,
    instructions: "You are a translator. Translate the following text to French. Return only the translation, no explanations.",
    name: "FrenchTranslator");

AIAgent spanishAgent = client.AsAIAgent(
    model: deploymentName,
    instructions: "You are a translator. Translate the following text to Spanish. Return only the translation, no explanations.",
    name: "SpanishTranslator");

// Build and configure the Functions host
using IHost app = FunctionsApplication
    .CreateBuilder(args)
    .ConfigureFunctionsWebApplication()
    .ConfigureDurableAgents(options =>
    {
        // Register all agents for use in orchestrations and HTTP endpoints
        options.AddAIAgent(mainAgent);
        options.AddAIAgent(frenchAgent);
        options.AddAIAgent(spanishAgent);
    })
    .Build();

app.Run();

Zaktualizuj element function_app.py , aby zarejestrować agentów tłumaczenia obok istniejącego MyDurableAgentelementu :

import os
from azure.identity import DefaultAzureCredential
from agent_framework_azurefunctions import AgentFunctionApp
from agent_framework.openai import OpenAIChatCompletionClient

# Get the Azure OpenAI configuration
endpoint = os.getenv("AZURE_OPENAI_ENDPOINT")
if not endpoint:
    raise ValueError("AZURE_OPENAI_ENDPOINT is not set.")
deployment_name = os.getenv("AZURE_OPENAI_DEPLOYMENT", "gpt-4o-mini")
api_version = os.getenv("AZURE_OPENAI_API_VERSION")

# Create the Azure OpenAI client
chat_client = OpenAIChatCompletionClient(
    azure_endpoint=endpoint,
    model=deployment_name,
    api_version=api_version,
    credential=DefaultAzureCredential()
)

# Create the main agent from the first tutorial
main_agent = chat_client.as_agent(
    instructions="You are a helpful assistant that can answer questions and provide information.",
    name="MyDurableAgent"
)

# Create translation agents
french_agent = chat_client.as_agent(
    instructions="You are a translator. Translate the following text to French. Return only the translation, no explanations.",
    name="FrenchTranslator"
)

spanish_agent = chat_client.as_agent(
    instructions="You are a translator. Translate the following text to Spanish. Return only the translation, no explanations.",
    name="SpanishTranslator"
)

# Create the function app and register all agents
app = AgentFunctionApp(agents=[main_agent, french_agent, spanish_agent])

Utwórz funkcję orkiestracji

Funkcja orkiestracji koordynuje przepływ pracy między wieloma agentami. Pobiera zarejestrowanych agentów z trwałego kontekstu i organizuje ich wykonywanie, najpierw wywołując głównego agenta, a następnie równolegle kierując na agentów tłumaczenia.

Utwórz nowy plik o nazwie AgentOrchestration.cs w katalogu projektu:

using System.Collections.Generic;
using System.Threading.Tasks;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.DurableTask;
using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;

namespace MyDurableAgent;

public static class AgentOrchestration
{
    // Define a strongly-typed response structure for agent outputs
    public sealed record TextResponse(string Text);

    [Function("agent_orchestration_workflow")]
    public static async Task<Dictionary<string, string>> AgentOrchestrationWorkflow(
        [OrchestrationTrigger] TaskOrchestrationContext context)
    {
        var input = context.GetInput<string>() ?? throw new ArgumentNullException(nameof(context), "Input cannot be null");

        // Step 1: Get the main agent's response
        DurableAIAgent mainAgent = context.GetAgent("MyDurableAgent");
        AgentResponse<TextResponse> mainResponse = await mainAgent.RunAsync<TextResponse>(input);
        string agentResponse = mainResponse.Result.Text;

        // Step 2: Fan out - get the translation agents and run them concurrently
        DurableAIAgent frenchAgent = context.GetAgent("FrenchTranslator");
        DurableAIAgent spanishAgent = context.GetAgent("SpanishTranslator");

        Task<AgentResponse<TextResponse>> frenchTask = frenchAgent.RunAsync<TextResponse>(agentResponse);
        Task<AgentResponse<TextResponse>> spanishTask = spanishAgent.RunAsync<TextResponse>(agentResponse);

        // Step 3: Wait for both translation tasks to complete (fan-in)
        await Task.WhenAll(frenchTask, spanishTask);

        // Get the translation results
        TextResponse frenchResponse = (await frenchTask).Result;
        TextResponse spanishResponse = (await spanishTask).Result;

        // Step 4: Combine results into a dictionary
        var result = new Dictionary<string, string>
        {
            ["original"] = agentResponse,
            ["french"] = frenchResponse.Text,
            ["spanish"] = spanishResponse.Text
        };

        return result;
    }
}

Dodaj funkcję orkiestracji do pliku function_app.py :

from durabletask.task import OrchestrationContext, when_all

@app.orchestration_trigger(context_name="context")
def agent_orchestration_workflow(context: OrchestrationContext, input_text: str):
    """
    Orchestration function that coordinates multiple agents.
    Returns a dictionary with the original response and translations.
    """
    # Step 1: Get the main agent's response
    main_agent = app.get_agent(context, "MyDurableAgent")
    main_response = yield main_agent.run(input_text)
    agent_response = main_response.text

    # Step 2: Fan out - get the translation agents and run them concurrently
    french_agent = app.get_agent(context, "FrenchTranslator")
    spanish_agent = app.get_agent(context, "SpanishTranslator")

    parallel_tasks = [
        french_agent.run(agent_response),
        spanish_agent.run(agent_response)
    ]

    # Step 3: Wait for both translation tasks to complete (fan-in)
    translations = yield when_all(parallel_tasks)

    # Step 4: Combine results into a dictionary
    result = {
        "original": agent_response,
        "french": translations[0].text,
        "spanish": translations[1].text
    }

    return result

Note

W tym przewodniku omówiono hosting rozszerzenia Durable Extension w języku C# i Python. Aby uzyskać informacje na temat hostingu języka Go, zobacz Hosting A2A i repozytorium Agent Framework Go.

Testowanie aranżacji

Upewnij się, że lokalne zależności programowania z pierwszego samouczka nadal działają:

  • Azurite w jednym oknie terminalu
  • Trwały emulator harmonogramu zadań w innym oknie terminalu

Po uruchomieniu lokalnych zależności programistycznych:

  1. Uruchom aplikację usługi Azure Functions w nowym oknie terminalu:

    func start

  2. Rozszerzenie Durable Functions automatycznie tworzy wbudowane punkty końcowe HTTP do zarządzania orkiestracjami. Rozpocznij aranżację przy użyciu wbudowanego interfejsu API:

    curl -X POST http://localhost:7071/runtime/webhooks/durabletask/orchestrators/agent_orchestration_workflow \
  -H "Content-Type: application/json" \
  -d '"\"What are three popular programming languages?\""'

  1. Odpowiedź zawiera adresy URL do zarządzania instancją orkiestracji.

    {
  "id": "abc123def456",
  "statusQueryGetUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/abc123def456",
  "sendEventPostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/abc123def456/raiseEvent/{eventName}",
  "terminatePostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/abc123def456/terminate",
  "purgeHistoryDeleteUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/abc123def456"
}

  2. Wykonaj zapytanie o stan orkiestracji przy użyciu elementu statusQueryGetUri (zastąp abc123def456 rzeczywistym identyfikatorem instancji):

    curl http://localhost:7071/runtime/webhooks/durabletask/instances/abc123def456

  1. Sonduj punkt końcowy stanu do momentu, gdy runtimeStatus ma wartość Completed. Po zakończeniu zobaczysz dane wyjściowe orkiestracji z odpowiedzią głównego agenta i jego tłumaczeniami:

    {
  "name": "agent_orchestration_workflow",
  "instanceId": "abc123def456",
  "runtimeStatus": "Completed",
  "output": {
    "original": "Three popular programming languages are Python, JavaScript, and Java. Python is known for its simplicity...",
    "french": "Trois langages de programmation populaires sont Python, JavaScript et Java. Python est connu pour sa simplicité...",
    "spanish": "Tres lenguajes de programación populares son Python, JavaScript y Java. Python es conocido por su simplicidad..."
  }
}

Monitoruj orkiestrację na pulpicie nawigacyjnym

Panel kontrolny narzędzia Durable Task Scheduler zapewnia wgląd w orkiestrację:

  1. Otwórz http://localhost:8082 w przeglądarce.

  2. Wybierz centrum zadań skonfigurowane dla aplikacji.

  3. Wybierz kartę "Orkiestracje".

  4. Znajdź swoją instancję orkiestracji na liście.

  5. Wybierz instancję, aby wyświetlić:

    • Oś czasu orkiestracji
    • Wykonywanie agenta głównego, a następnie współbieżnych agentów tłumaczenia
    • Każde wykonanie agenta (MyDurableAgent, następnie francuski i hiszpański tłumacz)
    • Wizualizowane wzorce fan-out i fan-in
    • Czas i czas trwania dla każdego kroku

Wdrażanie orkiestracji na platformie Azure

Wdróż zaktualizowaną aplikację przy użyciu interfejsu wiersza polecenia dla deweloperów platformy Azure:

azd deploy

Spowoduje to wdrożenie zaktualizowanego kodu za pomocą nowej funkcji orkiestracji i dodatkowych agentów w aplikacji usługi Azure Functions utworzonej w pierwszym samouczku.

Testowanie wdrożonej aranżacji

Po wdrożeniu przetestuj swoją orkiestrację działającą na platformie Azure.

  1. Pobierz klucz systemowy dla rozszerzenia trwałego:

    SYSTEM_KEY=$(az functionapp keys list --name $(azd env get-value AZURE_FUNCTION_NAME) --resource-group $(azd env get-value AZURE_RESOURCE_GROUP) --query "systemKeys.durabletask_extension" -o tsv)

  1. Rozpocznij aranżację przy użyciu wbudowanego interfejsu API:

    curl -X POST "https://$(azd env get-value AZURE_FUNCTION_NAME).azurewebsites.net/runtime/webhooks/durabletask/orchestrators/agent_orchestration_workflow?code=$SYSTEM_KEY" \
  -H "Content-Type: application/json" \
  -d '"\"What are three popular programming languages?\""'
  1. Użyj statusQueryGetUri z odpowiedzi, aby sprawdzić, czy zadanie zostało ukończone, i zapoznać się z wynikami z tłumaczeniami.

Zabezpieczenia produkcyjne i rozwiązywanie problemów

Traktuj te granice dostępu oddzielnie:

  • Użytkownicy aplikacji: Uwierzytelnij osoby wywołujące i autoryzuj dostęp do każdej sesji, przepływu pracy i odpowiedzi zatwierdzania przez człowieka. Identyfikatory sesji, korelacji i żądań to identyfikatory, a nie poświadczenia.
  • Punkty końcowe funkcji: Klucze na poziomie funkcji chronią wygenerowane punkty końcowe HTTP, ale nie ustanawiają tożsamości użytkownika aplikacji ani własności sesji. Ochrona i obracanie kluczy; nie udostępniaj ich w kodzie aplikacji dostarczanym w przeglądarce.
  • Dostęp do modelu: Udziel tożsamości procesu roboczego wymagane uprawnienia modelu. Poświadczenia modelu nie udzielają dostępu do harmonogramu ani sesji innego użytkownika.
  • Dostęp do harmonogramu: Skonfiguruj uprawnienia procesu roboczego, klienta i pulpitu nawigacyjnego dla ich zamierzonych ról. Aby uzyskać wskazówki dotyczące infrastruktury, zobacz Durable Task integration (Integracja rozszerzenia Durable Task).
Objaw Sprawdzanie
instalacja pakietu Python lub uruchamianie usługi Functions kończy się niepowodzeniem Użyj Python 3.13 lub nowszej dla usługi Functions, zainstaluj pakiety usługi Functions w temacie Wprowadzenie i jawnie wybierz strategię wdrażania. Autonomiczna Python minimalna jest inna.
Żądania kończą się powodzeniem, ale pulpit nawigacyjny usługi DTS jest pusty Upewnij się, że dostawca magazynu usługi Functions to AzureManaged, a nie AzureStorage, i czy punkt końcowy harmonogramu i centrum zadań są zgodne z pulpitem nawigacyjnym.
Sesja wydaje się tracić kontekst Ponownie użyj dokładnie zwróconej x-ms-session-id wartości za pomocą sessionIdpolecenia ; sprawdź, czy nie ma innego centrum, przestrzeni nazw wdrożenia lub skonfigurowanego czyszczenia sesji.
Stare przepływy pracy kończą się niepowodzeniem po uaktualnieniu Sprawdź zgodność odtwarzania przed wznowieniem przebiegów historycznych. Migracja jednostek nie naprawia niezgodnych historii przepływu pracy.
Strumień kończy się, ale brakuje trwałego uzupełniania Traktuj dane wyjściowe strumienia jako tymczasowe i sprawdź wynik autorytatywny lub zachowany. Znacznik ukończenia strumienia nie jest trwałym potwierdzeniem zatwierdzenia.
Ukończona operacja zwraca 410 Sprawdź zachowany wynik i dostępność wyników. Nie zakładaj niepowodzenia, nie sonduj tak, jakby oczekujące lub automatycznie przesyłały nową operację.
Zapisy stanu jednostki przekraczają pojemność Sprawdź stan całej jednostki, rekordy chronione, zasady przechowywania i odwołania do ładunku. Sama kompaktowanie danych wejściowych modelu nie gwarantuje, że przechowywany stan pasuje.

Następne kroki

Punkty końcowe kompatybilne z OpenAI

Dodatkowe zasoby:

  • Ostatnia aktualizacja o