Bagikan melalui

Middleware Agen

Middleware dalam Agent Framework menyediakan cara yang ampuh untuk mencegat, memodifikasi, dan meningkatkan interaksi agen pada berbagai tahap eksekusi. Anda dapat menggunakan middleware untuk menerapkan masalah lintas pemotongan seperti pengelogan, validasi keamanan, penanganan kesalahan, dan transformasi hasil tanpa memodifikasi agen inti atau logika fungsi Anda.

Agent Framework dapat disesuaikan menggunakan tiga jenis middleware yang berbeda:

  1. Middleware Eksekusi Agen: Memungkinkan intersepsi semua eksekusi agen, sehingga input dan output dapat diperiksa dan/atau dimodifikasi sesuai kebutuhan.
  2. Middleware panggilan fungsi: Memungkinkan intersepsi semua panggilan fungsi yang dijalankan oleh agen, sehingga input dan output dapat diperiksa dan dimodifikasi sesuai kebutuhan.
  3. IChatClient middleware: Memungkinkan intersepsi panggilan ke IChatClient implementasi, di mana agen menggunakan IChatClient untuk panggilan inferensi, misalnya, saat menggunakan ChatClientAgent.

Semua jenis middleware diimplementasikan melalui panggilan balik fungsi, dan ketika beberapa instans middleware dari jenis yang sama terdaftar, mereka membentuk rantai, di mana setiap instans middleware diharapkan untuk memanggil berikutnya dalam rantai, melalui yang disediakan nextFunc.

Agen menjalankan dan memanggil jenis middleware fungsi dapat didaftarkan pada agen, dengan menggunakan pembuat agen dengan objek agen yang ada.

var middlewareEnabledAgent = originalAgent
    .AsBuilder()
        .Use(runFunc: CustomAgentRunMiddleware, runStreamingFunc: CustomAgentRunStreamingMiddleware)
        .Use(CustomFunctionCallingMiddleware)
    .Build();

Penting

Idealnya baik runFunc dan runStreamingFunc harus disediakan. Saat hanya menyediakan middleware non-streaming, agen akan menggunakannya untuk pemanggilan streaming dan non-streaming. Streaming hanya akan berjalan dalam mode non-streaming untuk mencukupkan harapan middleware.

Nota

Ada kelebihan beban tambahan, Use(sharedFunc: ...), yang memungkinkan Anda menyediakan middleware yang sama untuk non-streaming dan streaming tanpa memblokir streaming. Namun, middleware bersama tidak akan dapat mencegat atau mengambil alih output. Kelebihan beban ini harus digunakan untuk skenario di mana Anda hanya perlu memeriksa atau memodifikasi input sebelum mencapai agen.

IChatClient middleware dapat didaftarkan pada IChatClient sebelum digunakan dengan ChatClientAgent, dengan menggunakan pola pembangun klien obrolan.

var chatClient = new AzureOpenAIClient(new Uri("https://<myresource>.openai.azure.com"), new AzureCliCredential())
    .GetChatClient(deploymentName)
    .AsIChatClient();

var middlewareEnabledChatClient = chatClient
    .AsBuilder()
        .Use(getResponseFunc: CustomChatClientMiddleware, getStreamingResponseFunc: null)
    .Build();

var agent = new ChatClientAgent(middlewareEnabledChatClient, instructions: "You are a helpful assistant.");

IChatClient middleware juga dapat didaftarkan menggunakan metode pabrik saat membuat agen melalui salah satu metode pembantu pada klien SDK.

var agent = new AzureOpenAIClient(new Uri(endpoint), new AzureCliCredential())
    .GetChatClient(deploymentName)
    .AsAIAgent("You are a helpful assistant.", clientFactory: (chatClient) => chatClient
        .AsBuilder()
            .Use(getResponseFunc: CustomChatClientMiddleware, getStreamingResponseFunc: null)
        .Build());

Middleware Eksekusi Agen

Berikut adalah contoh middleware eksekusi agen, yang dapat memeriksa dan/atau memodifikasi input dan output dari agen yang dijalankan.

