البرنامج الوسيط للعامل

يوفر البرنامج الوسيط في إطار عمل العامل طريقة قوية لاعتراض وتعديل وتحسين تفاعلات العامل في مراحل مختلفة من التنفيذ. يمكنك استخدام البرامج الوسيطة لتنفيذ المخاوف الشاملة مثل التسجيل والتحقق من الأمان ومعالجة الأخطاء وتحويل النتائج دون تعديل الوكيل الأساسي أو منطق الدالة.

يمكن تخصيص إطار عمل العامل باستخدام ثلاثة أنواع مختلفة من البرامج الوسيطة:

1. برنامج وسيط تشغيل العامل: يسمح باعتراض جميع عمليات تشغيل العامل، بحيث يمكن فحص الإدخال والإخراج و/أو تعديلهما حسب الحاجة.
2. البرنامج الوسيط لاستدعاء الدالة: يسمح باعتراض جميع استدعاءات الدالة التي ينفذها العامل، بحيث يمكن فحص الإدخال والإخراج وتعديلهما حسب الحاجة.
3. IChatClient البرنامج الوسيط: يسمح باعتراض المكالمات إلى IChatClient تنفيذ، حيث يستخدم IChatClient العامل لاستدعاءات الاستدلال، على سبيل المثال، عند استخدام ChatClientAgent.

يتم تنفيذ جميع أنواع البرامج الوسيطة عبر رد اتصال دالة، وعند تسجيل مثيلات متعددة للبرامج الوسيطة من نفس النوع، فإنها تشكل سلسلة، حيث يتوقع من كل مثيل برنامج وسيط استدعاء التالي في السلسلة، عبر المقدمة nextFunc.

يمكن تسجيل أنواع البرامج الوسيطة لتشغيل العامل واستدعاء الدالة على عامل، باستخدام منشئ العامل مع كائن عامل موجود.

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

Important

من الناحية المثالية، يجب توفير كل من runFunc و runStreamingFunc . عند توفير البرامج الوسيطة غير المتدفقة فقط، سيستخدمها العامل لكل من استدعاءات الدفق وغير المتدفقة. سيتم تشغيل الدفق فقط في وضع عدم البث ليكفي توقعات البرامج الوسيطة.

Note

هناك تحميل زائد إضافي، Use(sharedFunc: ...)، يسمح لك بتوفير نفس البرامج الوسيطة لعدم الدفق والدفق دون حظر الدفق. ومع ذلك، لن يتمكن البرنامج الوسيط المشترك من اعتراض الإخراج أو تجاوزه. يجب استخدام هذا التحميل الزائد للسيناريوهات التي تحتاج فيها فقط إلى فحص الإدخال أو تعديله قبل أن يصل إلى العامل.

IChatClient يمكن تسجيل البرنامج الوسيط على IChatClient قبل استخدامه مع ChatClientAgent، باستخدام نمط منشئ عميل الدردشة.

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

دالة استدعاء البرامج الوسيطة

Note

يتم دعم برنامج وسيط استدعاء الدالة 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;
}

Tip

راجع نماذج .NET للحصول على أمثلة كاملة قابلة للتشغيل.

Note

لمزيد من المعلومات حول IChatClient البرامج الوسيطة، راجع البرامج الوسيطة المخصصة IChatClient.

يمكن تخصيص إطار عمل العامل باستخدام ثلاثة أنواع مختلفة من البرامج الوسيطة:

1. البرنامج الوسيط للعامل: يعترض تنفيذ تشغيل العامل، ما يسمح لك بفحص وتعديل المدخلات والمخرجات وتدفق التحكم.
2. البرنامج الوسيط للدالة: يعترض استدعاءات الدالة (الأداة) التي يتم إجراؤها أثناء تنفيذ العامل، مما يتيح التحقق من صحة الإدخال وتحويل النتائج والتحكم في التنفيذ.
3. برنامج الدردشة الوسيط: يعترض طلبات الدردشة الأساسية المرسلة إلى نماذج الذكاء الاصطناعي، ما يوفر الوصول إلى الرسائل الأولية والخيارات والاستجابات.

تدعم جميع الأنواع كلا من التطبيقات المستندة إلى الدالة والتطبيقات المستندة إلى الفئة. عند تسجيل برامج وسيطة متعددة من نفس النوع، فإنها تشكل سلسلة حيث يستدعي call_next كل منها رد الاتصال لمتابعة المعالجة. call_next لا يأخذ السياق كوسيطة؛ يقوم البرنامج الوسيط بتحول عنصر السياق المشترك مباشرة ثم ينتظر call_next().

Note

ترتيب البرامج الوسيطة مع نطاقات التسجيل المختلطة:

