Dokumentacja długotrwałego interfejsu API agenta (wersja zapoznawcza)

Ta dokumentacja zawiera listę powierzchni zestawu SDK AgentServer na potrzeby tworzenia długotrwałych, odpornych agentów hostowanych: odpornych elementów pierwotnych zadań, opcji odpornych odpowiedzi i rejestru przesyłania strumieniowego. Aby zapoznać się z pojęciami dotyczącymi tych interfejsów API, zobacz Odporność dla długotrwałych agentów hostowanych. Aby zapoznać się z przewodnikami zorientowanymi na zadania, zobacz przewodniki z instrukcjami.

Note

Długotrwałi agenci są w wersji zapoznawczej. Interfejsy API na tej stronie pochodzą z azure-ai-agentserver pakietów i mogą ulec zmianie.

Packages

Interfejsy API na tej stronie wymagają co najmniej następujących wersji pakietu. Pakiety Python są ogólnie dostępne. Pakiety .NET są dostępne w wersji zapoznawczej — zainstaluj je za pomocą polecenia --prerelease.

Protocol	Python (minimalna wersja)	.NET (minimalna wersja)
Podstawowe (zadania, przesyłanie strumieniowe, magazyn)	azure-ai-agentserver-core ≥ 2.0.0	Azure.AI.AgentServer.Core ≥ 1.0.0-beta.28
Responses	azure-ai-agentserver-responses ≥ 2.0.0	Azure.AI.AgentServer.Responses ≥ 1.0.0-beta.8
Wywołania	azure-ai-agentserver-invocations ≥ 1.0.0	Azure.AI.AgentServer.Invocations ≥ 1.0.0-beta.6

Note

Użyj najnowszej dostępnej wersji. Wymienione wersje to minimum, które obejmują zadania odporne, odporne odpowiedzi i interfejsy API rejestru przesyłania strumieniowego opisane tutaj.

Zadania odporne

Typy pierwotne zadań sprawiają, że jednostka pracy jest odporna na awarie. Podsystem zadań odpornych jest opt-in: wywołanie set_resilient_tasks_enabled(True) przed uruchomieniem hosta, więc struktura konstruuje TaskManager i uruchamia skanowanie odzyskiwania uruchamiania (zobacz Włączanie zadań odpornych).

Deklarowanie zadania

Python

from datetime import timedelta
from azure.ai.agentserver.core.tasks import task, multi_turn_task, TaskContext, RetryPolicy

@task(name="summarize", timeout=timedelta(minutes=10), retry=RetryPolicy())
async def summarize(ctx: TaskContext[str]) -> str:
    ...

@multi_turn_task(name="chat", steerable=True)
async def chat(ctx: TaskContext[dict]) -> dict:
    ...

Dekorator	Purpose
@task(*, name, title=None, timeout=None, retry=None)	Praca odporna na jeden strzał. Jedno dane wejściowe generuje jeden wynik.
@multi_turn_task(*, name, title=None, timeout=None, retry=None, steerable=False)	Łańcuch konwersacji, który pozostaje suspended między zakrętami. Ustaw steerable=True na kolejkę nowy obrót za aktywnym.

C#

using Azure.AI.AgentServer.Core.Tasks;

// Register the task engine, then declare tasks.
builder.Services.AddResilientTasks(credential);

builder.Services.AddResilientTaskBuilder(tasks =>
{
    tasks.AddTask<string, string>("summarize", async (TaskContext<string> ctx, CancellationToken ct) =>
    {
        // ...
        return result;
    });

    tasks.AddMultiTurnTask<ChatInput, ChatOutput>("chat", async (ctx, ct) =>
    {
        // ...
        return output;
    }, steerable: true);
});

Metoda	Purpose
services.AddResilientTasks(TokenCredential)	Rejestruje odporny aparat zadań i skaner odzyskiwania.
ResilientTaskBuilder.AddTask<TIn,TOut>(name, handler, configure?)	Deklaruje pracę odporną na jeden strzał.
ResilientTaskBuilder.AddMultiTurnTask<TIn,TOut>(name, handler, steerable, configure?)	Deklaruje łańcuch wieloeściowy.