async Task<AgentResponse> CustomAgentRunMiddleware(
    IEnumerable<ChatMessage> messages,
    AgentSession? session,
    AgentRunOptions? options,
    AIAgent innerAgent,
    CancellationToken cancellationToken)
{
    Console.WriteLine(messages.Count());
    var response = await innerAgent.RunAsync(messages, session, options, cancellationToken).ConfigureAwait(false);
    Console.WriteLine(response.Messages.Count);
    return response;
}

Middleware Streaming Eksekusi Agen

Berikut adalah contoh middleware streaming eksekusi agen, yang dapat memeriksa dan/atau memodifikasi input dan output dari eksekusi streaming agen.

    async IAsyncEnumerable<AgentResponseUpdate> CustomAgentRunStreamingMiddleware(
    IEnumerable<ChatMessage> messages,
    AgentSession? session,
    AgentRunOptions? options,
    AIAgent innerAgent,
    [EnumeratorCancellation] CancellationToken cancellationToken)
{
    Console.WriteLine(messages.Count());
    List<AgentResponseUpdate> updates = [];
    await foreach (var update in innerAgent.RunStreamingAsync(messages, session, options, cancellationToken))
    {
        updates.Add(update);
        yield return update;
    }

    Console.WriteLine(updates.ToAgentResponse().Messages.Count);
}

Middleware panggilan fungsi

Nota

Pemanggilan fungsi middleware saat ini hanya didukung dengan AIAgent yang menggunakan FunctionInvokingChatClient, misalnya, ChatClientAgent.

Berikut adalah contoh fungsi memanggil middleware, yang dapat memeriksa dan/atau memodifikasi fungsi yang dipanggil, dan hasil dari panggilan fungsi.

async ValueTask<object?> CustomFunctionCallingMiddleware(
    AIAgent agent,
    FunctionInvocationContext context,
    Func<FunctionInvocationContext, CancellationToken, ValueTask<object?>> next,
    CancellationToken cancellationToken)
{
    Console.WriteLine($"Function Name: {context!.Function.Name}");
    var result = await next(context, cancellationToken);
    Console.WriteLine($"Function Call Result: {result}");

    return result;
}

Dimungkinkan untuk mengakhiri perulangan panggilan fungsi dengan middleware panggilan fungsi dengan mengatur yang disediakan FunctionInvocationContext.Terminate ke true. Ini akan mencegah perulangan panggilan fungsi mengeluarkan permintaan ke layanan inferensi yang berisi hasil panggilan fungsi setelah pemanggilan fungsi. Jika ada lebih dari satu fungsi yang tersedia untuk pemanggilan selama iterasi ini, mungkin juga mencegah fungsi yang tersisa dijalankan.

Peringatan

Mengakhiri perulangan panggilan fungsi dapat mengakibatkan riwayat obrolan Anda dibiarkan dalam status tidak konsisten, misalnya, berisi konten panggilan fungsi tanpa konten hasil fungsi. Ini dapat mengakibatkan riwayat obrolan tidak dapat digunakan untuk eksekusi lebih lanjut.

Middleware IChatClient

Berikut adalah contoh middleware klien obrolan, yang dapat memeriksa dan/atau memodifikasi input dan output untuk permintaan ke layanan inferensi yang disediakan klien obrolan.

async Task<ChatResponse> CustomChatClientMiddleware(
    IEnumerable<ChatMessage> messages,
    ChatOptions? options,
    IChatClient innerChatClient,
    CancellationToken cancellationToken)
{
    Console.WriteLine(messages.Count());
    var response = await innerChatClient.GetResponseAsync(messages, options, cancellationToken);
    Console.WriteLine(response.Messages.Count);

    return response;
}

Nota

Untuk informasi selengkapnya tentang IChatClient middleware, lihat Middleware IChatClient kustom.

Function-Based Middleware

Middleware berbasis fungsi adalah cara paling sederhana untuk mengimplementasikan middleware menggunakan fungsi asinkron. Pendekatan ini sangat ideal untuk operasi stateless dan menyediakan solusi ringan untuk skenario middleware umum.

