Intergiciel de l’agent

L’intergiciel dans Agent Framework offre un moyen puissant d’intercepter, de modifier et d’améliorer les interactions de l’agent à différentes étapes de l’exécution. Vous pouvez utiliser l’intergiciel pour implémenter des préoccupations croisées telles que la journalisation, la validation de la sécurité, la gestion des erreurs et la transformation des résultats sans modifier votre logique principale d’agent ou de fonction.

Agent Framework peut être personnalisé à l’aide de trois types différents d’intergiciels :

Intergiciel d’exécution de l’agent : autorise l’interception de toutes les exécutions de l’agent, afin que l’entrée et la sortie puissent être inspectées et/ou modifiées selon les besoins.
Intergiciel d’appel de fonction : autorise l’interception de tous les appels de fonction exécutés par l’agent, afin que l’entrée et la sortie puissent être inspectées et modifiées selon les besoins.
IChatClient middleware : autorise l’interception des appels à une IChatClient implémentation, où un agent utilise IChatClient pour les appels d’inférence, par exemple lors de l’utilisation ChatClientAgent.

Tous les types d’intergiciels sont implémentés via un rappel de fonction, et lorsque plusieurs instances d’intergiciels du même type sont inscrites, elles forment une chaîne, où chaque instance d’intergiciel est censée appeler la suivante dans la chaîne, via une chaîne fournie nextFunc.

Les types d’intergiciels d’exécution et de fonction de l’agent peuvent être inscrits sur un agent à l’aide du générateur d’agents avec un objet agent existant.

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

Important

Idéalement, les deux runFunc et runStreamingFunc doivent être fournis. Lorsque vous fournissez simplement l’intergiciel non en streaming, l’agent l’utilisera pour les appels de diffusion en continu et de non-diffusion en continu. La diffusion en continu s’exécute uniquement en mode non streaming pour suffire aux attentes des intergiciels.

Note

Il existe une surcharge supplémentaire, Use(sharedFunc: ...)qui vous permet de fournir le même intergiciel pour le non-streaming et la diffusion en continu sans bloquer la diffusion en continu. Toutefois, le middleware partagé ne pourra pas intercepter ou remplacer la sortie. Cette surcharge doit être utilisée pour les scénarios où vous devez uniquement inspecter ou modifier l’entrée avant d’atteindre l’agent.

IChatClient l’intergiciel peut être inscrit sur un IChatClient intergiciel avant d’être utilisé avec un ChatClientAgentmodèle de générateur client de conversation.

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

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

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

Avertissement

DefaultAzureCredential est pratique pour le développement, mais nécessite une considération minutieuse en production. En production, envisagez d’utiliser des informations d’identification spécifiques (par exemple ManagedIdentityCredential) pour éviter les problèmes de latence, la détection involontaire des informations d’identification et les risques de sécurité potentiels liés aux mécanismes de secours.

IChatClient le middleware peut également être inscrit à l’aide d’une méthode de fabrique lors de la construction d’un agent via l’une des méthodes d’assistance des clients SDK.

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

Middleware d’exécution de l’agent

Voici un exemple d’intergiciel d’exécution d’agent, qui peut inspecter et/ou modifier l’entrée et la sortie de l’exécution de l’agent.

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

Intergiciel d’exécution de streaming de l’agent

Voici un exemple d’intergiciel d’exécution d’agent, qui peut inspecter et/ou modifier l’entrée et la sortie de l’exécution de streaming de l’agent.

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

Intergiciel appelant la fonction

Note

Le middleware d’appel de fonction est actuellement pris en charge uniquement avec un AIAgent qui utilise FunctionInvokingChatClient, par exemple ChatClientAgent.

Voici un exemple d’intergiciel appelant une fonction, qui peut inspecter et/ou modifier la fonction appelée et le résultat de l’appel de fonction.

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

Il est possible d’arrêter la boucle d’appel de fonction avec l’intergiciel d’appel de fonction en définissant la valeur true fournie FunctionInvocationContext.Terminate . Cela empêche la boucle d’appel de fonction d’émettre une requête au service d’inférence contenant les résultats de l’appel de fonction après l’appel de fonction. S’il existe plusieurs fonctions disponibles pour l’appel pendant cette itération, il peut également empêcher l’exécution des fonctions restantes.

Avertissement

La fin de la boucle d’appel de fonction peut entraîner la gauche de votre historique des conversations dans un état incohérent, par exemple, contenant du contenu d’appel de fonction sans contenu de résultat de fonction. Cela peut entraîner l’inutilisable de l’historique des conversations pour d’autres exécutions.

