ПО агента-посредника

По промежуточному слоям в Agent Framework предоставляет эффективный способ перехвата, изменения и улучшения взаимодействия агентов на различных этапах выполнения. Вы можете использовать ПО промежуточного слоя для реализации перекрестных проблем, таких как ведение журнала, проверка безопасности, обработка ошибок и преобразование результатов без изменения основного агента или логики функции.

Платформа агента может быть настроена с помощью трех различных типов ПО промежуточного слоя:

ПО промежуточного слоя запуска агента: позволяет перехватывать все запуски агента, чтобы входные и выходные данные могли быть проверены и (или) изменены по мере необходимости.
ПО промежуточного слоя вызовов функций: позволяет перехватывать все вызовы функций, выполняемые агентом, чтобы входные и выходные данные могли быть проверены и изменены по мере необходимости.
IChatClient ПО промежуточного слоя: позволяет перехватывать вызовы реализации IChatClient , где агент используется IChatClient для вызовов вывода, например при использовании ChatClientAgent.

Все типы по промежуточного слоя реализуются с помощью обратного вызова функции, и при регистрации нескольких экземпляров по промежуточного слоя одного типа они образуют цепочку, где каждый экземпляр ПО промежуточного слоя должен вызывать следующий в цепочке через предоставленный nextFunc.

Типы по промежуточному слоям запуска агента и функции можно зарегистрировать в агенте с помощью построителя агентов с существующим объектом агента.

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

Это важно

В идеале runFuncrunStreamingFunc и должны быть предоставлены. При предоставлении только по промежуточного слоя, отличного от потоковой передачи, агент будет использовать его как для потоковых, так и для непотоковых вызовов. Потоковая передача будет выполняться только в режиме без потоковой передачи, чтобы достаточно ожиданий ПО промежуточного слоя.

Замечание

Существует дополнительная перегрузка, Use(sharedFunc: ...)которая позволяет предоставлять одно и то же ПО промежуточного слоя для непотоковой и потоковой передачи без блокировки потоковой передачи. Однако общий ПО промежуточного слоя не сможет перехватывать или переопределять выходные данные. Эту перегрузку следует использовать для сценариев, когда необходимо проверить или изменить входные данные, прежде чем он достигнет агента.

IChatClient по промежуточному слоям можно зарегистрировать на сервере IChatClientChatClientAgentдо его использования с помощью шаблона построитель клиентов чата.

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

Предупреждение

DefaultAzureCredential удобно для разработки, но требует тщательного рассмотрения в рабочей среде. В рабочей среде рекомендуется использовать определенные учетные данные (например, ManagedIdentityCredential), чтобы избежать проблем с задержкой, непреднамеренного проверки учетных данных и потенциальных рисков безопасности из резервных механизмов.

IChatClient Промежуточное программное обеспечение также можно зарегистрировать, используя фабричный метод при создании агента с помощью одного из вспомогательных методов с клиентами 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());

По промежуточному слоям запуска агента

Ниже приведен пример по промежуточного слоя запуска агента, который может проверять и /или изменять входные и выходные данные из запуска агента.

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

ПО промежуточного слоя потоковой передачи агента

Ниже приведен пример запуска по промежуточного слоя потоковой передачи агента, который может проверять и /или изменять входные и выходные данные из запуска потоковой передачи агента.

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

По промежуточному слоям вызовов функций

Замечание

Промежуточное программное обеспечение для вызова функций в настоящее время поддерживается только с AIAgent, с использованием FunctionInvokingChatClient, например, ChatClientAgent.

Ниже приведен пример промежуточного слоя вызовов функций, который может проверять и/или изменять вызываемую функцию, а также результат её вызова.

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

Можно завершить цикл вызова функции с помощью по промежуточного слоя вызова функции, задав предоставленное FunctionInvocationContext.Terminate значение true. Это приведет к тому, что цикл вызова функции не будет выдавать запрос в службу вывода, содержащую результаты вызова функции после вызова функции. Если во время итерации было доступно несколько функций, они также могут предотвратить выполнение оставшихся функций.

Предупреждение

Завершение цикла вызова функции может привести к тому, что журнал чата остается в несогласованном состоянии, например, содержащий содержимое вызова функции без содержимого результата функции. Это может привести к тому, что журнал чата непригоден для дальнейших запусков.

ПО промежуточного слоя IChatClient

Ниже приведен пример промежуточного слоя клиента чата, который может проверять и/или изменять входные и выходные данные для запроса в службу вывода заключений, предоставляемую клиентом чата.

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

Подсказка

Полные примеры запуска см. в примерах .NET .

Замечание

Для получения дополнительной информации о промежуточном ПО, смотрите IChatClient.

Платформа агента может быть настроена с помощью трех различных типов ПО промежуточного слоя:

ПО промежуточного слоя агента: перехватывает выполнение агента, позволяя проверять и изменять входные данные, выходные данные и поток управления.
ПО промежуточного слоя функций: перехват вызовов функции (средства), выполняемых во время выполнения агента, включение проверки входных данных, преобразования результатов и управления выполнением.
По промежуточному слоя чата: перехватывает базовые запросы чата, отправленные моделям ИИ, предоставляя доступ к необработанным сообщениям, параметрам и ответам.

Все типы поддерживают реализации на основе функций и на основе классов. При регистрации нескольких по промежуточного слоя одного типа они образуют цепочку, в которой каждый вызывает next вызываемую для продолжения обработки.

Замечание

Порядок по промежуточного слоя с смешанными областями регистрации:

ПО промежуточного слоя уровня агента упаковывает по промежуточному по промежуточному слоям уровня запуска.
Для по промежуточного [A1, A2] слоя агента и по промежуточному слоям [R1, R2]запуска выполняется порядок выполнения: A1 -> A2 -> R1 -> R2 -> Agent -> R2 -> R1 -> A2 -> A1
ПО промежуточного слоя функций и чата следует тому же принципу упаковки во время вызова средства или чата.