Middleware Agen

Middleware agen mencegat dan memodifikasi agen menjalankan eksekusi. Ini menggunakan yang AgentRunContext berisi:

  • agent: Agen yang sedang dipanggil
  • messages: Daftar pesan obrolan dalam percakapan
  • is_streaming: Boolean menunjukkan apakah respons sedang streaming
  • metadata: Kamus untuk menyimpan data tambahan antar middleware
  • result: Respons agen (dapat dimodifikasi)
  • terminate: Bendera untuk menghentikan pemrosesan lebih lanjut
  • kwargs: Argumen kata kunci tambahan yang diteruskan ke metode eksekusi agen

Yang next dapat dipanggil melanjutkan rantai middleware atau menjalankan agen jika itu middleware terakhir.

Berikut adalah contoh pengelogan sederhana dengan logika sebelum dan sesudah next dapat dipanggil:

async def logging_agent_middleware(
    context: AgentRunContext,
    next: Callable[[AgentRunContext], Awaitable[None]],
) -> None:
    """Agent middleware that logs execution timing."""
    # Pre-processing: Log before agent execution
    print("[Agent] Starting execution")

    # Continue to next middleware or agent execution
    await next(context)

    # Post-processing: Log after agent execution
    print("[Agent] Execution completed")

Middleware Fungsi

Middleware fungsi mencegat panggilan fungsi dalam agen. Ini menggunakan yang FunctionInvocationContext berisi:

  • function: Fungsi yang sedang dipanggil
  • arguments: Argumen yang divalidasi untuk fungsi
  • metadata: Kamus untuk menyimpan data tambahan antar middleware
  • result: Nilai pengembalian fungsi (dapat dimodifikasi)
  • terminate: Bendera untuk menghentikan pemrosesan lebih lanjut
  • kwargs: Argumen kata kunci tambahan diteruskan ke metode obrolan yang memanggil fungsi ini

Yang next dapat dipanggil berlanjut ke middleware berikutnya atau menjalankan fungsi aktual.

Berikut adalah contoh pengelogan sederhana dengan logika sebelum dan sesudah next dapat dipanggil:

async def logging_function_middleware(
    context: FunctionInvocationContext,
    next: Callable[[FunctionInvocationContext], Awaitable[None]],
) -> None:
    """Function middleware that logs function execution."""
    # Pre-processing: Log before function execution
    print(f"[Function] Calling {context.function.name}")

    # Continue to next middleware or function execution
    await next(context)

    # Post-processing: Log after function execution
    print(f"[Function] {context.function.name} completed")

Middleware Obrolan

Middleware obrolan mencegat permintaan obrolan yang dikirim ke model AI. Ini menggunakan yang ChatContext berisi:

  • chat_client: Klien obrolan yang dipanggil
  • messages: Daftar pesan yang dikirim ke layanan AI
  • options: Opsi untuk permintaan obrolan
  • is_streaming: Boolean menunjukkan apakah ini adalah pemanggilan streaming
  • metadata: Kamus untuk menyimpan data tambahan antar middleware
  • result: Respons obrolan dari AI (dapat dimodifikasi)
  • terminate: Bendera untuk menghentikan pemrosesan lebih lanjut
  • kwargs: Argumen kata kunci tambahan diteruskan ke klien obrolan

Yang next dapat dipanggil berlanjut ke middleware berikutnya atau mengirim permintaan ke layanan AI.

Berikut adalah contoh pengelogan sederhana dengan logika sebelum dan sesudah next dapat dipanggil:

async def logging_chat_middleware(
    context: ChatContext,
    next: Callable[[ChatContext], Awaitable[None]],
) -> None:
    """Chat middleware that logs AI interactions."""
    # Pre-processing: Log before AI call
    print(f"[Chat] Sending {len(context.messages)} messages to AI")

    # Continue to next middleware or AI service
    await next(context)

    # Post-processing: Log after AI response
    print("[Chat] AI response received")

Dekorator Middleware Fungsi

