Orkestrasi Alur Kerja Kerangka Kerja Agen Microsoft - Berurutan

Dalam orkestrasi berurutan, agen diatur dalam alur. Setiap agen memproses tugas secara bergantian, meneruskan outputnya ke agen berikutnya secara berurutan. Ini sangat ideal untuk alur kerja di mana setiap langkah dibangun berdasarkan yang sebelumnya, seperti tinjauan dokumen, alur pemrosesan data, atau penalaran multi-tahap.

Penting

Riwayat percakapan lengkap dari agen sebelumnya diteruskan ke agen berikutnya secara berurutan. Setiap agen dapat melihat semua pesan sebelumnya, yang memungkinkan pemrosesan sadar konteks.

Apa yang akan Anda Pelajari

• Cara membuat alur kerja agen yang berurutan
• Cara menyusun agen di mana setiapnya dibangun di atas output sebelumnya
• Cara menambahkan persetujuan human-in-the-loop untuk panggilan alat sensitif
• Cara menggabungkan agen dengan pelaksana kustom untuk tugas khusus
• Cara melacak alur percakapan melalui pipeline

Tentukan Agen Anda

Dalam orkestrasi berurutan, agen diatur dalam alur di mana setiap agen memproses tugas secara bergantian, meneruskan output ke agen berikutnya secara berurutan.

Menyiapkan Klien Azure OpenAI

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;
using Microsoft.Agents.AI;

// 1) Set up the Azure OpenAI client
var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") ??
    throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
var deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT_NAME") ?? "gpt-4o-mini";
var client = new AIProjectClient(new Uri(endpoint), new DefaultAzureCredential())
    .GetProjectOpenAIClient()
    .GetProjectResponsesClient()
    .AsIChatClient(deploymentName);

Peringatan

DefaultAzureCredential nyaman untuk pengembangan tetapi membutuhkan pertimbangan yang cermat dalam produksi. Dalam produksi, pertimbangkan untuk menggunakan kredensial tertentu (misalnya, ManagedIdentityCredential) untuk menghindari masalah latensi, pemeriksaan kredensial yang tidak diinginkan, dan potensi risiko keamanan dari mekanisme fallback.

Buat agen khusus yang akan berfungsi secara berurutan:

// 2) Helper method to create translation agents
static ChatClientAgent GetTranslationAgent(string targetLanguage, IChatClient chatClient) =>
    new(chatClient,
        $"You are a translation assistant who only responds in {targetLanguage}. Respond to any " +
        $"input by outputting the name of the input language and then translating the input to {targetLanguage}.");

// Create translation agents for sequential processing
var translationAgents = (from lang in (string[])["French", "Spanish", "English"]
                         select GetTranslationAgent(lang, client));

Menyiapkan Orkestrasi Berurutan

Bangun alur kerja menggunakan AgentWorkflowBuilder:

// 3) Build sequential workflow
var workflow = AgentWorkflowBuilder.BuildSequential(translationAgents);

Jalankan Alur Kerja Berurutan

Jalankan alur kerja dan proses peristiwa:

// 4) Run the workflow
var messages = new List<ChatMessage> { new(ChatRole.User, "Hello, world!") };

await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, messages);
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));

string? lastExecutorId = null;
List<ChatMessage> result = [];
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    if (evt is AgentResponseUpdateEvent e)
    {
        if (e.ExecutorId != lastExecutorId)
        {
            lastExecutorId = e.ExecutorId;
            Console.WriteLine();
            Console.Write($"{e.ExecutorId}: ");
        }

        Console.Write(e.Update.Text);
    }
    else if (evt is WorkflowOutputEvent outputEvt)
    {
        result = outputEvt.As<List<ChatMessage>>()!;
        break;
    }
}

// Display final result
Console.WriteLine();
foreach (var message in result)
{
    Console.WriteLine($"{message.Role}: {message.Text}");
}

Sampel Output

French_Translation: User: Hello, world!
French_Translation: Assistant: English detected. Bonjour, le monde !
Spanish_Translation: Assistant: French detected. ¡Hola, mundo!
English_Translation: Assistant: Spanish detected. Hello, world!

Orkestrasi Berurutan dengan Human-in-the-Loop

