Menambahkan Middleware ke Agen

Pelajari cara menambahkan middleware ke agen Anda dalam beberapa langkah sederhana. Middleware memungkinkan Anda untuk mencegat dan memodifikasi interaksi agen untuk pencatatan, keamanan, dan kepentingan lintas fungsi lainnya.

Prasyarat

Untuk prasyarat dan menginstal paket NuGet, lihat langkah Membuat dan menjalankan agen sederhana dalam tutorial ini.

Langkah 1: Buat Agen Sederhana

Pertama, buat agen dasar dengan alat fungsi.

using System;
using System.ComponentModel;
using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;

[Description("The current datetime offset.")]
static string GetDateTime()
    => DateTimeOffset.Now.ToString();

AIAgent baseAgent = new AIProjectClient(
    new Uri("<your-foundry-project-endpoint>"),
    new DefaultAzureCredential())
        .AsAIAgent(
            model: "gpt-4o-mini",
            instructions: "You are an AI assistant that helps people find information.",
            tools: [AIFunctionFactory.Create(GetDateTime, name: nameof(GetDateTime))]);

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.

Langkah 2: Buat Middleware untuk Menjalankan Agen Anda

Selanjutnya, buat fungsi yang akan dipanggil untuk setiap agen yang dijalankan. Ini memungkinkan Anda untuk memeriksa input dan output dari agen.

Kecuali niatnya adalah menggunakan middleware untuk berhenti menjalankan eksekusi, fungsi harus memanggil RunAsync pada yang disediakan innerAgent.

Middleware sampel ini hanya memeriksa input dan output dari agen yang dijalankan dan menghasilkan jumlah pesan yang diteruskan ke dalam dan ke luar agen.

using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

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

Langkah 3: Tambahkan Agen Jalankan Middleware ke Agen Anda

Gunakan pola pembangunan untuk menambahkan fungsi middleware ini ke dalam baseAgent yang Anda buat di langkah 1. Ini membuat agen baru yang menerapkan middleware. baseAgent Aslinya tidak dimodifikasi.

var middlewareEnabledAgent = baseAgent
    .AsBuilder()
        .Use(runFunc: CustomAgentRunMiddleware, runStreamingFunc: null)
    .Build();

Sekarang, saat menjalankan agen perangkat lunak dengan kueri, middleware harus dipanggil, menghasilkan jumlah pesan input dan jumlah pesan respons.

Console.WriteLine(await middlewareEnabledAgent.RunAsync("What's the current time?"));

Langkah 4: Buat Function Calling Middleware

Nota

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

Anda juga dapat membuat middleware yang dipanggil setiap kali fungsi alat dijalankan. Berikut adalah contoh middleware pemanggilan fungsi yang dapat memeriksa dan/atau memodifikasi fungsi yang dipanggil dan hasil dari panggilan fungsi.

Kecuali niatnya adalah menggunakan middleware untuk tidak menjalankan alat fungsi, middleware harus memanggil yang disediakan nextFunc.

using System.Threading;
using System.Threading.Tasks;

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;
}

Langkah 5: Tambahkan Middleware pemanggilan fungsi ke agen Anda

Sama seperti menambahkan middleware yang dijalankan oleh agen, Anda dapat menambahkan middleware untuk memanggil fungsi sebagai berikut:

var middlewareEnabledAgent = baseAgent
    .AsBuilder()
        .Use(CustomFunctionCallingMiddleware)
    .Build();

Sekarang, saat menjalankan agen dengan kueri yang memanggil fungsi, middleware harus dipanggil, menghasilkan nama fungsi dan hasil panggilan.

Console.WriteLine(await middlewareEnabledAgent.RunAsync("What's the current time?"));

Langkah 6: Buat Middleware Klien Percakapan

Untuk agen yang dibangun menggunakan IChatClient, Anda mungkin ingin mencegat panggilan dari agen ke IChatClient. Dalam hal ini, dimungkinkan untuk menggunakan middleware untuk 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.

using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

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

    return response;
}

Nota

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

Langkah 7: Tambahkan Middleware untuk klien Obrolan ke `IChatClient`

Jika Anda ingin menambahkan middleware ke IChatClient, Anda dapat menggunakan pola builder. Setelah menambahkan middleware, Anda dapat menggunakan IChatClient dengan agen Anda seperti biasa.