Dekorator menyediakan deklarasi jenis middleware eksplisit tanpa memerlukan anotasi jenis. Mereka membantu ketika:

  • Anda tidak menggunakan anotasi jenis
  • Anda memerlukan deklarasi jenis middleware eksplisit
  • Anda ingin mencegah ketidakcocokan tipe
from agent_framework import agent_middleware, function_middleware, chat_middleware

@agent_middleware  # Explicitly marks as agent middleware
async def simple_agent_middleware(context, next):
    """Agent middleware with decorator - types are inferred."""
    print("Before agent execution")
    await next(context)
    print("After agent execution")

@function_middleware  # Explicitly marks as function middleware
async def simple_function_middleware(context, next):
    """Function middleware with decorator - types are inferred."""
    print(f"Calling function: {context.function.name}")
    await next(context)
    print("Function call completed")

@chat_middleware  # Explicitly marks as chat middleware
async def simple_chat_middleware(context, next):
    """Chat middleware with decorator - types are inferred."""
    print(f"Processing {len(context.messages)} chat messages")
    await next(context)
    print("Chat processing completed")

Class-Based Middleware

Middleware berbasis kelas berguna untuk operasi stateful atau logika kompleks yang mendapat manfaat dari pola desain berorientasi objek.

Kelas Middleware Agen

Middleware agen berbasis kelas menggunakan process metode yang memiliki tanda tangan dan perilaku yang sama dengan middleware berbasis fungsi. Metode ini process menerima parameter dan context yang sama next dan dipanggil dengan cara yang sama persis.

from agent_framework import AgentMiddleware, AgentRunContext

class LoggingAgentMiddleware(AgentMiddleware):
    """Agent middleware that logs execution."""

    async def process(
        self,
        context: AgentRunContext,
        next: Callable[[AgentRunContext], Awaitable[None]],
    ) -> None:
        # Pre-processing: Log before agent execution
        print("[Agent Class] Starting execution")

        # Continue to next middleware or agent execution
        await next(context)

        # Post-processing: Log after agent execution
        print("[Agent Class] Execution completed")

Kelas Middleware Fungsi

Middleware fungsi berbasis kelas juga menggunakan process metode dengan tanda tangan dan perilaku yang sama dengan middleware berbasis fungsi. Metode ini menerima parameter dan context yang samanext.

from agent_framework import FunctionMiddleware, FunctionInvocationContext

class LoggingFunctionMiddleware(FunctionMiddleware):
    """Function middleware that logs function execution."""

    async def process(
        self,
        context: FunctionInvocationContext,
        next: Callable[[FunctionInvocationContext], Awaitable[None]],
    ) -> None:
        # Pre-processing: Log before function execution
        print(f"[Function Class] Calling {context.function.name}")

        # Continue to next middleware or function execution
        await next(context)

        # Post-processing: Log after function execution
        print(f"[Function Class] {context.function.name} completed")

Kelas Middleware Obrolan

Middleware obrolan berbasis kelas mengikuti pola yang sama dengan process metode yang memiliki tanda tangan dan perilaku yang identik dengan middleware obrolan berbasis fungsi.

from agent_framework import ChatMiddleware, ChatContext

class LoggingChatMiddleware(ChatMiddleware):
    """Chat middleware that logs AI interactions."""

    async def process(
        self,
        context: ChatContext,
        next: Callable[[ChatContext], Awaitable[None]],
    ) -> None:
        # Pre-processing: Log before AI call
        print(f"[Chat Class] Sending {len(context.messages)} messages to AI")

        # Continue to next middleware or AI service
        await next(context)

        # Post-processing: Log after AI response
        print("[Chat Class] AI response received")

Pendaftaran Middleware

Middleware dapat didaftarkan pada dua tingkat dengan cakupan dan perilaku yang berbeda.

Middleware Agent-Level vs Run-Level

from agent_framework.azure import AzureAIAgentClient
from azure.identity.aio import AzureCliCredential