Orkestrasi berurutan mendukung interaksi human-in-the-loop melalui persetujuan alat. Saat agen menggunakan alat yang dibalut dengan ApprovalRequiredAIFunction, alur kerja menjeda dan memancarkan RequestInfoEvent yang berisi ToolApprovalRequestContent. Sistem eksternal (seperti operator manusia) dapat memeriksa panggilan alat, menyetujui atau menolaknya, dan alur kerja dilanjutkan.

Petunjuk / Saran

Untuk detail selengkapnya tentang model permintaan dan respons, lihat Human-in-the-Loop.

Tentukan Agen dengan Alat yang Memerlukan Persetujuan

Buat agen di mana alat sensitif dibungkus dengan ApprovalRequiredAIFunction:

ChatClientAgent deployAgent = new(
    client,
    "You are a DevOps engineer. Check staging status first, then deploy to production.",
    "DeployAgent",
    "Handles deployments",
    [
        AIFunctionFactory.Create(CheckStagingStatus),
        new ApprovalRequiredAIFunction(AIFunctionFactory.Create(DeployToProduction))
    ]);

ChatClientAgent verifyAgent = new(
    client,
    "You are a QA engineer. Verify that the deployment was successful and summarize the results.",
    "VerifyAgent",
    "Verifies deployments");

Membangun dan Menjalankan dengan Penanganan Persetujuan

Bangun alur kerja berurutan secara normal. Alur persetujuan ditangani melalui aliran peristiwa:

var workflow = AgentWorkflowBuilder.BuildSequential([deployAgent, verifyAgent]);

await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    if (evt is RequestInfoEvent e &&
        e.Request.TryGetDataAs(out ToolApprovalRequestContent? approvalRequest))
    {
        await run.SendResponseAsync(
            e.Request.CreateResponse(approvalRequest.CreateResponse(approved: true)));
    }
}

Nota

AgentWorkflowBuilder.BuildSequential() mendukung persetujuan alat di luar kotak — tidak diperlukan konfigurasi tambahan. Saat agen memanggil alat yang dibungkus dengan ApprovalRequiredAIFunction, alur kerja secara otomatis terhenti dan memancarkan RequestInfoEvent.

Petunjuk / Saran

Untuk contoh lengkap alur persetujuan ini yang dapat dieksekusi, lihat GroupChatToolApproval contoh. Pola penanganan yang sama RequestInfoEvent berlaku untuk orkestrasi lain.

Konsep utama

• Pemrosesan Berurutan: Setiap agen memproses output agen sebelumnya secara berurutan
• AgentWorkflowBuilder.BuildSequential(): Membuat alur kerja dari kumpulan agen
• ChatClientAgent: Mewakili agen yang didukung oleh klien obrolan dengan instruksi tertentu
• InProcessExecution.RunStreamingAsync(): Menjalankan alur kerja dan mengembalikan StreamingRun untuk streaming peristiwa real time
• Penanganan Peristiwa: Memantau kemajuan agen melalui AgentResponseUpdateEvent dan penyelesaian melalui WorkflowOutputEvent
• Persetujuan Alat: Membungkus alat sensitif dengan ApprovalRequiredAIFunction untuk memerlukan persetujuan manusia sebelum eksekusi
• RequestInfoEvent: Dikeluarkan ketika alat memerlukan persetujuan; berisi ToolApprovalRequestContent dengan detail panggilan alat

Dalam orkestrasi berurutan, setiap agen memproses tugas secara bergantian, dengan output mengalir dari satu ke yang berikutnya. Mulailah dengan menentukan agen untuk proses dua tahap:

import os
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential

# 1) Create agents using FoundryChatClient
chat_client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["FOUNDRY_MODEL"],
    credential=AzureCliCredential(),
)

writer = chat_client.as_agent(
    instructions=(
        "You are a concise copywriter. Provide a single, punchy marketing sentence based on the prompt."
    ),
    name="writer",
)

reviewer = chat_client.as_agent(
    instructions=(
        "You are a thoughtful reviewer. Give brief feedback on the previous assistant message."
    ),
    name="reviewer",
)

Menyiapkan Orkestrasi Berurutan