var chatClient = new AIProjectClient(
    new Uri("<your-foundry-project-endpoint>"),
    new DefaultAzureCredential())
        .GetProjectOpenAIClient()
        .GetProjectResponsesClient()
        .AsIChatClient("gpt-4o-mini");

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 AIProjectClient(
    new Uri("<your-foundry-project-endpoint>"),
    new DefaultAzureCredential())
        .AsAIAgent(
            model: "gpt-4o-mini",
            instructions: "You are a helpful assistant.",
            clientFactory: (chatClient) => chatClient
                .AsBuilder()
                    .Use(getResponseFunc: CustomChatClientMiddleware, getStreamingResponseFunc: null)
                .Build());

Langkah 1: Buat Agen Sederhana

Pertama, buat agen dasar:

import asyncio
from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from azure.identity.aio import AzureCliCredential

async def main():
    credential = AzureCliCredential()

    async with Agent(

        client=FoundryChatClient(credential=credential),
        name="GreetingAgent",
        instructions="You are a friendly greeting assistant.",
    ) as agent:
        result = await agent.run("Hello!")
        print(result.text)

if __name__ == "__main__":
    asyncio.run(main())

Langkah 2: Buat Middleware Anda

Buat middleware pengelogan sederhana untuk melihat kapan agen Anda berjalan:

from collections.abc import Awaitable, Callable

from agent_framework import AgentContext

async def logging_agent_middleware(
    context: AgentContext,
    call_next: Callable[[], Awaitable[None]],
) -> None:
    """Simple middleware that logs agent execution."""
    print("Agent starting...")

    # Continue to agent execution
    await call_next()

    print("Agent finished!")

Langkah 3: Tambahkan Middleware ke Agen Anda

Tambahkan middleware saat membuat agen Anda:

async def main():
    credential = AzureCliCredential()

    async with Agent(

        client=FoundryChatClient(credential=credential),
        name="GreetingAgent",
        instructions="You are a friendly greeting assistant.",
        middleware=[logging_agent_middleware],  # Add your middleware here
    ) as agent:
        result = await agent.run("Hello!")
        print(result.text)

Langkah 4: Buat Middleware Fungsi

Jika agen Anda menggunakan fungsi, Anda dapat mencegat panggilan fungsi dan mengatur nilai runtime khusus alat sebelum alat dijalankan:

from collections.abc import Awaitable, Callable

from agent_framework import FunctionInvocationContext

def get_time(ctx: FunctionInvocationContext) -> str:
    """Get the current time."""
    from datetime import datetime
    source = ctx.kwargs.get("request_source", "direct")
    return f"[{source}] {datetime.now().strftime('%H:%M:%S')}"

async def inject_function_kwargs(
    context: FunctionInvocationContext,
    call_next: Callable[[], Awaitable[None]],
) -> None:
    """Middleware that adds tool-only runtime values before execution."""
    context.kwargs.setdefault("request_source", "middleware")

    await call_next()

# Add both the function and middleware to your agent
async with Agent(
    client=FoundryChatClient(credential=credential),
    name="TimeAgent",
    instructions="You can tell the current time.",
    tools=[get_time],
    middleware=[inject_function_kwargs],
) as agent:
    result = await agent.run("What time is it?")

Langkah 5: Gunakan middleware Run-Level

Anda juga dapat menambahkan middleware untuk eksekusi tertentu:

# Use middleware for this specific run only
result = await agent.run(
    "This is important!",
    middleware=[logging_function_middleware]
)

Apa Selanjutnya?

Untuk skenario yang lebih canggih, lihat Panduan Pengguna Middleware Agen, yang mencakup:

Berbagai jenis middleware (agen, fungsi, obrolan).
Middleware berbasis kelas untuk skenario kompleks.
Penghentian proses middleware dan penggantian hasil.
Pola middleware tingkat lanjut dan praktik terbaik.

Contoh lengkap

Middleware berbasis kelas

# Copyright (c) Microsoft. All rights reserved.

import asyncio
import time
from collections.abc import Awaitable, Callable
from random import randint
from typing import Annotated

from agent_framework import (
    AgentContext,
    AgentMiddleware,
    AgentResponse,
    FunctionInvocationContext,
    FunctionMiddleware,
    Message,
    tool,
)
from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from azure.identity.aio import AzureCliCredential
from pydantic import Field

"""
Class-based MiddlewareTypes Example

This sample demonstrates how to implement middleware using class-based approach by inheriting
from AgentMiddleware and FunctionMiddleware base classes. The example includes:

- SecurityAgentMiddleware: Checks for security violations in user queries and blocks requests
  containing sensitive information like passwords or secrets
- LoggingFunctionMiddleware: Logs function execution details including timing and parameters

This approach is useful when you need stateful middleware or complex logic that benefits
from object-oriented design patterns.
"""


# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
    location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
    """Get the weather for a given location."""
    conditions = ["sunny", "cloudy", "rainy", "stormy"]
    return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."