ПО агента-посредника

ПО промежуточного слоя агента перехватывает и изменяет выполнение агента. Он использует те AgentContext , которые содержат:

agent: вызываемый агент
messages: список сообщений чата в беседе
is_streaming: логическое значение, указывающее, является ли ответ потоковым.
metadata: словарь для хранения дополнительных данных между ПО промежуточного слоя
result: ответ агента (можно изменить)
terminate: флаг для остановки дальнейшей обработки
kwargs: дополнительные аргументы ключевых слов, переданные методу запуска агента

Вызываемая next программа продолжает цепочку ПО промежуточного слоя или выполняет агент, если это последнее ПО промежуточного слоя.

На основе функций

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

На основе классов

ПО промежуточного слоя агента на основе классов использует process метод, имеющий ту же сигнатуру и поведение, что и ПО промежуточного слоя на основе функций.

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

ПО промежуточного слоя функций

ПО промежуточного слоя функций перехватывает вызовы функций в агентах. Он использует те FunctionInvocationContext , которые содержат:

function: вызываемая функция
arguments: проверенные аргументы для функции
metadata: словарь для хранения дополнительных данных между ПО промежуточного слоя
result: возвращаемое значение функции (можно изменить)
terminate: флаг для остановки дальнейшей обработки
kwargs: дополнительные аргументы ключевых слов, переданные методу чата, который вызвал эту функцию

Вызывающий next объект продолжает выполнять следующую ПО промежуточного слоя или выполняет фактическую функцию.

На основе функций

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

На основе классов

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

ПО промежуточного слоя чата

ПО промежуточного слоя чата перехватывает запросы чата, отправленные моделям ИИ. Он использует те ChatContext , которые содержат:

chat_client: вызываемый клиент чата
messages: список сообщений, отправляемых в службу ИИ
options: параметры запроса чата
is_streaming: логическое значение, указывающее, является ли это вызов потоковой передачи
metadata: словарь для хранения дополнительных данных между ПО промежуточного слоя
result: ответ чата из ИИ (можно изменить)
terminate: флаг для остановки дальнейшей обработки
kwargs: дополнительные аргументы ключевых слов, переданные клиенту чата

Вызываемое next по промежуточному слоя продолжается до следующего по промежуточного слоя или отправляет запрос в службу ИИ.

На основе функций

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

На основе классов

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

Декораторы по промежуточного слоя

Декораторы предоставляют явное объявление типа ПО промежуточного слоя без необходимости примечания типов. Они полезны, если вы не используете заметки типа или хотите предотвратить несоответствие типов:

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

Регистрация ПО промежуточного слоя

ПО промежуточного слоя можно зарегистрировать на двух уровнях с различными областями и поведением.

Agent-Level и по промежуточному поверх 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?")

Ключевые отличия:

Уровень агента: постоянный во всех запусках, настроенный один раз при создании агента
Уровень выполнения: применяется только к определенным запускам, разрешает настройку каждого запроса.
Порядок выполнения: ПО промежуточного слоя агента (внешняя) → выполнение ПО промежуточного слоя (внутренний) → агента

Завершение по промежуточного слоя

ПО промежуточного слоя может завершить выполнение рано с помощью context.terminate. Это полезно для проверок безопасности, ограничения скорости или сбоев проверки.

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)

Что означает завершение:

Настройка context.terminate = True сигналов о том, что обработка должна остановиться
Вы можете предоставить пользовательский результат, прежде чем завершать отправку отзывов пользователей
Выполнение агента полностью пропускается при завершении по промежуточного слоя

Переопределение результата по промежуточного слоя

По промежуточному слоям можно переопределить результаты как в сценариях, не являющихся потоковой передачей, так и в сценариях потоковой передачи, что позволяет изменять или полностью заменить ответы агента.

Тип context.result результата зависит от того, является ли вызов агента потоковой или непотоковой:

Непотоковые: context.result содержит полный AgentResponse ответ
Потоковая передача: context.result содержит асинхронный генератор, который дает AgentResponseUpdate блоки

Вы можете различать context.is_streaming эти сценарии и обрабатывать переопределения результатов соответствующим образом.

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

Этот подход по промежуточному по промежуточному слоя позволяет реализовать сложные преобразования ответа, фильтрацию содержимого, улучшение результатов и настройку потоковой передачи при сохранении логики агента в чистом и ориентированном режиме.

Полные примеры по промежуточного слоя

ПО промежуточного слоя на основе классов

# 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())

ПО промежуточного слоя на основе функций

# 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())

ПО промежуточного слоя на основе декоратора

# 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())

Дальнейшие шаги

Фоновые ответы агента

Обратная связь

Были ли сведения на этой странице полезными?

Last updated on 2026-02-20

Поделиться через

ПО агента-посредника

По промежуточному слоям запуска агента

ПО промежуточного слоя потоковой передачи агента

По промежуточному слоям вызовов функций

ПО промежуточного слоя IChatClient

ПО агента-посредника

На основе функций

На основе классов

ПО промежуточного слоя функций

На основе функций

На основе классов

ПО промежуточного слоя чата

На основе функций

На основе классов

Декораторы по промежуточного слоя

Регистрация ПО промежуточного слоя

Agent-Level и по промежуточному поверх Run-Level

Завершение по промежуточного слоя

Переопределение результата по промежуточного слоя

Полные примеры по промежуточного слоя

ПО промежуточного слоя на основе классов

ПО промежуточного слоя на основе функций

ПО промежуточного слоя на основе декоратора

Дальнейшие шаги

Обратная связь

Дополнительные ресурсы