Kelas SequentialBuilder membuat alur tempat agen memproses tugas secara berurutan. Setiap agen melihat riwayat percakapan lengkap dan menambahkan respons mereka:

from agent_framework.orchestrations import SequentialBuilder

# 2) Build sequential workflow: writer -> reviewer
workflow = SequentialBuilder(participants=[writer, reviewer]).build()

Jalankan Alur Kerja Berurutan

Jalankan alur kerja dan kumpulkan percakapan akhir yang menunjukkan kontribusi setiap agen:

from typing import Any, cast
from agent_framework import Message, WorkflowEvent

# 3) Run and print final conversation
outputs: list[list[Message]] = []
async for event in workflow.run("Write a tagline for a budget-friendly eBike.", stream=True):
    if event.type == "output":
        outputs.append(cast(list[Message], event.data))

if outputs:
    print("===== Final Conversation =====")
    messages: list[Message] = outputs[-1]
    for i, msg in enumerate(messages, start=1):
        name = msg.author_name or ("assistant" if msg.role == "assistant" else "user")
        print(f"{'-' * 60}\n{i:02d} [{name}]\n{msg.text}")

Sampel Output

===== Final Conversation =====
------------------------------------------------------------
01 [user]
Write a tagline for a budget-friendly eBike.
------------------------------------------------------------
02 [writer]
Ride farther, spend less—your affordable eBike adventure starts here.
------------------------------------------------------------
03 [reviewer]
This tagline clearly communicates affordability and the benefit of extended travel, making it
appealing to budget-conscious consumers. It has a friendly and motivating tone, though it could
be slightly shorter for more punch. Overall, a strong and effective suggestion!

Tingkat Lanjut: Menggabungkan Agen dengan Eksekutor Kustom

Orkestrasi berurutan mendukung penggabungan agen dengan eksekutor kustom untuk pemrosesan khusus. Ini berguna ketika Anda memerlukan logika kustom yang tidak memerlukan LLM:

Tentukan Pelaksana Kustom

Nota

Ketika eksekutor kustom mengikuti agen dalam urutan, handler-nya menerima AgentExecutorResponse (karena agen dibungkus secara internal oleh AgentExecutor). Gunakan agent_response.full_conversation untuk mengakses riwayat percakapan lengkap.

from agent_framework import AgentExecutorResponse, Executor, WorkflowContext, handler
from agent_framework import Message

class Summarizer(Executor):
    """Simple summarizer: consumes full conversation and appends an assistant summary."""

    @handler
    async def summarize(
        self,
        agent_response: AgentExecutorResponse,
        ctx: WorkflowContext[list[Message]]
    ) -> None:
        if not agent_response.full_conversation:
            await ctx.send_message([Message("assistant", ["No conversation to summarize."])])
            return

        users = sum(1 for m in agent_response.full_conversation if m.role == "user")
        assistants = sum(1 for m in agent_response.full_conversation if m.role == "assistant")
        summary = Message("assistant", [f"Summary -> users:{users} assistants:{assistants}"])
        await ctx.send_message(list(agent_response.full_conversation) + [summary])

Membangun Alur Kerja Berurutan Campuran

# Create a content agent
content = chat_client.as_agent(
    instructions="Produce a concise paragraph answering the user's request.",
    name="content",
)

# Build sequential workflow: content -> summarizer
summarizer = Summarizer(id="summarizer")
workflow = SequentialBuilder(participants=[content, summarizer]).build()

Output Sampel dengan Eksekutor Kustom

------------------------------------------------------------
01 [user]
Explain the benefits of budget eBikes for commuters.
------------------------------------------------------------
02 [content]
Budget eBikes offer commuters an affordable, eco-friendly alternative to cars and public transport.
Their electric assistance reduces physical strain and allows riders to cover longer distances quickly,
minimizing travel time and fatigue. Budget models are low-cost to maintain and operate, making them accessible
for a wider range of people. Additionally, eBikes help reduce traffic congestion and carbon emissions,
supporting greener urban environments. Overall, budget eBikes provide cost-effective, efficient, and
sustainable transportation for daily commuting needs.
------------------------------------------------------------
03 [assistant]
Summary -> users:1 assistants:1