Intergiciel IChatClient

Voici un exemple d’intergiciel client de conversation, qui peut inspecter et/ou modifier l’entrée et la sortie de la demande au service d’inférence fourni par le client de conversation.

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

Conseil / Astuce

Consultez les exemples .NET pour obtenir des exemples exécutables complets.

Note

Pour plus d’informations sur IChatClient middleware, consultez l’intergiciel personnalisé IChatClient.

Agent Framework peut être personnalisé à l’aide de trois types différents d’intergiciels :

Intergiciel de l’agent : intercepte l’exécution de l’agent, ce qui vous permet d’inspecter et de modifier les entrées, les sorties et le flux de contrôle.
Intergiciel de fonction : intercepte les appels de fonction (outil) effectués pendant l’exécution de l’agent, en activant la validation d’entrée, la transformation des résultats et le contrôle d’exécution.
Intergiciel de conversation : intercepte les demandes de conversation sous-jacentes envoyées aux modèles IA, en fournissant l’accès aux messages bruts, aux options et aux réponses.

Tous les types prennent en charge les implémentations basées sur des fonctions et basées sur des classes. Lorsque plusieurs intergiciels du même type sont inscrits, ils forment une chaîne où chacun appelle l’intergiciel pour continuer le next traitement.

Intergiciel de l’agent

L’intergiciel agent intercepte et modifie l’exécution de l’exécution de l’agent. Il utilise les AgentContext éléments suivants :

agent: l’agent appelé
messages: liste des messages de conversation dans la conversation
is_streaming: booléen indiquant si la réponse est en streaming
metadata: Dictionnaire pour le stockage de données supplémentaires entre intergiciels
result: la réponse de l’agent (peut être modifiée)
terminate: indicateur pour arrêter le traitement ultérieur
kwargs: arguments de mot clé supplémentaires passés à la méthode d’exécution de l’agent

L’intergiciel next continue la chaîne d’intergiciels ou exécute l’agent s’il s’agit du dernier middleware.

Basé sur des fonctions