class SecurityAgentMiddleware(AgentMiddleware):
    """Agent middleware that checks for security violations."""

    async def process(
        self,
        context: AgentContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        # Check for potential security violations in the query
        # Look at the last user message
        last_message = context.messages[-1] if context.messages else None
        if last_message and last_message.text:
            query = last_message.text
            if "password" in query.lower() or "secret" in query.lower():
                print("[SecurityAgentMiddleware] Security Warning: Detected sensitive information, blocking request.")
                # Override the result with warning message
                context.result = AgentResponse(
                    messages=[Message("assistant", ["Detected sensitive information, the request is blocked."])]
                )
                # Simply don't call call_next() to prevent execution
                return

        print("[SecurityAgentMiddleware] Security check passed.")
        await call_next()


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

    async def process(
        self,
        context: FunctionInvocationContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        function_name = context.function.name
        print(f"[LoggingFunctionMiddleware] About to call function: {function_name}.")

        start_time = time.time()

        await call_next()

        end_time = time.time()
        duration = end_time - start_time

        print(f"[LoggingFunctionMiddleware] Function {function_name} completed in {duration:.5f}s.")


async def main() -> None:
    """Example demonstrating class-based middleware."""
    print("=== Class-based MiddlewareTypes Example ===")

    # For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
    # authentication option.
    async with (
        AzureCliCredential() as credential,
        Agent(
            client=FoundryChatClient(credential=credential),
            name="WeatherAgent",
            instructions="You are a helpful weather assistant.",
            tools=get_weather,
            middleware=[SecurityAgentMiddleware(), LoggingFunctionMiddleware()],
        ) as agent,
    ):
        # Test with normal query
        print("\n--- Normal Query ---")
        query = "What's the weather like in Seattle?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")

        # Test with security-related query
        print("--- Security Test ---")
        query = "What's the password for the weather service?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")


if __name__ == "__main__":
    asyncio.run(main())

Middleware berbasis fungsi

# Copyright (c) Microsoft. All rights reserved.

import asyncio
import time
from collections.abc import Awaitable, Callable
from random import randint
from typing import Annotated

from agent_framework import (
    AgentContext,
    AgentMiddleware,
    AgentResponse,
    FunctionInvocationContext,
    FunctionMiddleware,
    Message,
    tool,
)
from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from azure.identity.aio import AzureCliCredential
from pydantic import Field

"""
Class-based MiddlewareTypes Example

This sample demonstrates how to implement middleware using class-based approach by inheriting
from AgentMiddleware and FunctionMiddleware base classes. The example includes:

- SecurityAgentMiddleware: Checks for security violations in user queries and blocks requests
  containing sensitive information like passwords or secrets
- LoggingFunctionMiddleware: Logs function execution details including timing and parameters

This approach is useful when you need stateful middleware or complex logic that benefits
from object-oriented design patterns.
"""


# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
    location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
    """Get the weather for a given location."""
    conditions = ["sunny", "cloudy", "rainy", "stormy"]
    return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."


class SecurityAgentMiddleware(AgentMiddleware):
    """Agent middleware that checks for security violations."""

    async def process(
        self,
        context: AgentContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        # Check for potential security violations in the query
        # Look at the last user message
        last_message = context.messages[-1] if context.messages else None
        if last_message and last_message.text:
            query = last_message.text
            if "password" in query.lower() or "secret" in query.lower():
                print("[SecurityAgentMiddleware] Security Warning: Detected sensitive information, blocking request.")
                # Override the result with warning message
                context.result = AgentResponse(
                    messages=[Message("assistant", ["Detected sensitive information, the request is blocked."])]
                )
                # Simply don't call call_next() to prevent execution
                return

        print("[SecurityAgentMiddleware] Security check passed.")
        await call_next()


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

    async def process(
        self,
        context: FunctionInvocationContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        function_name = context.function.name
        print(f"[LoggingFunctionMiddleware] About to call function: {function_name}.")

        start_time = time.time()

        await call_next()

        end_time = time.time()
        duration = end_time - start_time

        print(f"[LoggingFunctionMiddleware] Function {function_name} completed in {duration:.5f}s.")


async def main() -> None:
    """Example demonstrating class-based middleware."""
    print("=== Class-based MiddlewareTypes Example ===")

    # For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
    # authentication option.
    async with (
        AzureCliCredential() as credential,
        Agent(
            client=FoundryChatClient(credential=credential),
            name="WeatherAgent",
            instructions="You are a helpful weather assistant.",
            tools=get_weather,
            middleware=[SecurityAgentMiddleware(), LoggingFunctionMiddleware()],
        ) as agent,
    ):
        # Test with normal query
        print("\n--- Normal Query ---")
        query = "What's the weather like in Seattle?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")

        # Test with security-related query
        print("--- Security Test ---")
        query = "What's the password for the weather service?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")


