Бөлісу құралы:

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

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

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

  1. ПО промежуточного слоя запуска агента: позволяет перехватывать все запуски агента, чтобы входные и выходные данные могли быть проверены и (или) изменены по мере необходимости.
  2. ПО промежуточного слоя вызовов функций: позволяет перехватывать все вызовы функций, выполняемые агентом, чтобы входные и выходные данные могли быть проверены и изменены по мере необходимости.
  3. 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 AIProjectClient(
    new Uri("<your-foundry-project-endpoint>"),
    new DefaultAzureCredential())
        .GetProjectOpenAIClient()
        .GetProjectResponsesClient()
        .AsIChatClient(deploymentName);

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 AIProjectClient(
    new Uri("<your-foundry-project-endpoint>"),
    new DefaultAzureCredential())
        .AsAIAgent(
            model: deploymentName,
            instructions: "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.

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

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

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

Замечание

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

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

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

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

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

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

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

async def inject_tool_runtime_defaults(
    context: AgentContext,
    call_next: Callable[[], Awaitable[None]],
) -> None:
    """Agent middleware that sets tool-only runtime defaults."""
    print("[Agent] Starting execution")
    context.function_invocation_kwargs.setdefault("tenant", "contoso")
    context.function_invocation_kwargs.setdefault("request_source", "agent-middleware")

    await call_next()

    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,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        print("[Agent Class] Starting execution")
        await call_next()
        print("[Agent Class] Execution completed")

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

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

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

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

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

async def inject_function_kwargs(
    context: FunctionInvocationContext,
    call_next: Callable[[], Awaitable[None]],
) -> None:
    """Function middleware that enriches tool runtime values."""
    context.kwargs.setdefault("tenant", "contoso")
    context.kwargs.setdefault("request_source", "function-middleware")

    await call_next()

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

from agent_framework import FunctionMiddleware, FunctionInvocationContext

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

    async def process(
        self,
        context: FunctionInvocationContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        print(f"[Function Class] Calling {context.function.name}")
        await call_next()
        print(f"[Function Class] {context.function.name} completed")

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

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

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

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

Замечание

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

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

async def logging_chat_middleware(
    context: ChatContext,
    call_next: Callable[[], 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 call_next()

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

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

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

from agent_framework import agent_middleware, function_middleware, chat_middleware

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

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

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

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

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

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

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

# Agent-level middleware: Applied to ALL runs of the agent
async with Agent(
    client=FoundryChatClient(credential=credential),
    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.result и вызвав MiddlewareTermination. Это полезно для проверок безопасности, ограничения скорости или сбоев проверки.

from agent_framework import AgentContext, AgentResponse, Message, MiddlewareTermination

async def blocking_middleware(
    context: AgentContext,
    call_next: Callable[[], 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.result = AgentResponse(
                messages=[Message(role="assistant", contents=["This request was blocked by middleware."])]
            )
            raise MiddlewareTermination(result=context.result)

    # If no issues, continue normally
    await call_next()

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

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

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

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

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

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

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

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

    # Execute the original agent logic
    await call_next()

    # 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.stream:
            # 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 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())

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

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

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

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

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

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

  • Last updated on