async def logging_agent_middleware(
    context: AgentContext,
    next: Callable[[AgentContext], 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")

Basé sur des classes

L’intergiciel d’agent basé sur une classe utilise une process méthode qui a la même signature et le même comportement que l’intergiciel basé sur les fonctions.

from agent_framework import AgentMiddleware, AgentContext

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

    async def process(
        self,
        context: AgentContext,
        next: Callable[[AgentContext], Awaitable[None]],
    ) -> None:
        print("[Agent Class] Starting execution")
        await next(context)
        print("[Agent Class] Execution completed")

Intergiciel de fonction

L’intergiciel de fonction intercepte les appels de fonction au sein des agents. Il utilise les FunctionInvocationContext éléments suivants :

function: fonction appelée
arguments: arguments validés pour la fonction
metadata: Dictionnaire pour le stockage de données supplémentaires entre intergiciels
result: valeur de retour de la fonction (peut être modifiée)
terminate: indicateur pour arrêter le traitement ultérieur
kwargs: arguments de mot clé supplémentaires passés à la méthode de conversation qui a appelé cette fonction

L’appelant next passe au middleware suivant ou exécute la fonction réelle.

Basé sur des fonctions

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")

Basé sur des classes

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:
        print(f"[Function Class] Calling {context.function.name}")
        await next(context)
        print(f"[Function Class] {context.function.name} completed")

Intergiciel de conversation

L’intergiciel de conversation intercepte les demandes de conversation envoyées aux modèles IA. Il utilise les ChatContext éléments suivants :

chat_client: le client de conversation appelé
messages: liste des messages envoyés au service IA
options: options de la demande de conversation
is_streaming: booléen indiquant s’il s’agit d’un appel de diffusion en continu
metadata: Dictionnaire pour le stockage de données supplémentaires entre intergiciels
result: Réponse de conversation de l’IA (peut être modifiée)
terminate: indicateur pour arrêter le traitement ultérieur
kwargs: arguments de mot clé supplémentaires passés au client de conversation

L’appelant next continue à l’intergiciel suivant ou envoie la requête au service IA.

Basé sur des fonctions

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")

Basé sur des classes

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:
        print(f"[Chat Class] Sending {len(context.messages)} messages to AI")
        await next(context)
        print("[Chat Class] AI response received")

Décorateurs d’intergiciels

Les décorateurs fournissent une déclaration de type middleware explicite sans nécessiter d’annotations de type. Ils sont utiles lorsque vous n’utilisez pas d’annotations de type ou que vous souhaitez empêcher les incompatibilités de type :

from agent_framework import agent_middleware, function_middleware, chat_middleware

@agent_middleware
async def simple_agent_middleware(context, next):
    print("Before agent execution")
    await next(context)
    print("After agent execution")

@function_middleware
async def simple_function_middleware(context, next):
    print(f"Calling function: {context.function.name}")
    await next(context)
    print("Function call completed")

@chat_middleware
async def simple_chat_middleware(context, next):
    print(f"Processing {len(context.messages)} chat messages")
    await next(context)
    print("Chat processing completed")

Inscription d’intergiciel

Les intergiciels peuvent être inscrits à deux niveaux avec des étendues et des comportements différents.

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?")

Principales différences :

Niveau de l’agent : persistant sur toutes les exécutions, configuré une fois lors de la création de l’agent
Niveau d’exécution : appliqué uniquement aux exécutions spécifiques, autorise la personnalisation par requête
Ordre d’exécution : Intergiciel agent (externe) → Exécuter l’intergiciel (le plus interne) → Exécution de l’agent

Arrêt du middleware

L’intergiciel peut arrêter l’exécution tôt à l’aide context.terminatede . Cela est utile pour les vérifications de sécurité, la limitation du débit ou les échecs de validation.

async def blocking_middleware(
    context: AgentContext,
    next: Callable[[AgentContext], 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)

Ce que signifie l’arrêt :

Définition des context.terminate = True signaux que le traitement doit arrêter
Vous pouvez fournir un résultat personnalisé avant de terminer pour donner des commentaires aux utilisateurs
L’exécution de l’agent est complètement ignorée lorsque l’intergiciel se termine

Remplacement des résultats du middleware

Le middleware peut remplacer les résultats dans les scénarios de diffusion en continu et de diffusion en continu, ce qui vous permet de modifier ou de remplacer complètement les réponses de l’agent.

Le type context.result de résultat varie selon que l’appel de l’agent est en streaming ou non en streaming :

Non-streaming : context.result contient une AgentResponse réponse complète
Streaming : context.result contient un générateur asynchrone qui génère AgentResponseUpdate des blocs

Vous pouvez utiliser context.is_streaming pour différencier ces scénarios et gérer les remplacements de résultats de manière appropriée.

async def weather_override_middleware(
    context: AgentContext,
    next: Callable[[AgentContext], 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=[Content.from_text(text=chunk)])

            context.result = override_stream()
        else:
            # Non-streaming override
            custom_message = "".join(custom_message_parts)
            context.result = AgentResponse(
                messages=[Message(role="assistant", contents=[custom_message])]
            )

Cette approche d’intergiciel vous permet d’implémenter une transformation de réponse sophistiquée, un filtrage de contenu, une amélioration des résultats et une personnalisation du streaming tout en conservant la logique de votre agent propre et prioritaire.

Exemples complets d’intergiciels

Intergiciel basé sur des classes

# 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.azure import AzureAIAgentClient
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,
        AzureAIAgentClient(credential=credential).as_agent(
            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())

Intergiciel basé sur des fonctions

# 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.azure import AzureAIAgentClient
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,
        AzureAIAgentClient(credential=credential).as_agent(
            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())

Intergiciel basé sur le décorateur

# 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.azure import AzureAIAgentClient
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,
        AzureAIAgentClient(credential=credential).as_agent(
            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())

Prochaines étapes

Réponses en arrière-plan de l’agent

Commentaires

Cette page a-t-elle été utile ?

Last updated on 2026-02-13

Partager via

Intergiciel de l’agent

Middleware d’exécution de l’agent

Intergiciel d’exécution de streaming de l’agent

Intergiciel appelant la fonction

Intergiciel IChatClient

Intergiciel de l’agent

Basé sur des fonctions

Basé sur des classes

Intergiciel de fonction

Basé sur des fonctions

Basé sur des classes

Intergiciel de conversation

Basé sur des fonctions

Basé sur des classes

Décorateurs d’intergiciels

Inscription d’intergiciel

middleware Agent-Level vs Run-Level

Arrêt du middleware

Remplacement des résultats du middleware

Exemples complets d’intergiciels

Intergiciel basé sur des classes

Intergiciel basé sur des fonctions

Intergiciel basé sur le décorateur

Prochaines étapes

Commentaires

Ressources supplémentaires