Bagikan melalui

Fitur Agen Tahan Lama

Saat Anda membangun agen AI dengan Microsoft Agent Framework, ekstensi tugas tahan lama untuk Microsoft Agent Framework menambahkan kemampuan tingkat lanjut ke agen standar Anda termasuk manajemen status percakapan otomatis, orkestrasi deterministik, dan pola human-in-the-loop. Ekstensi ini juga memudahkan untuk menghosting agen Anda pada komputasi tanpa server yang disediakan oleh Azure Functions, memberikan penskalaan dinamis dan model penagihan per permintaan yang hemat biaya.

Orkestrasi Multi-Agen Deterministik

Ekstensi tugas yang tahan lama mendukung pembuatan alur kerja deterministik yang mengoordinasikan beberapa agen dengan menggunakan orkestrasi Azure Durable Functions.

Orkestrasi adalah alur kerja berbasis kode yang mengoordinasikan beberapa operasi (seperti panggilan agen, panggilan API eksternal, atau timer) dengan cara yang dapat diandalkan. Deterministik berarti kode orkestrasi menjalankan cara yang sama ketika diputar ulang setelah kegagalan, membuat alur kerja dapat diandalkan dan dapat di-debug—saat Anda memutar ulang riwayat orkestrasi, Anda dapat melihat dengan tepat apa yang terjadi di setiap langkah.

Orkestrasi dijalankan dengan andal, bertahan terhadap kegagalan yang terjadi di antara panggilan agen, serta memberikan proses yang dapat diprediksi dan dapat diulang. Ini membuatnya ideal untuk skenario multi-agen yang kompleks di mana Anda memerlukan urutan eksekusi terjamin dan toleransi kesalahan.

Orkestrasi Berurutan

Dalam pola multi-agen berurutan, agen khusus dijalankan dalam urutan tertentu, di mana setiap output agen dapat memengaruhi eksekusi agen berikutnya. Pola ini mendukung logika kondisional dan percabangan berdasarkan respons agen.

Saat menggunakan agen dalam orkestrasi, Anda harus menggunakan context.GetAgent() API untuk mendapatkan instance DurableAIAgent, yang merupakan subkelas khusus dari tipe standar AIAgent yang membungkus salah satu agen terdaftar Anda. Pembungkus DurableAIAgent memastikan agar panggilan agen dilacak dengan benar dan mendapatkan titik pemeriksaan oleh kerangka orkestrasi berdaya tahan.

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");
    AgentThread spamThread = spamDetectionAgent.GetNewThread();

    AgentRunResponse<DetectionResult> spamDetectionResponse = await spamDetectionAgent.RunAsync<DetectionResult>(
        message: $"Analyze this email for spam: {email.EmailContent}",
        thread: spamThread);
    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");
    AgentThread emailThread = emailAssistantAgent.GetNewThread();

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

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

Saat menggunakan agen dalam orkestrasi, Anda harus menggunakan metode app.get_agent() untuk mendapatkan instans agen yang tahan lama, yang merupakan pembungkus khusus yang membungkus salah satu agen terdaftar Anda. Pembungkus agen tahan lama memastikan bahwa panggilan agen dilacak dan diperiksa dengan benar oleh kerangka orkestrasi yang tangguh.

import azure.durable_functions as df
from typing import cast
from agent_framework.azure 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: df.DurableOrchestrationContext):
    email = context.get_input()

    # Check if the email is spam
    spam_agent = app.get_agent(context, "SpamDetectionAgent")
    spam_thread = spam_agent.get_new_thread()

    spam_result_raw = yield spam_agent.run(
        messages=f"Analyze this email for spam: {email['content']}",
        thread=spam_thread,
        response_format=SpamDetectionResult
    )
    spam_result = cast(SpamDetectionResult, spam_result_raw.get("structured_response"))

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

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

    email_response_raw = yield email_agent.run(
        messages=f"Draft a professional response to: {email['content']}",
        thread=email_thread,
        response_format=EmailResponse
    )
    email_response = cast(EmailResponse, email_response_raw.get("structured_response"))

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

Orkestrasi mengoordinasikan pekerjaan di berbagai agen dan mampu bertahan dari kegagalan yang terjadi di antara panggilan agen. Konteks orkestrasi menyediakan metode untuk mengambil dan berinteraksi dengan agen terhosting dalam orkestrasi.

Orkestrasi Paralel

Dalam pola multi-agen paralel, Anda menjalankan beberapa agen secara bersamaan dan kemudian menggabungkan hasilnya. Pola ini berguna untuk mengumpulkan beragam perspektif atau memproses subtugas independen secara bersamaan.

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<AgentRunResponse<TextResponse>> technicalTask = 
        technicalAgent.RunAsync<TextResponse>($"Research technical aspects of {topic}");
    Task<AgentRunResponse<TextResponse>> marketTask = 
        marketAgent.RunAsync<TextResponse>($"Research market trends for {topic}");
    Task<AgentRunResponse<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");
    AgentRunResponse<TextResponse> summaryResponse = 
        await summaryAgent.RunAsync<TextResponse>($"Summarize this research:\n{allResearch}");

    return summaryResponse.Result.Text;
}