Uruchamianie zadania

Python

# One-shot: run returns the output directly.
result = await summarize.run(input="...")

# Multi-turn: same task_id resumes the chain; if_last_input_id enforces ordering.
r = await chat.run(task_id="conv-7", input={"msg": "hi"}, if_last_input_id=prev_id)

# Fire-and-track: start returns a TaskRun handle.
run = await chat.start(task_id="conv-7", input={"msg": "hi"})

Call	Zwroty	Notatki
await task.run(*, input, task_id=None, if_last_input_id=None)	Output	Uruchamia polecenie do ukończenia i zwraca wynik.
task.start(*, input, task_id=None, if_last_input_id=None)	TaskRun[Output]	Zwraca uchwyt, który można oczekiwać lub anulować.

C#

// Resolve the task and run it.
var run = await task.StartAsync(input, new RunOptions
{
    TaskId = "conv-7",
    IfLastInputId = prevId,
});
var output = await run.GetResultAsync();

RunOptions prowadzi TaskId, InputIdi IfLastInputId (warunek wstępny zamawiania).

TaskContext

Procedura obsługi otrzymuje TaskContext opis bieżącej próby.

Python (TaskContext[Input])	C# (TaskContext<T>)	Description
input	Input	Wartość przekazana przez obiekt wywołujący. Utrwalone przed uruchomieniem programu obsługi.
task_id	TaskId	Trwała tożsamość pracy.
input_id	InputId	Tożsamość per-turn/per-input.
entry_mode	EntryMode	fresh, resumedlub recovered.
metadata	Metadata	Mały trwały stan klucz-wartość (TaskMetadata).
retry_attempt	RetryAttempt	0 w pierwszej próbie.
—	RecoveryCount	Liczba odzyskiwania po awarii dla tej próby.
is_steered_turn	IsSteeredTurn	True jeśli ten obrót został podwyższony z kolejki sterującej.
pending_input_count	PendingInputCount	Ile nowszych kolei jest w kolejce.
cancel	Cancellation	Sygnał anulowania współpracy (każda przyczyna).
cancel_requested	CancelRequested	Przyczyna: zażądano jawnego anulowania.
timeout_exceeded	TimeoutExceeded	Przyczyna: wyzwolony limit czasu dla zadania.
shutdown	Shutdown	Kontener jest zamykany.
await ctx.exit_for_recovery()	await ctx.ExitForRecoveryAsync()	Odroczenie niedokończonej pracy; pozostawia rekord w toku na późniejszy okres istnienia.

entry_mode / EntryMode wartości:

Value	Meaning
fresh / Fresh	Pierwsze wykonanie dla tego elementu (task_id, input_id).
resumed / Resumed	Kolejny zwrot istniejącego łańcucha.
recovered / Recovered	Poprzedni okres istnienia prowadził tę próbę i nie zakończył; wywołano ponownie przy użyciu utrwalionych danych wejściowych.

TaskRun

Dojście zwrócone przez start / StartAsync.

Python (TaskRun[Output])	C# (TaskRun<T>)	Description
task_id	TaskId	Tożsamość służbowa.
input_id	InputId	Tożsamość wejściowa.
metadata	Metadata	Odwołanie do metadanych na żywo podczas lotu.
is_queued	IsQueued	True jeśli te dane wejściowe zostały wprowadzone w kolejce za aktywnym skrętem sterowanym.
await run.result()	await run.GetResultAsync()	Poczekaj na dane wyjściowe.
await run.cancel()	await run.CancelAsync()	Kooperatywne anulowanie.
await run	run.GetAwaiter()	Bezpośrednio można oczekiwać.

Zasady ponawiania prób

Ponowne próby mają zastosowanie do błędów programu obsługi, a nie do odzyskiwania awaryjnego. Odzyskiwanie awaryjne ponownie wprowadza tę samą próbę i nie korzysta z budżetu ponawiania prób.

