Aracılığıyla paylaş

Aracı Yazılım

Agent Framework'teki ara yazılım, yürütmenin çeşitli aşamalarında aracı etkileşimlerini kesme, değiştirme ve geliştirme için güçlü bir yol sağlar. Ara yazılımı kullanarak temel aracınızı veya işlev mantığınızı değiştirmeden günlüğe kaydetme, güvenlik doğrulaması, hata işleme ve sonuç dönüştürme gibi çapraz kesme sorunlarını uygulayabilirsiniz.

Aracı Çerçevesi üç farklı ara yazılım türü kullanılarak özelleştirilebilir:

  1. Aracı Çalıştırma ara yazılımı: Giriş ve çıkışın gerektiğinde denetlenebilmesi ve/veya değiştirilebilmesi için tüm aracı çalıştırmalarının kesilmesine izin verir.
  2. İşlev çağrı ara yazılımı: Giriş ve çıkışın gerektiği gibi incelenip değiştirilebilmesi için aracı tarafından yürütülen tüm işlev çağrılarının kesilmesine izin verir.
  3. IChatClientara yazılım: Bir aracının çıkarım çağrıları için kullandığı IChatClient bir IChatClient uygulamaya çağrıların kesilmesine izin verir, örneğin kullanırkenChatClientAgent.

Tüm ara yazılım türleri bir işlev geri çağırması aracılığıyla uygulanır ve aynı türdeki birden çok ara yazılım örneği kaydedildiğinde, sağlanan aracılığıyla her ara yazılım örneğinin zincirde bir sonrakini çağırmasının beklendiği bir nextFunczincir oluştururlar.

Aracı çalıştırması ve ara yazılım türlerini çağıran işlev, aracı oluşturucusu mevcut bir aracı nesnesiyle kullanılarak bir aracıya kaydedilebilir.

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

Önemli

İdeal olarak hem hem de runFuncrunStreamingFunc sağlanmalıdır. Aracı, yalnızca akış dışı ara yazılımı sağlarken hem akış hem de akışsız çağrılar için bu ara yazılımı kullanır. Akış yalnızca ara yazılım beklentilerini yeterli hale getirmek için akış dışı modda çalışır.

Uyarı

Akışı engellemeden akış dışı ve akış için aynı ara yazılımı sağlamanıza olanak tanıyan ek bir aşırı yükleme Use(sharedFunc: ...)vardır. Ancak, paylaşılan ara yazılım çıkışı kesemez veya geçersiz kılamaz. Bu aşırı yükleme, yalnızca aracıya ulaşmadan önce girişi incelemeniz veya değiştirmeniz gereken senaryolar için kullanılmalıdır.

IChatClientara yazılım, ile kullanılmadan önce sohbet istemcisi oluşturucu deseni kullanılarak üzerine IChatClientkaydedilebilirChatClientAgent.

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

Uyarı

DefaultAzureCredential geliştirme için uygundur ancak üretimde dikkatli bir şekilde dikkate alınması gerekir. Üretimde gecikme sorunları, istenmeyen kimlik bilgisi yoklama ve geri dönüş mekanizmalarından kaynaklanan olası güvenlik risklerini önlemek için belirli bir kimlik bilgisi (ör ManagedIdentityCredential. ) kullanmayı göz önünde bulundurun.

IChatClient Ara yazılım, SDK istemcilerindeki yardımcı yöntemlerden biri aracılığıyla aracı oluştururken fabrika yöntemi kullanılarak da kaydedilebilir.

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

Aracı Çalıştırma Ara Yazılımı

Aracı çalıştırmasının giriş ve çıkışını inceleyip/veya değiştirebilen aracı çalıştırma ara yazılımının bir örneği aşağıda verilmiştir.

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

Aracı Çalıştırma Akışı Ara Yazılımı

Aracı akış çalıştırmasının giriş ve çıkışını inceleyip/veya değiştirebilen aracı çalıştırması ara yazılımının bir örneği aşağıda verilmiştir.

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

ara yazılımı çağıran işlev

Uyarı

Şu anda, fonksiyon çağrı ara yazılımı sadece AIAgent kullanan bir FunctionInvokingChatClient ile desteklenmektedir, örneğin, ChatClientAgent.

