Notatka
Dostęp do tej strony wymaga autoryzacji. Może spróbować zalogować się lub zmienić katalogi.
Dostęp do tej strony wymaga autoryzacji. Możesz spróbować zmienić katalogi.
Dowiedz się, jak dodać oprogramowanie pośredniczące do agentów w kilku prostych krokach. Oprogramowanie pośredniczące pozwala na przechwytywanie i modyfikowanie interakcji agenta w celu rejestrowania, ochrony i innych zagadnień przekrojowych.
Wymagania wstępne
Aby uzyskać wymagania wstępne i zainstalować pakiety NuGet, zobacz krok Tworzenie i uruchamianie prostego agenta w tym samouczku.
Krok 1. Tworzenie prostego agenta
Najpierw utwórz podstawowego agenta za pomocą narzędzia funkcji.
using System;
using System.ComponentModel;
using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;
[Description("The current datetime offset.")]
static string GetDateTime()
=> DateTimeOffset.Now.ToString();
AIAgent baseAgent = new AIProjectClient(
new Uri("<your-foundry-project-endpoint>"),
new DefaultAzureCredential())
.AsAIAgent(
model: "gpt-4o-mini",
instructions: "You are an AI assistant that helps people find information.",
tools: [AIFunctionFactory.Create(GetDateTime, name: nameof(GetDateTime))]);
Ostrzeżenie
DefaultAzureCredential jest wygodne do programowania, ale wymaga starannego rozważenia w środowisku produkcyjnym. W środowisku produkcyjnym rozważ użycie określonego poświadczenia (np. ManagedIdentityCredential), aby uniknąć problemów z opóźnieniami, niezamierzonego sondowania poświadczeń i potencjalnych zagrożeń bezpieczeństwa wynikających z mechanizmów awaryjnych.
Krok 2. Tworzenie oprogramowania pośredniczącego w uruchamianiu agenta
Następnie utwórz funkcję, która zostanie wywołana dla każdego uruchomienia agenta. Umożliwia inspekcję danych wejściowych i wyjściowych agenta.
Jeśli intencją nie jest użycie middleware do zatrzymania wykonywania procesu, funkcja powinna wywołać RunAsync na podanym obiekcie innerAgent.
To przykładowe oprogramowanie pośredniczące wyłącznie kontroluje dane wejściowe i wyjściowe z pracy agenta oraz wyświetla liczbę komunikatów przekazywanych do agenta oraz od agenta.
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
async Task<AgentResponse> CustomAgentRunMiddleware(
IEnumerable<ChatMessage> messages,
AgentSession? session,
AgentRunOptions? options,
AIAgent innerAgent,
CancellationToken cancellationToken)
{
Console.WriteLine($"Input: {messages.Count()}");
var response = await innerAgent.RunAsync(messages, session, options, cancellationToken).ConfigureAwait(false);
Console.WriteLine($"Output: {response.Messages.Count}");
return response;
}
Krok 3. Dodaj middleware uruchamiania agenta do swojego agenta
Aby dodać tę funkcję middleware do utworzonego baseAgent w kroku 1, użyj wzorca konstruktora.
Spowoduje to utworzenie nowego agenta, który ma zastosowane oprogramowanie pośredniczące.
Oryginał baseAgent nie jest modyfikowany.
var middlewareEnabledAgent = baseAgent
.AsBuilder()
.Use(runFunc: CustomAgentRunMiddleware, runStreamingFunc: null)
.Build();
Teraz, podczas uruchamiania agenta z zapytaniem, powinno zostać uruchomione oprogramowanie pośredniczące, podając liczbę wiadomości wejściowych i liczbę wiadomości odpowiedzi.
Console.WriteLine(await middlewareEnabledAgent.RunAsync("What's the current time?"));
Krok 4: Tworzenie funkcji do wywoływania middleware
Uwaga / Notatka
Średnie oprogramowanie do wywoływania funkcji jest obecnie obsługiwane tylko w przypadku komponentu AIAgent, który używa FunctionInvokingChatClient, na przykład ChatClientAgent.
Możesz również utworzyć oprogramowanie pośredniczące, które jest wywoływane dla każdego wywoływanego narzędzia funkcji. Oto przykład oprogramowania pośredniczącego wywołującego funkcję, który może sprawdzać i/lub modyfikować wywoływaną funkcję oraz wynik wywołania funkcji.
O ile celem nie jest użycie middleware do niewykonania funkcji narzędzia, middleware powinno wywołać podany nextFunc.
using System.Threading;
using System.Threading.Tasks;
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;
}
Krok 5. Dodawanie wywoływania funkcji middleware do agenta
Podobnie jak w przypadku dodawania oprogramowania pośredniczącego uruchamiającego agenta, można dodać funkcję wywołującą oprogramowanie pośredniczące w następujący sposób:
var middlewareEnabledAgent = baseAgent
.AsBuilder()
.Use(CustomFunctionCallingMiddleware)
.Build();
Teraz podczas wykonywania agenta za pomocą zapytania, które wywołuje funkcję, oprogramowanie pośredniczące powinno zostać wywołane, wyprowadzając nazwę funkcji i wynik wywołania.
Console.WriteLine(await middlewareEnabledAgent.RunAsync("What's the current time?"));
Krok 6: Tworzenie middleware klienta czatu
W przypadku agentów utworzonych przy użyciu programu IChatClient można chcieć przechwycić wywołania przechodzące od agenta do IChatClient.
W takim przypadku można użyć middleware dla IChatClient.
Oto przykład oprogramowania pośredniczącego klienta czatu, które może sprawdzać i/lub modyfikować dane wejściowe i wyjściowe dla żądania do usługi inferencyjnej, którą zapewnia klient czatu.
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
async Task<ChatResponse> CustomChatClientMiddleware(
IEnumerable<ChatMessage> messages,
ChatOptions? options,
IChatClient innerChatClient,
CancellationToken cancellationToken)
{
Console.WriteLine($"Input: {messages.Count()}");
var response = await innerChatClient.GetResponseAsync(messages, options, cancellationToken);
Console.WriteLine($"Output: {response.Messages.Count}");
return response;
}
Uwaga / Notatka
Aby uzyskać więcej informacji na temat IChatClient oprogramowania pośredniczącego, zobacz Niestandardowe oprogramowanie pośredniczące IChatClient.
Krok 7. Dodawanie oprogramowania pośredniczącego klienta czatu do elementu IChatClient
Aby dodać oprogramowanie pośredniczące do programu IChatClient, możesz użyć wzorca konstruktora.
Po dodaniu oprogramowania pośredniczącego, możesz użyć IChatClient ze swoim agentem jak zwykle.
var chatClient = new AIProjectClient(
new Uri("<your-foundry-project-endpoint>"),
new DefaultAzureCredential())
.GetProjectOpenAIClient()
.GetProjectResponsesClient()
.AsIChatClient("gpt-4o-mini");
var middlewareEnabledChatClient = chatClient
.AsBuilder()
.Use(getResponseFunc: CustomChatClientMiddleware, getStreamingResponseFunc: null)
.Build();
var agent = new ChatClientAgent(middlewareEnabledChatClient, instructions: "You are a helpful assistant.");
IChatClient Oprogramowanie pośredniczące można również zarejestrować za pomocą metody fabrycznej podczas tworzenia agenta przy użyciu jednej z metod pomocniczych na klientach SDK.
var agent = new AIProjectClient(
new Uri("<your-foundry-project-endpoint>"),
new DefaultAzureCredential())
.AsAIAgent(
model: "gpt-4o-mini",
instructions: "You are a helpful assistant.",
clientFactory: (chatClient) => chatClient
.AsBuilder()
.Use(getResponseFunc: CustomChatClientMiddleware, getStreamingResponseFunc: null)
.Build());
Krok 1. Tworzenie prostego agenta
Najpierw utwórz podstawowego agenta:
import asyncio
from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from azure.identity.aio import AzureCliCredential
async def main():
credential = AzureCliCredential()
async with Agent(
client=FoundryChatClient(credential=credential),
name="GreetingAgent",
instructions="You are a friendly greeting assistant.",
) as agent:
result = await agent.run("Hello!")
print(result.text)
if __name__ == "__main__":
asyncio.run(main())
Krok 2. Tworzenie oprogramowania pośredniczącego
Utwórz proste middleware logujące, aby zobaczyć, kiedy agent jest uruchamiany.
from collections.abc import Awaitable, Callable
from agent_framework import AgentContext
async def logging_agent_middleware(
context: AgentContext,
call_next: Callable[[], Awaitable[None]],
) -> None:
"""Simple middleware that logs agent execution."""
print("Agent starting...")
# Continue to agent execution
await call_next()
print("Agent finished!")
Krok 3: Dodaj oprogramowanie pośredniczące do swojego agenta
Dodaj oprogramowanie pośredniczące podczas tworzenia agenta:
async def main():
credential = AzureCliCredential()
async with Agent(
client=FoundryChatClient(credential=credential),
name="GreetingAgent",
instructions="You are a friendly greeting assistant.",
middleware=[logging_agent_middleware], # Add your middleware here
) as agent:
result = await agent.run("Hello!")
print(result.text)
Krok 4: Utwórz funkcję pośredniczącą
Jeśli agent używa funkcji, możesz przechwytywać wywołania funkcji i ustawiać wartości środowiska uruchomieniowego tylko dla narzędzi przed wykonaniem narzędzia:
from collections.abc import Awaitable, Callable
from agent_framework import FunctionInvocationContext
def get_time(ctx: FunctionInvocationContext) -> str:
"""Get the current time."""
from datetime import datetime
source = ctx.kwargs.get("request_source", "direct")
return f"[{source}] {datetime.now().strftime('%H:%M:%S')}"
async def inject_function_kwargs(
context: FunctionInvocationContext,
call_next: Callable[[], Awaitable[None]],
) -> None:
"""Middleware that adds tool-only runtime values before execution."""
context.kwargs.setdefault("request_source", "middleware")
await call_next()
# Add both the function and middleware to your agent
async with Agent(
client=FoundryChatClient(credential=credential),
name="TimeAgent",
instructions="You can tell the current time.",
tools=[get_time],
middleware=[inject_function_kwargs],
) as agent:
result = await agent.run("What time is it?")
Krok 5. Używanie oprogramowania pośredniczącego Run-Level
Możesz również dodać oprogramowanie pośredniczące dla określonych uruchomień.
# Use middleware for this specific run only
result = await agent.run(
"This is important!",
middleware=[logging_function_middleware]
)
Co dalej?
Aby uzyskać bardziej zaawansowane scenariusze, zobacz Podręcznik użytkownika oprogramowania pośredniczącego Agent, który obejmuje:
- Różne typy oprogramowania pośredniczącego (agent, funkcja, czat).
- Oprogramowanie pośredniczące oparte na klasach dla złożonych scenariuszy.
- Zakończenie działania oprogramowania pośredniczącego i przedefiniowanie wyników.
- Zaawansowane wzorce oprogramowania pośredniczącego i najlepsze rozwiązania.
Kompletne przykłady
oprogramowanie pośredniczące oparte na klasach
# 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())
Oprogramowanie pośredniczące oparte na funkcjach
# 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())
Oprogramowanie pośredniczące oparte na dekoratorze
# 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())