Orkestrasi Berurutan dengan Human-in-the-Loop

Orkestrasi berurutan mendukung interaksi manusia dalam lingkaran pengendali dengan dua cara: persetujuan alat untuk mengontrol panggilan alat sensitif, dan permintaan informasi dengan jeda setelah setiap respons agen untuk mengumpulkan umpan balik.

Petunjuk / Saran

Untuk detail selengkapnya tentang model permintaan dan respons, lihat Human-in-the-Loop.

Persetujuan Alat dalam Alur Kerja Berurutan

Gunakan @tool(approval_mode="always_require") untuk menandai alat yang memerlukan persetujuan manusia sebelum eksekusi. Alur kerja menjeda dan menghasilkan request_info peristiwa saat agen mencoba menggunakan alat.

@tool(approval_mode="always_require")
def execute_database_query(query: str) -> str:
    return f"Query executed successfully: {query}"


database_agent = Agent(
    client=chat_client,
    name="DatabaseAgent",
    instructions="You are a database assistant.",
    tools=[execute_database_query],
)

workflow = SequentialBuilder(participants=[database_agent]).build()

Proses aliran peristiwa dan tangani permintaan persetujuan:

async def process_event_stream(stream):
    responses = {}
    async for event in stream:
        if event.type == "request_info" and event.data.type == "function_approval_request":
            responses[event.request_id] = event.data.to_function_approval_response(approved=True)
    return responses if responses else None

stream = workflow.run("Check the schema and update all pending orders", stream=True)

pending_responses = await process_event_stream(stream)
while pending_responses is not None:
    stream = workflow.run(stream=True, responses=pending_responses)
    pending_responses = await process_event_stream(stream)

Petunjuk / Saran

Untuk contoh lengkap yang dapat dijalankan, lihat sequential_builder_tool_approval.py. Persetujuan tool berfungsi dengan SequentialBuilder tanpa konfigurasi builder tambahan.

Permintaan Info untuk Tanggapan Agen

Gunakan .with_request_info() untuk menjeda setelah agen tertentu merespons, memungkinkan input eksternal (seperti tinjauan manusia) sebelum agen berikutnya dimulai:

drafter = Agent(
    client=chat_client,
    name="drafter",
    instructions="You are a document drafter. Create a brief draft on the given topic.",
)

editor = Agent(
    client=chat_client,
    name="editor",
    instructions="You are an editor. Review and improve the draft. Incorporate any human feedback.",
)

finalizer = Agent(
    client=chat_client,
    name="finalizer",
    instructions="You are a finalizer. Create a polished final version.",
)

# Enable request info for the editor agent only
workflow = (
    SequentialBuilder(participants=[drafter, editor, finalizer])
    .with_request_info(agents=["editor"])
    .build()
)

async def process_event_stream(stream):
    responses = {}
    async for event in stream:
        if event.type == "request_info":
            responses[event.request_id] = AgentRequestInfoResponse.approve()
    return responses if responses else None

stream = workflow.run("Write a brief introduction to artificial intelligence.", stream=True)

pending_responses = await process_event_stream(stream)
while pending_responses is not None:
    stream = workflow.run(stream=True, responses=pending_responses)
    pending_responses = await process_event_stream(stream)

Petunjuk / Saran

Lihat sampel lengkap: pengesahan alat secara berurutan dan informasi permintaan secara berurutan.

Konsep utama

• Konteks Bersama: Setiap peserta menerima riwayat percakapan lengkap, termasuk semua pesan sebelumnya
• Order Matters: Agen-agen menjalankan secara ketat dalam urutan yang ditentukan dalam daftar participants
• Peserta Fleksibel: Anda dapat mencampur agen dan pelaksana kustom dalam urutan apa pun
• Alur Percakapan: Setiap agen/pelaksana menambahkan ke percakapan, membangun dialog lengkap
• Persetujuan Alat: Gunakan @tool(approval_mode="always_require") untuk operasi sensitif yang memerlukan peninjauan manusia
• Permintaan Informasi: Gunakan .with_request_info(agents=[...]) untuk berhenti sejenak setelah agen tertentu demi umpan balik eksternal

Langkah selanjutnya

Orkestrasi Bersamaan