Çağrılan işlevi ve işlev çağrısının sonucunu inceleyip/veya değiştirebilen ara yazılımı çağıran işlev örneği aşağıda verilmiştir.

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

Sağlanan FunctionInvocationContext.Terminate değerini true olarak ayarlayarak işlev çağrısı döngüsünü işlev çağrısı ara yazılımıyla sonlandırmak mümkündür. Bu, işlev çağrısı döngüsünün işlev çağrısından sonra işlev çağrısı sonuçlarını içeren çıkarım hizmetine bir istek vermesini engeller. Bu yineleme sırasında çağrı için kullanılabilecek birden fazla işlev varsa, kalan işlevlerin yürütülmesini de engelleyebilir.

Uyarı

İşlev çağrısı döngüsünü sonlandırmak, sohbet geçmişinizin tutarsız bir durumda bırakılmasına neden olabilir; örneğin işlev sonucu içeriği olmayan işlev çağrısı içeriği içerebilir. Bu, sohbet geçmişinin daha fazla çalıştırma için kullanılamaz olmasına neden olabilir.

IChatClient ara yazılımı

Burada, sohbet istemcisinin sağladığı çıkarım hizmetine yönelik isteğin giriş ve çıkışını denetleyebilen ve/veya değiştirebilen bir sohbet istemcisi ara yazılımı örneği verilmiştir.

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

Tavsiye

Çalıştırılabilir örneklerin tamamı için .NET örneklerine bakın.

Uyarı

Ara yazılım hakkında IChatClient daha fazla bilgi için bkz. Özel IChatClient ara yazılımı.

Aracı Çerçevesi üç farklı ara yazılım türü kullanılarak özelleştirilebilir:

  1. Aracı ara yazılımı: Aracı çalıştırma yürütmesini durdurarak girişleri, çıkışları ve denetim akışını incelemenize ve değiştirmenize olanak tanır.
  2. İşlev ara yazılımı: Aracı yürütme sırasında yapılan işlev (araç) çağrılarını keserek giriş doğrulama, sonuç dönüştürme ve yürütme denetimini etkinleştirir.
  3. Sohbet ara yazılımı: Yapay zeka modellerine gönderilen temel sohbet isteklerini keserek ham iletilere, seçeneklere ve yanıtlara erişim sağlar.

Tüm türler hem işlev tabanlı hem de sınıf tabanlı uygulamaları destekler. Aynı türdeki birden çok ara yazılım kaydedildiğinde, her birinin işlemeye devam etmek için çağrılabilir öğesini çağırdığı next bir zincir oluşturur.

Aracı Yazılım

Aracı ara yazılım, aracı çalıştırma yürütmeyi durdurur ve değiştirir. AgentContext Aşağıdakini içerir:

  • agent: Çağrılan aracı
  • messages: Konuşmadaki sohbet iletilerinin listesi
  • is_streaming: Yanıtın akışla aktarılıp aktarılmadığını gösteren Boole değeri
  • metadata: Ara yazılım arasında ek veri depolama sözlüğü
  • result: Aracının yanıtı (değiştirilebilir)
  • terminate: Daha fazla işlemeyi durdurmak için bayrak ekleme
  • kwargs: Aracı çalıştırma yöntemine geçirilen ek anahtar sözcük bağımsız değişkenleri

Çağrılabilir next , ara yazılım zincirine devam eder veya aracı son ara yazılımsa yürütür.

İşlev tabanlı

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

Sınıf tabanlı

Sınıf tabanlı aracı ara yazılımı, işlev tabanlı ara yazılımla aynı imzaya ve davranışa sahip bir process yöntem kullanır.

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

İşlev Ara Yazılımı

İşlev ara yazılımı, aracılar içindeki işlev çağrılarını durdurur. FunctionInvocationContext Aşağıdakini içerir:

  • function: Çağrılan işlev
  • arguments: İşlev için doğrulanmış bağımsız değişkenler
  • metadata: Ara yazılım arasında ek veri depolama sözlüğü
  • result: İşlevin dönüş değeri (değiştirilebilir)
  • terminate: Daha fazla işlemeyi durdurmak için bayrak ekleme
  • kwargs: Bu işlevi çağıran sohbet yöntemine geçirilen ek anahtar sözcük bağımsız değişkenleri