Python

from azure.ai.agentserver.core.tasks import RetryPolicy

RetryPolicy(
    initial_delay=timedelta(seconds=1),   # base delay
    backoff_coefficient=2.0,              # multiplier per attempt (>= 1.0)
    max_delay=timedelta(seconds=60),      # cap
    max_attempts=3,                       # total tries, including the first
)
RetryPolicy.no_retry()                    # single attempt

Formuła opóźnienia: min(initial_delay * backoff_coefficient ** attempt, max_delay).

C#

TaskRetryPolicy.ExponentialBackoff(maxAttempts: 3, initialDelay: TimeSpan.FromSeconds(1),
    backoffCoefficient: 2.0, maxDelay: TimeSpan.FromSeconds(60), jitter: true);
TaskRetryPolicy.FixedDelay(maxAttempts, delay, jitter);
TaskRetryPolicy.LinearBackoff(maxAttempts, initialDelay, increment, maxDelay, jitter);
TaskRetryPolicy.NoRetry;

Właściwości: InitialDelay, , BackoffCoefficient, MaxAttemptsMaxDelay, Jitter, RetryOn.

Stan zadania i wyjątki

TaskStatus (C#) / wartości stanu: Pending, InProgress, Suspended, Completed.

wyjątek Python	Wyjątek w języku C#	Podniesione, gdy
TaskConflictError	TaskConflictException	Równoczesny niekierunkowy start hit zadania w locie.
TaskCancelled	TaskCancelledException	Przebieg został anulowany.
TaskFailed	TaskFailedException	Procedura obsługi nie powiodła się terminalnie.
TaskDeferred	TaskDeferredException	Praca została odroczona na potrzeby odzyskiwania.
SteeringQueueFull	SteeringQueueFullException	Kolejka sterująca znajduje się w pojemności.
LastInputIdPreconditionFailed	LastInputIdPreconditionFailedException	if_last_input_id / IfLastInputId nie pasuje.
InputTooLarge	InputTooLargeException	Dane wejściowe przekroczyły limit ładunku zadania (~10 MiB).

Włączanie zadań odpornych

Podsystem zadań odpornych jest opt-in. Wywołaj wywołanie set_resilient_tasks_enabled(True) przed uruchomieniem hosta (zazwyczaj w czasie importowania), dlatego AgentServerHost skonstruuje TaskManager i uruchamia skanowanie odzyskiwania uruchamiania. Bez niego get_task_manager() program zgłasza TaskManagerNotInitialized i .run().start() / nie może uruchomić zadania — deklarowanie @task podsystemu lub @multi_turn_task nie włącza go samodzielnie.

Python

from azure.ai.agentserver.core.tasks import set_resilient_tasks_enabled
set_resilient_tasks_enabled(True)   # call at import time, before host startup

C#

AddResilientTasks(credential) rejestruje skaner odzyskiwania podczas uruchamiania hosta; nie jest wymagany żaden oddzielny przełącznik.

Opcje odpowiedzi odpornych

Ustaw wartość na ResponsesServerOptions podczas konstruowania hosta. Obie flagi są domyślnie wyłączone.

Python

from azure.ai.agentserver.responses import ResponsesAgentServerHost, ResponsesServerOptions

app = ResponsesAgentServerHost(
    options=ResponsesServerOptions(
        resilient_background=True,       # re-invoke stored background responses after a crash
        steerable_conversations=True,    # queue a new turn instead of 409 conversation_locked
        default_fetch_history_count=100,
        sse_keep_alive_interval_seconds=None,
        shutdown_grace_period_seconds=10,
    ),
)

C#

var options = new ResponsesServerOptions
{
    ResilientBackground = true,
    SteerableConversations = true,
    DefaultFetchHistoryCount = 100,
};

Python	C#	Domyślnie	Description
resilient_background	ResilientBackground	False	Ponownie wywołaj procedurę obsługi po ponownym uruchomieniu. Dotyczy tylko store=trueodpowiedzi . background=true
steerable_conversations	SteerableConversations	False	Kolejkowanie współbieżnego obrotu zamiast odrzucania go.
default_fetch_history_count	DefaultFetchHistoryCount	100	Maksymalna liczba elementów historii nawodnionych na kolei.
sse_keep_alive_interval_seconds	—	None	Interwał utrzymania aktywności SSE.
shutdown_grace_period_seconds	—	10	Sekundy oczekiwania na pracę w locie po zamknięciu.

Rozpoznawanie odzyskiwania ResponseContext

Dostępne w przypadku włączenia context odporności programu obsługi odpowiedzi.

Python	C#	Description
context.is_recovery	IsRecovery	True gdy program obsługi został ponownie wywołany po awarii.
context.persisted_response	PersistedResponse	Ostatnia trwała migawka odpowiedzi z punktem kontrolnym.
context.conversation_chain_metadata	ConversationChainMetadata	Małe odwołania krzyżowe i znaki wodne.
—	ConversationChainId	Stabilna tożsamość łańcucha konwersacji.
context.is_steered_turn	IsSteeredTurn	True jeśli ten obrót został podwyższony z kolejki sterującej.
context.pending_input_count	PendingInputCount	Nowsze zamienia oczekiwanie w kolejce.
await context.exit_for_recovery()	await ExitForRecoveryAsync()	Odroczenie odzyskiwania po zamknięciu; pozostawia odpowiedź in_progress.

Rejestr przesyłania strumieniowego

Wybierz jedną kopię zapasową podczas uruchamiania, a następnie wyszukaj strumienie według identyfikatora poszczególnych obrotu.

Python

from azure.ai.agentserver.core.streaming import streams

streams.use_in_memory_live()                                        # no replay, no restart survival
streams.use_in_memory_replay(cursor_fn=lambda e: e["n"], ttl_seconds=600)
streams.use_file_backed_replay(storage_dir=Path("/streams"),
                               cursor_fn=lambda e: e["n"])

stream = await streams.get_or_create(invocation_id)
await stream.emit({"n": 0, "delta": "..."})
async for event in stream.subscribe(after=last_cursor):
    ...
await stream.close()

Konfigurator	Powtórka	Przetrwa ponowne uruchomienie
use_in_memory_live()	No	No
use_in_memory_replay(*, cursor_fn=None, ttl_seconds=None)	Tak, w ciągu czasu wygaśnięcia	No
use_file_backed_replay(*, storage_dir=None, cursor_fn=None, ttl_seconds=None, serializer=None, deserializer=None)	Yes	Yes

Operacje rejestru: await streams.get_or_create(id), await streams.delete(id). Przekaż cursor_fn , aby włączyć i subscribe(after=...)last_cursor().

C#

using Azure.AI.AgentServer.Core.Streaming;

builder.Services.AddEventStreams(options =>
{
    // Configure the backing (live / replay / file-backed) via EventStreamOptions.
});

// Resolve the registry and obtain a per-turn stream.
EventStream stream = await registry.GetOrCreateAsync(invocationId);

Kopie zapasowe: BroadcastEventStream (na żywo), ReplayEventStream (odtwarzanie w pamięci), FileBackedReplayEventStream (trwałe). Rejestr: EventStreamRegistry / InMemoryEventStreamRegistry.

Wyjątki przesyłania strumieniowego: EventStreamClosedError/EventStreamClosedException,EventStreamNotFoundException/EventStreamNotFoundError i podstawowe .EventStreamError/EventStreamException

Treści powiązane

• Odporność długotrwałych hostowanych agentów
• Odzyskiwanie długotrwałej pracy po awarii
• Kierowanie kolei agenta w locie
• Przesyłanie strumieniowe długotrwałych danych wyjściowych agenta za pomocą ponownego połączenia
• Zarządzanie stanem długotrwałych agentów
• Trwały magazyn stanów dla hostowanych agentów