if __name__ == "__main__":
    asyncio.run(main())

Middleware berbasis dekorator

# Copyright (c) Microsoft. All rights reserved.

import asyncio
import time
from collections.abc import Awaitable, Callable
from random import randint
from typing import Annotated

from agent_framework import (
    AgentContext,
    AgentMiddleware,
    AgentResponse,
    FunctionInvocationContext,
    FunctionMiddleware,
    Message,
    tool,
)
from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from azure.identity.aio import AzureCliCredential
from pydantic import Field

"""
Class-based MiddlewareTypes Example

This sample demonstrates how to implement middleware using class-based approach by inheriting
from AgentMiddleware and FunctionMiddleware base classes. The example includes:

- SecurityAgentMiddleware: Checks for security violations in user queries and blocks requests
  containing sensitive information like passwords or secrets
- LoggingFunctionMiddleware: Logs function execution details including timing and parameters

This approach is useful when you need stateful middleware or complex logic that benefits
from object-oriented design patterns.
"""


# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
    location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
    """Get the weather for a given location."""
    conditions = ["sunny", "cloudy", "rainy", "stormy"]
    return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."


class SecurityAgentMiddleware(AgentMiddleware):
    """Agent middleware that checks for security violations."""

    async def process(
        self,
        context: AgentContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        # Check for potential security violations in the query
        # Look at the last user message
        last_message = context.messages[-1] if context.messages else None
        if last_message and last_message.text:
            query = last_message.text
            if "password" in query.lower() or "secret" in query.lower():
                print("[SecurityAgentMiddleware] Security Warning: Detected sensitive information, blocking request.")
                # Override the result with warning message
                context.result = AgentResponse(
                    messages=[Message("assistant", ["Detected sensitive information, the request is blocked."])]
                )
                # Simply don't call call_next() to prevent execution
                return

        print("[SecurityAgentMiddleware] Security check passed.")
        await call_next()


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

    async def process(
        self,
        context: FunctionInvocationContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        function_name = context.function.name
        print(f"[LoggingFunctionMiddleware] About to call function: {function_name}.")

        start_time = time.time()

        await call_next()

        end_time = time.time()
        duration = end_time - start_time

        print(f"[LoggingFunctionMiddleware] Function {function_name} completed in {duration:.5f}s.")


async def main() -> None:
    """Example demonstrating class-based middleware."""
    print("=== Class-based MiddlewareTypes Example ===")

    # For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
    # authentication option.
    async with (
        AzureCliCredential() as credential,
        Agent(
            client=FoundryChatClient(credential=credential),
            name="WeatherAgent",
            instructions="You are a helpful weather assistant.",
            tools=get_weather,
            middleware=[SecurityAgentMiddleware(), LoggingFunctionMiddleware()],
        ) as agent,
    ):
        # Test with normal query
        print("\n--- Normal Query ---")
        query = "What's the weather like in Seattle?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")

        # Test with security-related query
        print("--- Security Test ---")
        query = "What's the password for the weather service?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")


if __name__ == "__main__":
    asyncio.run(main())

Langkah selanjutnya

Chat-Level Middleware

Saran dan Komentar

Apakah halaman ini membantu?

Last updated on 2026-04-01

Bagikan melalui

Menambahkan Middleware ke Agen

Prasyarat

Langkah 1: Buat Agen Sederhana

Langkah 2: Buat Middleware untuk Menjalankan Agen Anda

Langkah 3: Tambahkan Agen Jalankan Middleware ke Agen Anda

Langkah 4: Buat Function Calling Middleware

Langkah 5: Tambahkan Middleware pemanggilan fungsi ke agen Anda

Langkah 6: Buat Middleware Klien Percakapan

Langkah 7: Tambahkan Middleware untuk klien Obrolan ke IChatClient

Langkah 1: Buat Agen Sederhana

Langkah 2: Buat Middleware Anda

Langkah 3: Tambahkan Middleware ke Agen Anda

Langkah 4: Buat Middleware Fungsi

Langkah 5: Gunakan middleware Run-Level

Apa Selanjutnya?

Contoh lengkap

Middleware berbasis kelas

Middleware berbasis fungsi

Middleware berbasis dekorator

Langkah selanjutnya

Saran dan Komentar

Sumber Daya Tambahan:

Langkah 7: Tambahkan Middleware untuk klien Obrolan ke `IChatClient`