Çağrılabilir next , sonraki ara yazılıma devam eder veya gerçek işlevi yürütür.

İşlev tabanlı

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

Sınıf tabanlı

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

Sohbet Ara Yazılımı

Sohbet ara yazılımı, yapay zeka modellerine gönderilen sohbet isteklerini durdurur. ChatContext Aşağıdakini içerir:

  • chat_client: Çağrılan sohbet istemcisi
  • messages: Yapay zeka hizmetine gönderilen iletilerin listesi
  • options: Sohbet isteği seçenekleri
  • is_streaming: Bunun bir akış çağrısı olup olmadığını gösteren Boole değeri
  • metadata: Ara yazılım arasında ek veri depolama sözlüğü
  • result: Yapay zekadan gelen sohbet yanıtı (değiştirilebilir)
  • terminate: Daha fazla işlemeyi durdurmak için bayrak ekleme
  • kwargs: Sohbet istemcisine geçirilen ek anahtar sözcük bağımsız değişkenleri

Çağrılabilen next bir sonraki ara yazılıma devam eder veya isteği yapay zeka hizmetine gönderir.

İşlev tabanlı

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

Sınıf tabanlı

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

Ara Yazılım Dekoratörleri

Dekoratörler, tür ek açıklamalarına gerek kalmadan açık ara yazılım türü bildirimi sağlar. Bunlar, tür ek açıklamalarını kullanmadığınızda veya tür uyuşmazlıklarını önlemek istediğinizde yararlıdır:

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

Ara Yazılım Kaydı

Ara yazılım, farklı kapsamlara ve davranışlara sahip iki düzeyde kaydedilebilir.

Agent-Level ve Run-Level Ara Yazılımı karşılaştırması

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

Önemli Farklar:

  • Aracı düzeyi: Aracı oluşturulurken bir kez yapılandırılan tüm çalıştırmalarda kalıcı
  • Çalıştırma düzeyi: Yalnızca belirli çalıştırmalara uygulanır, istek başına özelleştirmeye izin verir
  • Yürütme Sırası: Aracı ara yazılımı (en dışta) → Ara yazılımı (en içteki) → Aracı yürütmeyi çalıştırma

Ara Yazılım Sonlandırma

Ara yazılım kullanarak context.terminateyürütmeyi erken sonlandırabilir. Bu, güvenlik denetimleri, hız sınırlama veya doğrulama hataları için kullanışlıdır.

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)

Sonlandırmanın anlamı:

  • İşlemenin durması gerektiğini belirten sinyalleri ayarlama context.terminate = True
  • Kullanıcılara geri bildirimde bulunmak için sonlandırmadan önce özel bir sonuç sağlayabilirsiniz
  • Ara yazılım sonlandırıldığında aracı yürütmesi tamamen atlanır

Ara Yazılım Sonucu Geçersiz Kılma

Ara yazılım hem akış dışı hem de akışsız senaryolarda sonuçları geçersiz kılabilir ve aracı yanıtlarını değiştirmenize veya tamamen değiştirmenize olanak sağlar.

içindeki context.result sonuç türü, aracı çağrısının akışlı mı yoksa akışsız mı olduğuna bağlıdır:

  • Akış dışı: context.result tam yanıt içeren bir AgentResponse içerir
  • Akış: context.result Öbekleri AgentResponseUpdate veren zaman uyumsuz bir oluşturucu içerir

Bu senaryoları ayırt etmek ve sonuç geçersiz kılmalarını uygun şekilde işlemek için kullanabilirsiniz 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])]
            )

Bu ara yazılım yaklaşımı, aracı mantığınızı temiz ve odaklanmış durumda tutarken gelişmiş yanıt dönüştürme, içerik filtreleme, sonuç geliştirme ve akış özelleştirmesi gerçekleştirmenizi sağlar.

Tam ara yazılım örnekleri

Sınıf tabanlı ara yazılım

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

İşlev tabanlı ara yazılım

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

Dekoratör tabanlı ara yazılım

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

Sonraki Adımlar

Aracı Arka Plan Yanıtları

  • Last updated on