• يلتف البرنامج الوسيط على مستوى العامل البرامج الوسيطة على مستوى التشغيل.
• بالنسبة إلى البرنامج [A1, A2] الوسيط للعامل وتشغيل البرنامج [R1, R2]الوسيط، يكون ترتيب التنفيذ هو: A1 -> A2 -> R1 -> R2 -> Agent -> R2 -> R1 -> A2 -> A1.
• يتبع البرنامج الوسيط للدالة/الدردشة نفس مبدأ الالتفاف في وقت الأدوات/مكالمة الدردشة.

Tip

للحصول على عامل تحكم موحد ومغلق الفشل يمتد على عامل ودردشة ودالة وسيطة، راجع خطافات العامل. ينسق Agent Hooks أيضا التدفق الأساسي وسلوك الاستمرارية الذي لا يمكن أن توفره البرامج الوسيطة العادية في حد ذاتها.

البرنامج الوسيط للعامل

يعترض البرنامج الوسيط العامل تنفيذ تشغيل العامل ويعدله. ويستخدم الذي يحتوي على 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 :

• client: عميل الدردشة الذي يتم استدعاؤه
• messages: قائمة بالرسائل التي يتم إرسالها إلى خدمة الذكاء الاصطناعي
• options: خيارات طلب الدردشة
• stream: منطقي يشير إلى ما إذا كان هذا استدعاء دفق
• metadata: قاموس لتخزين بيانات إضافية بين البرامج الوسيطة
• result: استجابة الدردشة من الذكاء الاصطناعي (يمكن تعديلها)
• kwargs: وسيطات كلمات أساسية إضافية تم تمريرها إلى عميل الدردشة
• function_invocation_kwargs: قيم وقت تشغيل الأداة فقط التي سيتم إعادة توجيهها بواسطة طبقة الدردشة

call_next يستمر رد الاتصال إلى البرنامج الوسيط التالي أو يرسل الطلب إلى خدمة الذكاء الاصطناعي.

Note

يتم تشغيل البرنامج الوسيط للدردشة داخل حلقة استدعاء الدالة. وهذا يعني أنه ينفذ لكل استدعاء نموذج، بما في ذلك المكالمات التي ترسل نتائج الأداة مرة أخرى إلى النموذج أثناء تسلسل استدعاء أداة متعددة الأدوار.

مستند إلَى الدالة

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

نظرة عامة على البرامج الوسيطة

يعترض البرنامج الوسيط في Go سلوك العامل ويعدله على مستوى التشغيل. جميع البرامج الوسيطة تنفذ الواجهة agent.Middleware .

البرامج الوسيطة المضمنة والموفرة من إطار العمل

المكون	التسجيل	الطبقة	الغرض
مكالمة تلقائية	agent/harness/toolautocall	البرامج الوسيطة لموفر الخدمة	استدعاء أدوات الدالة تلقائيا
الإنتاج المنظم	agent.WithStructuredOutput	البرامج الوسيطة لموفر الخدمة	يعالج تحليل الإخراج المنظم
التتبع المفتوح	provider/otelprovider	البرنامج الوسيط للعامل	تتبع استدعاءات العامل
تشغيل المسجل	agent.Config.Logger	البرنامج الوسيط للعامل	تفاعلات عامل السجلات

موفرو السياق هم مكونات دورة الحياة المجاورة بدلا من agent.Middleware عمليات التنفيذ. يتم تشغيلها بعد أن يدخل البرنامج الوسيط للعامل المخصص في التشغيل وقبل أن يستدعي البرنامج الوسيط للموفر النموذج.

تسجيل البرامج الوسيطة

import otelprovider "github.com/microsoft/agent-framework-go/provider/otelprovider"

a := foundryprovider.NewAgent(endpoint, token, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
    Config: agent.Config{
        Middlewares: []agent.Middleware{
            otelprovider.NewMiddleware(otelprovider.MiddlewareConfig{}),
            myCustomMiddleware,
        },
    },
})

يتم تطبيق البرامج الوسيطة المسجلة في agent.Config.Middlewares بالترتيب المعلن؛ يقوم البرنامج الوسيط الأول بتضمين الطبقة المخصصة في أقصى الحدود. تقوم هذه الطبقة المخصصة بتضمين موفري المحفوظات وموفري السياق والبرامج الوسيطة لموفر الخدمة.

إنشاء برامج وسيطة

استخدم agent.MiddlewareFunc عندما يكون نوع البنية الكاملة غير ضروري:

addGuidance := agent.MiddlewareFunc(
    func(next agent.RunFunc, ctx context.Context, messages []*message.Message, options ...agent.Option) iter.Seq2[*agent.ResponseUpdate, error] {
        guided := append([]*message.Message{message.NewText("Keep the response concise and avoid exposing secrets.")}, messages...)
        return next(ctx, guided, options...)
    },
)

a := foundryprovider.NewAgent(endpoint, token, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
    Config: agent.Config{
        Middlewares: []agent.Middleware{addGuidance},
    },
})

الخطوات التالية

تعريف البرنامج الوسيط