agent/agent-framework
2
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

.NET: Add workflow as an agent with observability sample (#1612)

Browse Source 
* Add workflow as an agent with observability sample

* Address comment

* Fix formatting

* enable sensitive data

* enable sensitive data for sub agents

* adjust aggregator handlers
This commit is contained in:
Tao Chen
2025-11-03 12:36:33 -08:00
committed by GitHub
Unverified
parent 12d17acdc0
commit b0ee7028a6
8 changed files with 297 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dotnet/agent-framework-dotnet.slnx
+1
View File
@@ -123,6 +123,7 @@
<Folder Name="/Samples/GettingStarted/Workflows/Observability/">
<Project Path="samples/GettingStarted/Workflows/Observability/ApplicationInsights/ApplicationInsights.csproj" />
<Project Path="samples/GettingStarted/Workflows/Observability/AspireDashboard/AspireDashboard.csproj" />
<Project Path="samples/GettingStarted/Workflows/Observability/WorkflowAsAnAgent/WorkflowAsAnAgentObservability.csproj" />
</Folder>
<Folder Name="/Samples/GettingStarted/Workflows/Visualization/">
<Project Path="samples/GettingStarted/Workflows/Visualization/Visualization.csproj" Id="99bf0bc6-2440-428e-b3e7-d880e4b7a5fd" />
dotnet/samples/GettingStarted/AgentOpenTelemetry/Program.cs
+4 -2
View File
@@ -125,7 +125,7 @@ var agent = new ChatClientAgent(instrumentedChatClient,
instructions: "You are a helpful assistant that provides concise and informative responses.",
tools: [AIFunctionFactory.Create(GetWeatherAsync)])
.AsBuilder()
.UseOpenTelemetry(SourceName) // enable telemetry at the agent level
.UseOpenTelemetry(SourceName, configure: (cfg) => cfg.EnableSensitiveData = true) // enable telemetry at the agent level
.Build();
var thread = agent.GetNewThread();
@@ -134,6 +134,8 @@ appLogger.LogInformation("Agent created successfully with ID: {AgentId}", agent.
// Create a parent span for the entire agent session
using var sessionActivity = activitySource.StartActivity("Agent Session");
Console.WriteLine($"Trace ID: {sessionActivity?.TraceId} ");
var sessionId = Guid.NewGuid().ToString("N");
sessionActivity?
.SetTag("agent.name", "OpenTelemetryDemoAgent")
@@ -147,7 +149,7 @@ using (appLogger.BeginScope(new Dictionary<string, object> { ["SessionId"] = ses
while (true)
{
Console.Write("You: ");
Console.Write("You (or 'exit' to quit): ");
var userInput = Console.ReadLine();
if (string.IsNullOrWhiteSpace(userInput) || userInput.Equals("exit", StringComparison.OrdinalIgnoreCase))
dotnet/samples/GettingStarted/Workflows/Agents/WorkflowAsAnAgent/Program.cs
+3 -3
View File
@@ -6,7 +6,7 @@ using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;
namespace WorkflowAsAnAgentsSample;
namespace WorkflowAsAnAgentSample;
/// <summary>
/// This sample introduces the concepts workflows as agents, where a workflow can be
@@ -61,9 +61,9 @@ public static class Program
Dictionary<string, List<AgentRunResponseUpdate>> buffer = [];
await foreach (AgentRunResponseUpdate update in agent.RunStreamingAsync(input, thread))
{
if (update.MessageId is null)
if (update.MessageId is null || string.IsNullOrEmpty(update.Text))
{
// skip updates that don't have a message ID
// skip updates that don't have a message ID or text
continue;
}
Console.Clear();
dotnet/samples/GettingStarted/Workflows/Agents/WorkflowAsAnAgent/WorkflowFactory.cs
+20 -21
View File
@@ -4,7 +4,7 @@ using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;
namespace WorkflowAsAnAgentsSample;
namespace WorkflowAsAnAgentSample;
internal static class WorkflowFactory
{
@@ -41,44 +41,43 @@ internal static class WorkflowFactory
/// <summary>
/// Executor that starts the concurrent processing by sending messages to the agents.
/// </summary>
private sealed class ConcurrentStartExecutor() :
Executor<List<ChatMessage>>("ConcurrentStartExecutor")
private sealed class ConcurrentStartExecutor() : Executor("ConcurrentStartExecutor")
{
/// <summary>
/// Starts the concurrent processing by sending messages to the agents.
/// </summary>
/// <param name="message">The user message to process</param>
/// <param name="context">Workflow context for accessing workflow services and adding events</param>
/// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests.
/// The default is <see cref="CancellationToken.None"/>.</param>
public override async ValueTask HandleAsync(List<ChatMessage> message, IWorkflowContext context, CancellationToken cancellationToken = default)
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder)
{
// Broadcast the message to all connected agents. Receiving agents will queue
// the message but will not start processing until they receive a turn token.
await context.SendMessageAsync(message, cancellationToken: cancellationToken);
// Broadcast the turn token to kick off the agents.
await context.SendMessageAsync(new TurnToken(emitEvents: true), cancellationToken: cancellationToken);
return routeBuilder
.AddHandler<List<ChatMessage>>(this.RouteMessages)
.AddHandler<TurnToken>(this.RouteTurnTokenAsync);
}
private ValueTask RouteMessages(List<ChatMessage> messages, IWorkflowContext context, CancellationToken cancellationToken)
{
return context.SendMessageAsync(messages, cancellationToken: cancellationToken);
}
private ValueTask RouteTurnTokenAsync(TurnToken token, IWorkflowContext context, CancellationToken cancellationToken)
{
return context.SendMessageAsync(token, cancellationToken: cancellationToken);
}
}
/// <summary>
/// Executor that aggregates the results from the concurrent agents.
/// </summary>
private sealed class ConcurrentAggregationExecutor() :
Executor<ChatMessage>("ConcurrentAggregationExecutor")
private sealed class ConcurrentAggregationExecutor() : Executor<List<ChatMessage>>("ConcurrentAggregationExecutor")
{
private readonly List<ChatMessage> _messages = [];
/// <summary>
/// Handles incoming messages from the agents and aggregates their responses.
/// </summary>
/// <param name="message">The message from the agent</param>
/// <param name="message">The messages from the agent</param>
/// <param name="context">Workflow context for accessing workflow services and adding events</param>
/// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests.
/// The default is <see cref="CancellationToken.None"/>.</param>
public override async ValueTask HandleAsync(ChatMessage message, IWorkflowContext context, CancellationToken cancellationToken = default)
public override async ValueTask HandleAsync(List<ChatMessage> message, IWorkflowContext context, CancellationToken cancellationToken = default)
{
this._messages.Add(message);
this._messages.AddRange(message);
if (this._messages.Count == 2)
{
dotnet/samples/GettingStarted/Workflows/Concurrent/Concurrent/Program.cs
+4 -4
View File
@@ -97,21 +97,21 @@ internal sealed class ConcurrentStartExecutor() :
/// Executor that aggregates the results from the concurrent agents.
/// </summary>
internal sealed class ConcurrentAggregationExecutor() :
Executor<ChatMessage>("ConcurrentAggregationExecutor")
Executor<List<ChatMessage>>("ConcurrentAggregationExecutor")
{
private readonly List<ChatMessage> _messages = [];
/// <summary>
/// Handles incoming messages from the agents and aggregates their responses.
/// </summary>
/// <param name="message">The message from the agent</param>
/// <param name="message">The messages from the agent</param>
/// <param name="context">Workflow context for accessing workflow services and adding events</param>
/// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests.
/// The default is <see cref="CancellationToken.None"/>.</param>
/// <returns>A task representing the asynchronous operation</returns>
public override async ValueTask HandleAsync(ChatMessage message, IWorkflowContext context, CancellationToken cancellationToken = default)
public override async ValueTask HandleAsync(List<ChatMessage> message, IWorkflowContext context, CancellationToken cancellationToken = default)
{
this._messages.Add(message);
this._messages.AddRange(message);
if (this._messages.Count == 2)
{
dotnet/samples/GettingStarted/Workflows/Observability/WorkflowAsAnAgent/Program.cs
+140
View File
@@ -0,0 +1,140 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics;
using Azure.AI.OpenAI;
using Azure.Identity;
using Azure.Monitor.OpenTelemetry.Exporter;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;
using OpenTelemetry;
using OpenTelemetry.Resources;
using OpenTelemetry.Trace;
namespace WorkflowAsAnAgentObservabilitySample;
/// <summary>
/// This sample shows how to enable OpenTelemetry observability for workflows when
/// using them as <see cref="AIAgent"/>s.
///
/// In this example, we create a workflow that uses two language agents to process
/// input concurrently, one that responds in French and another that responds in English.
///
/// You will interact with the workflow in an interactive loop, sending messages and receiving
/// streaming responses from the workflow as if it were an agent who responds in both languages.
///
/// OpenTelemetry observability is enabled at multiple levels:
/// 1. At the chat client level, capturing telemetry for interactions with the Azure OpenAI service.
/// 2. At the agent level, capturing telemetry for agent operations.
/// 3. At the workflow level, capturing telemetry for workflow execution.
///
/// Traces will be sent to an Aspire dashboard via an OTLP endpoint, and optionally to
/// Azure Monitor if an Application Insights connection string is provided.
///
/// Learn how to set up an Aspire dashboard here:
/// https://learn.microsoft.com/en-us/dotnet/aspire/fundamentals/dashboard/standalone?tabs=bash
/// </summary>
/// <remarks>
/// Pre-requisites:
/// - Foundational samples should be completed first.
/// - This sample uses concurrent processing.
/// - An Azure OpenAI endpoint and deployment name.
/// - An Application Insights resource for telemetry (optional).
/// </remarks>
public static class Program
{
private const string SourceName = "Workflow.ApplicationInsightsSample";
private static readonly ActivitySource s_activitySource = new(SourceName);
private static async Task Main()
{
// Set up observability
var applicationInsightsConnectionString = Environment.GetEnvironmentVariable("APPLICATIONINSIGHTS_CONNECTION_STRING");
var otlpEndpoint = Environment.GetEnvironmentVariable("OTLP_ENDPOINT") ?? "http://localhost:4317";
var resourceBuilder = ResourceBuilder
.CreateDefault()
.AddService("WorkflowSample");
var traceProviderBuilder = Sdk.CreateTracerProviderBuilder()
.SetResourceBuilder(resourceBuilder)
.AddSource("Microsoft.Agents.AI.*") // Agent Framework telemetry
.AddSource("Microsoft.Extensions.AI.*") // Extensions AI telemetry
.AddSource(SourceName);
traceProviderBuilder.AddOtlpExporter(options => options.Endpoint = new Uri(otlpEndpoint));
if (!string.IsNullOrWhiteSpace(applicationInsightsConnectionString))
{
traceProviderBuilder.AddAzureMonitorTraceExporter(options => options.ConnectionString = applicationInsightsConnectionString);
}
using var traceProvider = traceProviderBuilder.Build();
// Set up the Azure OpenAI client
var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
var deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT_NAME") ?? "gpt-4o-mini";
var chatClient = new AzureOpenAIClient(new Uri(endpoint), new AzureCliCredential())
.GetChatClient(deploymentName)
.AsIChatClient()
.AsBuilder()
.UseOpenTelemetry(sourceName: SourceName, configure: (cfg) => cfg.EnableSensitiveData = true) // enable telemetry at the chat client level
.Build();
// Start a root activity for the application
using var activity = s_activitySource.StartActivity("main");
Console.WriteLine($"Operation/Trace ID: {Activity.Current?.TraceId}");
// Create the workflow and turn it into an agent with OpenTelemetry instrumentation
var workflow = WorkflowHelper.GetWorkflow(chatClient, SourceName);
var agent = new OpenTelemetryAgent(workflow.AsAgent("workflow-agent", "Workflow Agent"), SourceName)
{
EnableSensitiveData = true // enable sensitive data at the agent level such as prompts and responses
};
var thread = agent.GetNewThread();
// Start an interactive loop to interact with the workflow as if it were an agent
while (true)
{
Console.WriteLine();
Console.Write("User (or 'exit' to quit): ");
string? input = Console.ReadLine();
if (string.IsNullOrWhiteSpace(input) || input.Equals("exit", StringComparison.OrdinalIgnoreCase))
{
break;
}
await ProcessInputAsync(agent, thread, input);
}
// Helper method to process user input and display streaming responses. To display
// multiple interleaved responses correctly, we buffer updates by message ID and
// re-render all messages on each update.
static async Task ProcessInputAsync(AIAgent agent, AgentThread thread, string input)
{
Dictionary<string, List<AgentRunResponseUpdate>> buffer = [];
await foreach (AgentRunResponseUpdate update in agent.RunStreamingAsync(input, thread))
{
if (update.MessageId is null || string.IsNullOrEmpty(update.Text))
{
// skip updates that don't have a message ID or text
continue;
}
Console.Clear();
if (!buffer.TryGetValue(update.MessageId, out List<AgentRunResponseUpdate>? value))
{
value = [];
buffer[update.MessageId] = value;
}
value.Add(update);
foreach (var (messageId, segments) in buffer)
{
string combinedText = string.Concat(segments);
Console.WriteLine($"{segments[0].AuthorName}: {combinedText}");
Console.WriteLine();
}
}
}
}
}
dotnet/samples/GettingStarted/Workflows/Observability/WorkflowAsAnAgent/WorkflowAsAnAgentObservability.csproj
+27
View File
@@ -0,0 +1,27 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net9.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Azure.AI.OpenAI" />
<PackageReference Include="Azure.Identity" />
<PackageReference Include="Azure.Monitor.OpenTelemetry.Exporter" />
<PackageReference Include="Microsoft.Extensions.AI.OpenAI" />
<PackageReference Include="OpenTelemetry" />
<PackageReference Include="OpenTelemetry.Exporter.OpenTelemetryProtocol" />
<PackageReference Include="System.Diagnostics.DiagnosticSource" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\..\..\..\src\Microsoft.Agents.AI.Workflows\Microsoft.Agents.AI.Workflows.csproj" />
<ProjectReference Include="..\..\..\..\..\src\Microsoft.Agents.AI.AzureAI\Microsoft.Agents.AI.AzureAI.csproj" />
<ProjectReference Include="..\..\..\..\..\src\Microsoft.Agents.AI\Microsoft.Agents.AI.csproj" />
</ItemGroup>
</Project>
dotnet/samples/GettingStarted/Workflows/Observability/WorkflowAsAnAgent/WorkflowHelper.cs
+98
View File
@@ -0,0 +1,98 @@
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;
namespace WorkflowAsAnAgentObservabilitySample;
internal static class WorkflowHelper
{
/// <summary>
/// Creates a workflow that uses two language agents to process input concurrently.
/// </summary>
/// <param name="chatClient">The chat client to use for the agents</param>
/// <param name="sourceName">The source name for OpenTelemetry instrumentation</param>
/// <returns>A workflow that processes input using two language agents</returns>
internal static Workflow GetWorkflow(IChatClient chatClient, string sourceName)
{
// Create executors
var startExecutor = new ConcurrentStartExecutor();
var aggregationExecutor = new ConcurrentAggregationExecutor();
AIAgent frenchAgent = GetLanguageAgent("French", chatClient, sourceName);
AIAgent englishAgent = GetLanguageAgent("English", chatClient, sourceName);
// Build the workflow by adding executors and connecting them
return new WorkflowBuilder(startExecutor)
.AddFanOutEdge(startExecutor, targets: [frenchAgent, englishAgent])
.AddFanInEdge(aggregationExecutor, sources: [frenchAgent, englishAgent])
.WithOutputFrom(aggregationExecutor)
.Build();
}
/// <summary>
/// Creates a language agent for the specified target language.
/// </summary>
/// <param name="targetLanguage">The target language for translation</param>
/// <param name="chatClient">The chat client to use for the agent</param>
/// <param name="sourceName">The source name for OpenTelemetry instrumentation</param>
/// <returns>An AIAgent configured for the specified language</returns>
private static AIAgent GetLanguageAgent(string targetLanguage, IChatClient chatClient, string sourceName) =>
new ChatClientAgent(
chatClient,
instructions: $"You're a helpful assistant who always responds in {targetLanguage}.",
name: $"{targetLanguage}Agent"
)
.AsBuilder()
.UseOpenTelemetry(sourceName, configure: (cfg) => cfg.EnableSensitiveData = true) // enable telemetry at the agent level
.Build();
/// <summary>
/// Executor that starts the concurrent processing by sending messages to the agents.
/// </summary>
private sealed class ConcurrentStartExecutor() : Executor("ConcurrentStartExecutor")
{
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder)
{
return routeBuilder
.AddHandler<List<ChatMessage>>(this.RouteMessages)
.AddHandler<TurnToken>(this.RouteTurnTokenAsync);
}
private ValueTask RouteMessages(List<ChatMessage> messages, IWorkflowContext context, CancellationToken cancellationToken)
{
return context.SendMessageAsync(messages, cancellationToken: cancellationToken);
}
private ValueTask RouteTurnTokenAsync(TurnToken token, IWorkflowContext context, CancellationToken cancellationToken)
{
return context.SendMessageAsync(token, cancellationToken: cancellationToken);
}
}
/// <summary>
/// Executor that aggregates the results from the concurrent agents.
/// </summary>
private sealed class ConcurrentAggregationExecutor() : Executor<List<ChatMessage>>("ConcurrentAggregationExecutor")
{
private readonly List<ChatMessage> _messages = [];
/// <summary>
/// Handles incoming messages from the agents and aggregates their responses.
/// </summary>
/// <param name="message">The message from the agent</param>
/// <param name="context">Workflow context for accessing workflow services and adding events</param>
/// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests.
/// The default is <see cref="CancellationToken.None"/>.</param>
public override async ValueTask HandleAsync(List<ChatMessage> message, IWorkflowContext context, CancellationToken cancellationToken = default)
{
this._messages.AddRange(message);
if (this._messages.Count == 2)
{
var formattedMessages = string.Join(Environment.NewLine, this._messages.Select(m => $"{m.Text}"));
await context.YieldOutputAsync(formattedMessages, cancellationToken);
}
}
}
}