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

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

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

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

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

var middlewareEnabledAgent = originalAgent
    .AsBuilder()
        .Use(CustomAgentRunMiddleware)
        .Use(CustomFunctionCallingMiddleware)
    .Build();

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

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

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

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

IChatClient ПО промежуточного слоя также можно зарегистрировать с помощью метода фабрики при создании агента с помощью одного из вспомогательных методов на клиентах ПАКЕТА SDK.

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

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

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

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

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

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

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

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

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

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

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

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

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

Ниже приведен простой пример ведения журнала с логикой до и после next вызываемого вызова:

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

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

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

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

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

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

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

Ниже приведен простой пример ведения журнала с логикой до и после 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")

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

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

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

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

Ниже приведен простой пример ведения журнала с логикой до и после 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 agent_middleware, function_middleware, chat_middleware

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

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

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

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

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

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

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

from agent_framework import AgentMiddleware, AgentRunContext

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

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

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

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

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

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

from agent_framework import FunctionMiddleware, FunctionInvocationContext

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

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

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

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

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

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

from agent_framework import ChatMiddleware, ChatContext

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    # Execute the original agent logic
    await next(context)

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

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

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

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