// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Agents.AI.Purview.Models.Common;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.Logging;
namespace Microsoft.Agents.AI.Purview;
///
/// A delegating agent that connects to Microsoft Purview.
///
internal sealed class PurviewWrapper : IDisposable
{
private readonly ILogger _logger;
private readonly IScopedContentProcessor _scopedProcessor;
private readonly PurviewSettings _purviewSettings;
private readonly IBackgroundJobRunner _backgroundJobRunner;
///
/// Creates a new instance.
///
/// The scoped processor used to orchestrate the calls to Purview.
/// The settings for Purview integration.
/// The logger used for logging.
/// The runner used to manage background jobs.
public PurviewWrapper(IScopedContentProcessor scopedProcessor, PurviewSettings purviewSettings, ILogger logger, IBackgroundJobRunner backgroundJobRunner)
{
this._scopedProcessor = scopedProcessor;
this._purviewSettings = purviewSettings;
this._logger = logger;
this._backgroundJobRunner = backgroundJobRunner;
}
private static string GetSessionIdFromAgentSession(AgentSession? session, IEnumerable messages)
{
if (session is ChatClientAgentSession chatClientAgentSession &&
chatClientAgentSession.ConversationId != null)
{
return chatClientAgentSession.ConversationId;
}
foreach (ChatMessage message in messages)
{
if (message.AdditionalProperties != null &&
message.AdditionalProperties.TryGetValue(Constants.ConversationId, out object? conversationId) &&
conversationId != null)
{
return conversationId.ToString() ?? Guid.NewGuid().ToString();
}
}
return string.Empty;
}
///
/// Processes a prompt and response exchange at a chat client level.
///
/// The messages sent to the chat client.
/// The chat options used with the chat client.
/// The wrapped chat client.
/// The cancellation token used to interrupt async operations.
/// The chat client's response. This could be the response from the chat client or a message indicating that Purview has blocked the prompt or response.
public async Task ProcessChatContentAsync(IEnumerable messages, ChatOptions? options, IChatClient innerChatClient, CancellationToken cancellationToken)
{
string? resolvedUserId = null;
try
{
(bool shouldBlockPrompt, resolvedUserId) = await this._scopedProcessor.ProcessMessagesAsync(messages, options?.ConversationId, Activity.UploadText, this._purviewSettings, null, cancellationToken).ConfigureAwait(false);
if (shouldBlockPrompt)
{
if (this._logger.IsEnabled(LogLevel.Information))
{
this._logger.LogInformation("Prompt blocked by policy. Sending message: {Message}", this._purviewSettings.BlockedPromptMessage);
}
return new ChatResponse(new ChatMessage(ChatRole.System, this._purviewSettings.BlockedPromptMessage));
}
}
catch (Exception ex)
{
if (this._logger.IsEnabled(LogLevel.Error))
{
this._logger.LogError(ex, "Error processing prompt: {ExceptionMessage}", ex.Message);
}
if (!this._purviewSettings.IgnoreExceptions)
{
throw;
}
}
ChatResponse response = await innerChatClient.GetResponseAsync(messages, options, cancellationToken).ConfigureAwait(false);
try
{
(bool shouldBlockResponse, _) = await this._scopedProcessor.ProcessMessagesAsync(response.Messages, options?.ConversationId, Activity.DownloadText, this._purviewSettings, resolvedUserId, cancellationToken).ConfigureAwait(false);
if (shouldBlockResponse)
{
if (this._logger.IsEnabled(LogLevel.Information))
{
this._logger.LogInformation("Response blocked by policy. Sending message: {Message}", this._purviewSettings.BlockedResponseMessage);
}
return new ChatResponse(new ChatMessage(ChatRole.System, this._purviewSettings.BlockedResponseMessage));
}
}
catch (Exception ex)
{
if (this._logger.IsEnabled(LogLevel.Error))
{
this._logger.LogError(ex, "Error processing response: {ExceptionMessage}", ex.Message);
}
if (!this._purviewSettings.IgnoreExceptions)
{
throw;
}
}
return response;
}
///
/// Processes a prompt and response exchange at an agent level.
///
/// The messages sent to the agent.
/// The session used for this agent conversation.
/// The options used with this agent.
/// The wrapped agent.
/// The cancellation token used to interrupt async operations.
/// The agent's response. This could be the response from the agent or a message indicating that Purview has blocked the prompt or response.
public async Task ProcessAgentContentAsync(IEnumerable messages, AgentSession? session, AgentRunOptions? options, AIAgent innerAgent, CancellationToken cancellationToken)
{
string? resolvedUserId = null;
string sessionId = string.Empty;
try
{
sessionId = GetSessionIdFromAgentSession(session, messages);
if (string.IsNullOrEmpty(sessionId))
{
sessionId = Guid.NewGuid().ToString();
}
(bool shouldBlockPrompt, resolvedUserId) = await this._scopedProcessor.ProcessMessagesAsync(messages, sessionId, Activity.UploadText, this._purviewSettings, null, cancellationToken).ConfigureAwait(false);
if (shouldBlockPrompt)
{
if (this._logger.IsEnabled(LogLevel.Information))
{
this._logger.LogInformation("Prompt blocked by policy. Sending message: {Message}", this._purviewSettings.BlockedPromptMessage);
}
return new AgentResponse(new ChatMessage(ChatRole.System, this._purviewSettings.BlockedPromptMessage));
}
}
catch (Exception ex)
{
if (this._logger.IsEnabled(LogLevel.Error))
{
this._logger.LogError(ex, "Error processing prompt: {ExceptionMessage}", ex.Message);
}
if (!this._purviewSettings.IgnoreExceptions)
{
throw;
}
}
AgentResponse response = await innerAgent.RunAsync(messages, session, options, cancellationToken).ConfigureAwait(false);
try
{
string sessionIdResponse = GetSessionIdFromAgentSession(session, messages);
if (string.IsNullOrEmpty(sessionIdResponse))
{
if (string.IsNullOrEmpty(sessionId))
{
sessionIdResponse = Guid.NewGuid().ToString();
}
else
{
sessionIdResponse = sessionId;
}
}
(bool shouldBlockResponse, _) = await this._scopedProcessor.ProcessMessagesAsync(response.Messages, sessionIdResponse, Activity.DownloadText, this._purviewSettings, resolvedUserId, cancellationToken).ConfigureAwait(false);
if (shouldBlockResponse)
{
if (this._logger.IsEnabled(LogLevel.Information))
{
this._logger.LogInformation("Response blocked by policy. Sending message: {Message}", this._purviewSettings.BlockedResponseMessage);
}
return new AgentResponse(new ChatMessage(ChatRole.System, this._purviewSettings.BlockedResponseMessage));
}
}
catch (Exception ex)
{
if (this._logger.IsEnabled(LogLevel.Error))
{
this._logger.LogError(ex, "Error processing response: {ExceptionMessage}", ex.Message);
}
if (!this._purviewSettings.IgnoreExceptions)
{
throw;
}
}
return response;
}
///
public void Dispose()
{
#pragma warning disable VSTHRD002 // Need to wait for pending jobs to complete.
this._backgroundJobRunner.ShutdownAsync().GetAwaiter().GetResult();
#pragma warning restore VSTHRD002 // Need to wait for pending jobs to complete.
}
}