# Agent-level middleware: Applied to ALL runs of the agent
async with AzureAIAgentClient(async_credential=credential).as_agent(
    name="WeatherAgent",
    instructions="You are a helpful weather assistant.",
    tools=get_weather,
    middleware=[
        SecurityAgentMiddleware(),  # Applies to all runs
        TimingFunctionMiddleware(),  # Applies to all runs
    ],
) as agent:

    # This run uses agent-level middleware only
    result1 = await agent.run("What's the weather in Seattle?")

    # This run uses agent-level + run-level middleware
    result2 = await agent.run(
        "What's the weather in Portland?",
        middleware=[  # Run-level middleware (this run only)
            logging_chat_middleware,
        ]
    )

    # This run uses agent-level middleware only (no run-level)
    result3 = await agent.run("What's the weather in Vancouver?")

Perbedaan Utama:

  • Tingkat agen: Persisten di semua eksekusi, dikonfigurasi sekali saat membuat agen
  • Tingkat eksekusi: Hanya diterapkan ke eksekusi tertentu, memungkinkan penyesuaian per permintaan
  • Urutan Eksekusi: Middleware agen (terluar) → Jalankan middleware (terdahulu) → eksekusi Agen

Penghentian Middleware

Middleware dapat mengakhiri eksekusi lebih awal menggunakan context.terminate. Ini berguna untuk pemeriksaan keamanan, pembatasan tarif, atau kegagalan validasi.

async def blocking_middleware(
    context: AgentRunContext,
    next: Callable[[AgentRunContext], Awaitable[None]],
) -> None:
    """Middleware that blocks execution based on conditions."""
    # Check for blocked content
    last_message = context.messages[-1] if context.messages else None
    if last_message and last_message.text:
        if "blocked" in last_message.text.lower():
            print("Request blocked by middleware")
            context.terminate = True
            return

    # If no issues, continue normally
    await next(context)

Apa artinya penghentian:

  • Mengatur context.terminate = True sinyal bahwa pemrosesan harus berhenti
  • Anda dapat memberikan hasil kustom sebelum mengakhiri untuk memberikan umpan balik kepada pengguna
  • Eksekusi agen benar-benar dilewati ketika middleware berakhir

Penimpaan Hasil Middleware

Middleware dapat mengambil alih hasil dalam skenario non-streaming dan streaming, memungkinkan Anda untuk memodifikasi atau mengganti respons agen sepenuhnya.

Jenis hasil tergantung context.result pada apakah pemanggilan agen streaming atau non-streaming:

  • Non-streaming: context.result berisi AgentResponse dengan respons lengkap
  • Streaming: context.result berisi generator asinkron yang menghasilkan AgentResponseUpdate gugus

Anda dapat menggunakan context.is_streaming untuk membedakan antara skenario ini dan menangani penimpaan hasil dengan tepat.

async def weather_override_middleware(
    context: AgentRunContext,
    next: Callable[[AgentRunContext], Awaitable[None]]
) -> None:
    """Middleware that overrides weather results for both streaming and non-streaming."""

    # Execute the original agent logic
    await next(context)

    # Override results if present
    if context.result is not None:
        custom_message_parts = [
            "Weather Override: ",
            "Perfect weather everywhere today! ",
            "22°C with gentle breezes. ",
            "Great day for outdoor activities!"
        ]

        if context.is_streaming:
            # Streaming override
            async def override_stream() -> AsyncIterable[AgentResponseUpdate]:
                for chunk in custom_message_parts:
                    yield AgentResponseUpdate(contents=[TextContent(text=chunk)])

            context.result = override_stream()
        else:
            # Non-streaming override
            custom_message = "".join(custom_message_parts)
            context.result = AgentResponse(
                messages=[ChatMessage(role=Role.ASSISTANT, text=custom_message)]
            )

Pendekatan middleware ini memungkinkan Anda menerapkan transformasi respons canggih, pemfilteran konten, peningkatan hasil, dan kustomisasi streaming sambil menjaga logika agen Anda tetap bersih dan fokus.

Langkah selanjutnya

Respons Latar Belakang Agen

  • Last updated on