import azure.durable_functions as df
from agent_framework.azure import AgentFunctionApp

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

@app.orchestration_trigger(context_name="context")
def research_orchestration(context: df.DurableOrchestrationContext):
    topic = context.get_input()

    # 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 context.task_all([technical_task, market_task, competitor_task])

    # Aggregate results
    all_research = "\n\n".join([r.get('response', '') 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.get('response', '')

Eksekusi paralel dilacak menggunakan daftar tugas. Titik pemeriksaan otomatis memastikan bahwa eksekusi agen yang selesai tidak diulang atau hilang jika kegagalan terjadi selama agregasi.

Orkestrasi Human-in-the-Loop

Orkestrasi agen deterministik dapat menjeda input, persetujuan, atau peninjauan manusia tanpa menggunakan sumber daya komputasi. Eksekusi tahan lama memungkinkan orkestrasi untuk menunggu bertahun-hari atau bahkan berminggu-minggu sambil menunggu respons manusia. Ketika dikombinasikan dengan hosting tanpa server, semua sumber daya komputasi dimatikan selama periode tunggu, menghilangkan biaya komputasi hingga ada masukan dari manusia.

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");
    AgentRunResponse<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";
}

import azure.durable_functions as df
from datetime import timedelta
from agent_framework.azure import AgentFunctionApp

app = AgentFunctionApp(agents=[content_agent])

@app.orchestration_trigger(context_name="context")
def content_approval_workflow(context: df.DurableOrchestrationContext):
    topic = context.get_input()

    # 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", draft_content)

    # 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 context.task_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", draft_content)
            return result
        return "Content rejected"

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

Orkestrasi agen deterministik dapat menunggu peristiwa eksternal, secara stabil mempertahankan keadaan mereka sambil menunggu umpan balik manusia, menghadapi kegagalan, restart, dan periode menunggu yang diperpanjang. Ketika respons manusia tiba, orkestrasi secara otomatis berkelanjutan dengan konteks percakapan yang lengkap dan status eksekusi yang tetap utuh.

Memberikan Masukan dari Manusia

Untuk mengirim persetujuan atau input ke orkestrasi menunggu, Anda harus menaikkan peristiwa eksternal ke instans orkestrasi menggunakan SDK klien Durable Functions. Misalnya, peninjau mungkin menyetujui konten melalui formulir web yang memanggil:

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)

Efisiensi Biaya

Alur kerja human-in-the-loop (manusia dalam lingkaran) dengan agen persisten sangat hemat biaya saat dihosting pada paket Flex Consumption Azure Functions. Untuk alur kerja yang menunggu 24 jam untuk persetujuan, Anda hanya membayar untuk beberapa detik waktu eksekusi (waktu untuk menghasilkan konten, mengirim pemberitahuan, dan memproses respons)—bukan 24 jam menunggu. Selama periode tunggu, tidak ada sumber daya komputasi yang digunakan.

Kemampuan Dapat Diamati dengan Durable Task Scheduler

Durable Task Scheduler (DTS) adalah backend tahan lama yang direkomendasikan untuk agen tahan lama Anda, menawarkan performa terbaik, infrastruktur yang dikelola sepenuhnya, dan pengamatan bawaan melalui dasbor UI. Meskipun Azure Functions dapat menggunakan backend penyimpanan lain (seperti Azure Storage), DTS dioptimalkan khusus untuk beban kerja yang tahan lama dan memberikan kemampuan performa dan pemantauan yang unggul.

Wawasan Thread Agen

  • Riwayat percakapan: Menampilkan utas percakapan lengkap untuk setiap utas agen, termasuk semua pesan, panggilan alat, dan konteks percakapan kapan saja
  • Waktu tugas: Memantau berapa lama waktu yang diperlukan tugas tertentu dan interaksi agen untuk diselesaikan

Cuplikan layar dasbor Durable Task Scheduler memperlihatkan riwayat obrolan agen dengan utas percakapan dan pesan.

Wawasan Orkestrasi

  • Visualisasi multi-agen: Lihat alur eksekusi saat memanggil beberapa agen khusus dengan representasi visual eksekusi paralel dan pencabangan bersyarat
  • Riwayat eksekusi: Anda dapat mengakses log eksekusi yang terperinci
  • Pemantauan real-time: Melacak orkestrasi aktif, item kerja dalam antrean, dan status agen di seluruh penyebaran Anda
  • Metrik performa: Memantau waktu respons agen, penggunaan token, dan durasi orkestrasi

Cuplikan layar dasbor Durable Task Scheduler memperlihatkan visualisasi orkestrasi dengan beberapa interaksi agen dan eksekusi alur kerja.

Kemampuan Debugging

  • Melihat output agen terstruktur dan hasil panggilan alat
  • Melacak pemanggilan fungsi alat dan hasilnya
  • Memantau penanganan peristiwa eksternal untuk skenario Human-In-The-Loop (partisipasi manusia dalam proses)

Dasbor memungkinkan Anda memahami dengan tepat apa yang dilakukan agen Anda, mendiagnosis masalah dengan cepat, dan mengoptimalkan performa berdasarkan data eksekusi nyata.

Isi Terkait

  • Last updated on