diff --git a/dotnet/agent-framework-dotnet.slnx b/dotnet/agent-framework-dotnet.slnx index 002efdbab1..83e38eb685 100644 --- a/dotnet/agent-framework-dotnet.slnx +++ b/dotnet/agent-framework-dotnet.slnx @@ -35,6 +35,18 @@ + + + + + + + + + + + + diff --git a/dotnet/samples/DurableAgents/ConsoleApps/01_SingleAgent/01_SingleAgent.csproj b/dotnet/samples/DurableAgents/ConsoleApps/01_SingleAgent/01_SingleAgent.csproj new file mode 100644 index 0000000000..6dc2007867 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/01_SingleAgent/01_SingleAgent.csproj @@ -0,0 +1,30 @@ + + + net10.0 + Exe + enable + enable + SingleAgent + SingleAgent + + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/DurableAgents/ConsoleApps/01_SingleAgent/Program.cs b/dotnet/samples/DurableAgents/ConsoleApps/01_SingleAgent/Program.cs new file mode 100644 index 0000000000..e681df067f --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/01_SingleAgent/Program.cs @@ -0,0 +1,103 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Azure; +using Azure.AI.OpenAI; +using Azure.Identity; +using Microsoft.Agents.AI; +using Microsoft.Agents.AI.DurableTask; +using Microsoft.DurableTask.Client.AzureManaged; +using Microsoft.DurableTask.Worker.AzureManaged; +using Microsoft.Extensions.AI; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using OpenAI.Chat; + +// Get the Azure OpenAI endpoint and deployment name from environment variables. +string endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") + ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set."); +string deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT") + ?? throw new InvalidOperationException("AZURE_OPENAI_DEPLOYMENT is not set."); + +// Get DTS connection string from environment variable +string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING") + ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"; + +// Use Azure Key Credential if provided, otherwise use Azure CLI Credential. +string? azureOpenAiKey = Environment.GetEnvironmentVariable("AZURE_OPENAI_KEY"); +AzureOpenAIClient client = !string.IsNullOrEmpty(azureOpenAiKey) + ? new AzureOpenAIClient(new Uri(endpoint), new AzureKeyCredential(azureOpenAiKey)) + : new AzureOpenAIClient(new Uri(endpoint), new AzureCliCredential()); + +// Set up an AI agent following the standard Microsoft Agent Framework pattern. +const string JokerName = "Joker"; +const string JokerInstructions = "You are good at telling jokes."; + +AIAgent agent = client.GetChatClient(deploymentName).CreateAIAgent(JokerInstructions, JokerName); + +// Configure the console app to host the AI agent. +IHost host = Host.CreateDefaultBuilder(args) + .ConfigureLogging(logging => logging.SetMinimumLevel(LogLevel.Warning)) + .ConfigureServices(services => + { + services.ConfigureDurableAgents( + options => options.AddAIAgent(agent, timeToLive: TimeSpan.FromHours(1)), + workerBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString), + clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString)); + }) + .Build(); + +await host.StartAsync(); + +// Get the agent proxy from services +IServiceProvider services = host.Services; +AIAgent agentProxy = services.GetRequiredKeyedService(JokerName); + +// Console colors for better UX +Console.ForegroundColor = ConsoleColor.Cyan; +Console.WriteLine("=== Single Agent Console Sample ==="); +Console.ResetColor(); +Console.WriteLine("Enter a message for the Joker agent (or 'exit' to quit):"); +Console.WriteLine(); + +// Create a thread for the conversation +AgentThread thread = agentProxy.GetNewThread(); + +while (true) +{ + // Read input from stdin + Console.ForegroundColor = ConsoleColor.Yellow; + Console.Write("You: "); + Console.ResetColor(); + + string? input = Console.ReadLine(); + if (string.IsNullOrWhiteSpace(input) || input.Equals("exit", StringComparison.OrdinalIgnoreCase)) + { + break; + } + + // Run the agent + Console.ForegroundColor = ConsoleColor.Green; + Console.Write("Joker: "); + Console.ResetColor(); + + try + { + AgentRunResponse agentResponse = await agentProxy.RunAsync( + message: input, + thread: thread, + cancellationToken: CancellationToken.None); + + Console.WriteLine(agentResponse.Text); + Console.WriteLine(); + } + catch (Exception ex) + { + Console.ForegroundColor = ConsoleColor.Red; + Console.Error.WriteLine($"Error: {ex.Message}"); + Console.ResetColor(); + Console.WriteLine(); + } +} + +await host.StopAsync(); diff --git a/dotnet/samples/DurableAgents/ConsoleApps/01_SingleAgent/README.md b/dotnet/samples/DurableAgents/ConsoleApps/01_SingleAgent/README.md new file mode 100644 index 0000000000..7c921b0d87 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/01_SingleAgent/README.md @@ -0,0 +1,56 @@ +# Single Agent Sample + +This sample demonstrates how to use the durable agents extension to create a simple console app that hosts a single AI agent and provides interactive conversation via stdin/stdout. + +## Key Concepts Demonstrated + +- Using the Microsoft Agent Framework to define a simple AI agent with a name and instructions. +- Registering durable agents with the console app and running them interactively. +- Conversation management (via threads) for isolated interactions. + +## Environment Setup + +See the [README.md](../README.md) file in the parent directory for more information on how to configure the environment, including how to install and run common sample dependencies. + +## Running the Sample + +With the environment setup, you can run the sample: + +```bash +cd dotnet/samples/DurableAgents/ConsoleApps/01_SingleAgent +dotnet run --framework net10.0 +``` + +The app will prompt you for input. You can interact with the Joker agent: + +```text +=== Single Agent Console Sample === +Enter a message for the Joker agent (or 'exit' to quit): + +You: Tell me a joke about a pirate. +Joker: Why don't pirates ever learn the alphabet? Because they always get stuck at "C"! + +You: Now explain the joke. +Joker: The joke plays on the word "sea" (C), which pirates are famously associated with... + +You: exit +``` + +## Scriptable Usage + +You can also pipe input to the app for scriptable usage: + +```bash +echo "Tell me a joke about a pirate." | dotnet run +``` + +The app will read from stdin, process the input, and write the response to stdout. + +## Viewing Agent State + +You can view the state of the agent in the Durable Task Scheduler dashboard: + +1. Open your browser and navigate to `http://localhost:8082` +2. In the dashboard, you can view the state of the Joker agent, including its conversation history and current state + +The agent maintains conversation state across multiple interactions, and you can inspect this state in the dashboard to understand how the durable agents extension manages conversation context. diff --git a/dotnet/samples/DurableAgents/ConsoleApps/02_AgentOrchestration_Chaining/02_AgentOrchestration_Chaining.csproj b/dotnet/samples/DurableAgents/ConsoleApps/02_AgentOrchestration_Chaining/02_AgentOrchestration_Chaining.csproj new file mode 100644 index 0000000000..ef74da183b --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/02_AgentOrchestration_Chaining/02_AgentOrchestration_Chaining.csproj @@ -0,0 +1,30 @@ + + + net10.0 + Exe + enable + enable + AgentOrchestration_Chaining + AgentOrchestration_Chaining + + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/DurableAgents/ConsoleApps/02_AgentOrchestration_Chaining/Models.cs b/dotnet/samples/DurableAgents/ConsoleApps/02_AgentOrchestration_Chaining/Models.cs new file mode 100644 index 0000000000..593b468457 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/02_AgentOrchestration_Chaining/Models.cs @@ -0,0 +1,6 @@ +// Copyright (c) Microsoft. All rights reserved. + +namespace AgentOrchestration_Chaining; + +// Response model +public sealed record TextResponse(string Text); diff --git a/dotnet/samples/DurableAgents/ConsoleApps/02_AgentOrchestration_Chaining/Program.cs b/dotnet/samples/DurableAgents/ConsoleApps/02_AgentOrchestration_Chaining/Program.cs new file mode 100644 index 0000000000..a7c04f74e6 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/02_AgentOrchestration_Chaining/Program.cs @@ -0,0 +1,148 @@ +// Copyright (c) Microsoft. All rights reserved. + +using AgentOrchestration_Chaining; +using Azure; +using Azure.AI.OpenAI; +using Azure.Identity; +using Microsoft.Agents.AI; +using Microsoft.Agents.AI.DurableTask; +using Microsoft.DurableTask; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Client.AzureManaged; +using Microsoft.DurableTask.Worker; +using Microsoft.DurableTask.Worker.AzureManaged; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using OpenAI.Chat; +using Environment = System.Environment; + +// Get the Azure OpenAI endpoint and deployment name from environment variables. +string endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") + ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set."); +string deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT") + ?? throw new InvalidOperationException("AZURE_OPENAI_DEPLOYMENT is not set."); + +// Get DTS connection string from environment variable +string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING") + ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"; + +// Use Azure Key Credential if provided, otherwise use Azure CLI Credential. +string? azureOpenAiKey = Environment.GetEnvironmentVariable("AZURE_OPENAI_KEY"); +AzureOpenAIClient client = !string.IsNullOrEmpty(azureOpenAiKey) + ? new AzureOpenAIClient(new Uri(endpoint), new AzureKeyCredential(azureOpenAiKey)) + : new AzureOpenAIClient(new Uri(endpoint), new AzureCliCredential()); + +// Single agent used by the orchestration to demonstrate sequential calls on the same thread. +const string WriterName = "WriterAgent"; +const string WriterInstructions = + """ + You refine short pieces of text. When given an initial sentence you enhance it; + when given an improved sentence you polish it further. + """; + +AIAgent writerAgent = client.GetChatClient(deploymentName).CreateAIAgent(WriterInstructions, WriterName); + +// Orchestrator function +static async Task RunOrchestratorAsync(TaskOrchestrationContext context) +{ + DurableAIAgent writer = context.GetAgent("WriterAgent"); + AgentThread writerThread = writer.GetNewThread(); + + AgentRunResponse initial = await writer.RunAsync( + message: "Write a concise inspirational sentence about learning.", + thread: writerThread); + + AgentRunResponse refined = await writer.RunAsync( + message: $"Improve this further while keeping it under 25 words: {initial.Result.Text}", + thread: writerThread); + + return refined.Result.Text; +} + +// Configure the console app to host the AI agent. +IHost host = Host.CreateDefaultBuilder(args) + .ConfigureLogging(loggingBuilder => loggingBuilder.SetMinimumLevel(LogLevel.Warning)) + .ConfigureServices(services => + { + services.ConfigureDurableAgents( + options => options.AddAIAgent(writerAgent), + workerBuilder: builder => + { + builder.UseDurableTaskScheduler(dtsConnectionString); + builder.AddTasks(registry => registry.AddOrchestratorFunc(nameof(RunOrchestratorAsync), RunOrchestratorAsync)); + }, + clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString)); + }) + .Build(); + +await host.StartAsync(); + +DurableTaskClient durableClient = host.Services.GetRequiredService(); + +// Console colors for better UX +Console.ForegroundColor = ConsoleColor.Cyan; +Console.WriteLine("=== Single Agent Orchestration Chaining Sample ==="); +Console.ResetColor(); +Console.WriteLine("Starting orchestration..."); +Console.WriteLine(); + +try +{ + // Start the orchestration + string instanceId = await durableClient.ScheduleNewOrchestrationInstanceAsync( + orchestratorName: nameof(RunOrchestratorAsync)); + + Console.ForegroundColor = ConsoleColor.Gray; + Console.WriteLine($"Orchestration started with instance ID: {instanceId}"); + Console.WriteLine("Waiting for completion..."); + Console.ResetColor(); + + // Wait for orchestration to complete + OrchestrationMetadata status = await durableClient.WaitForInstanceCompletionAsync( + instanceId, + getInputsAndOutputs: true, + CancellationToken.None); + + Console.WriteLine(); + + if (status.RuntimeStatus == OrchestrationRuntimeStatus.Completed) + { + Console.ForegroundColor = ConsoleColor.Green; + Console.WriteLine("✓ Orchestration completed successfully!"); + Console.ResetColor(); + Console.WriteLine(); + Console.ForegroundColor = ConsoleColor.Yellow; + Console.Write("Result: "); + Console.ResetColor(); + Console.WriteLine(status.ReadOutputAs()); + } + else if (status.RuntimeStatus == OrchestrationRuntimeStatus.Failed) + { + Console.ForegroundColor = ConsoleColor.Red; + Console.WriteLine("✗ Orchestration failed!"); + Console.ResetColor(); + if (status.FailureDetails != null) + { + Console.WriteLine($"Error: {status.FailureDetails.ErrorMessage}"); + } + Environment.Exit(1); + } + else + { + Console.ForegroundColor = ConsoleColor.Yellow; + Console.WriteLine($"Orchestration status: {status.RuntimeStatus}"); + Console.ResetColor(); + } +} +catch (Exception ex) +{ + Console.ForegroundColor = ConsoleColor.Red; + Console.Error.WriteLine($"Error: {ex.Message}"); + Console.ResetColor(); + Environment.Exit(1); +} +finally +{ + await host.StopAsync(); +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/02_AgentOrchestration_Chaining/README.md b/dotnet/samples/DurableAgents/ConsoleApps/02_AgentOrchestration_Chaining/README.md new file mode 100644 index 0000000000..715a72ada0 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/02_AgentOrchestration_Chaining/README.md @@ -0,0 +1,53 @@ +# Single Agent Orchestration Sample + +This sample demonstrates how to use the durable agents extension to create a simple console app that orchestrates sequential calls to a single AI agent using the same conversation thread for context continuity. + +## Key Concepts Demonstrated + +- Orchestrating multiple interactions with the same agent in a deterministic order +- Using the same `AgentThread` across multiple calls to maintain conversational context +- Durable orchestration with automatic checkpointing and resumption from failures +- Waiting for orchestration completion using `WaitForInstanceCompletionAsync` + +## Environment Setup + +See the [README.md](../README.md) file in the parent directory for more information on how to configure the environment, including how to install and run common sample dependencies. + +## Running the Sample + +With the environment setup, you can run the sample: + +```bash +cd dotnet/samples/DurableAgents/ConsoleApps/02_AgentOrchestration_Chaining +dotnet run --framework net10.0 +``` + +The app will start the orchestration, wait for it to complete, and display the result: + +```text +=== Single Agent Orchestration Chaining Sample === +Starting orchestration... + +Orchestration started with instance ID: 86313f1d45fb42eeb50b1852626bf3ff +Waiting for completion... + +✓ Orchestration completed successfully! + +Result: Learning serves as the key, opening doors to boundless opportunities and a brighter future. +``` + +The orchestration will proceed to run the WriterAgent twice in sequence: + +1. First, it writes an inspirational sentence about learning +2. Then, it refines the initial output using the same conversation thread + +## Viewing Orchestration State + +You can view the state of the orchestration in the Durable Task Scheduler dashboard: + +1. Open your browser and navigate to `http://localhost:8082` +2. In the dashboard, you can see: + - **Orchestrations**: View the orchestration instance, including its runtime status, input, output, and execution history + - **Agents**: View the state of the WriterAgent, including conversation history maintained across the orchestration steps + +The orchestration instance ID is displayed in the console output. You can use this ID to find the specific orchestration in the dashboard and inspect its execution details, including the sequence of agent calls and their results. diff --git a/dotnet/samples/DurableAgents/ConsoleApps/03_AgentOrchestration_Concurrency/03_AgentOrchestration_Concurrency.csproj b/dotnet/samples/DurableAgents/ConsoleApps/03_AgentOrchestration_Concurrency/03_AgentOrchestration_Concurrency.csproj new file mode 100644 index 0000000000..017b5fe300 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/03_AgentOrchestration_Concurrency/03_AgentOrchestration_Concurrency.csproj @@ -0,0 +1,30 @@ + + + net10.0 + Exe + enable + enable + AgentOrchestration_Concurrency + AgentOrchestration_Concurrency + + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/DurableAgents/ConsoleApps/03_AgentOrchestration_Concurrency/Models.cs b/dotnet/samples/DurableAgents/ConsoleApps/03_AgentOrchestration_Concurrency/Models.cs new file mode 100644 index 0000000000..042e245f7f --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/03_AgentOrchestration_Concurrency/Models.cs @@ -0,0 +1,6 @@ +// Copyright (c) Microsoft. All rights reserved. + +namespace AgentOrchestration_Concurrency; + +// Response model +public sealed record TextResponse(string Text); diff --git a/dotnet/samples/DurableAgents/ConsoleApps/03_AgentOrchestration_Concurrency/Program.cs b/dotnet/samples/DurableAgents/ConsoleApps/03_AgentOrchestration_Concurrency/Program.cs new file mode 100644 index 0000000000..d955d1beed --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/03_AgentOrchestration_Concurrency/Program.cs @@ -0,0 +1,191 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Text.Json; +using AgentOrchestration_Concurrency; +using Azure; +using Azure.AI.OpenAI; +using Azure.Identity; +using Microsoft.Agents.AI; +using Microsoft.Agents.AI.DurableTask; +using Microsoft.DurableTask; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Client.AzureManaged; +using Microsoft.DurableTask.Worker; +using Microsoft.DurableTask.Worker.AzureManaged; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using OpenAI.Chat; + +// Get the Azure OpenAI endpoint and deployment name from environment variables. +string endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") + ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set."); +string deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT") + ?? throw new InvalidOperationException("AZURE_OPENAI_DEPLOYMENT is not set."); + +// Get DTS connection string from environment variable +string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING") + ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"; + +// Use Azure Key Credential if provided, otherwise use Azure CLI Credential. +string? azureOpenAiKey = Environment.GetEnvironmentVariable("AZURE_OPENAI_KEY"); +AzureOpenAIClient client = !string.IsNullOrEmpty(azureOpenAiKey) + ? new AzureOpenAIClient(new Uri(endpoint), new AzureKeyCredential(azureOpenAiKey)) + : new AzureOpenAIClient(new Uri(endpoint), new AzureCliCredential()); + +// Two agents used by the orchestration to demonstrate concurrent execution. +const string PhysicistName = "PhysicistAgent"; +const string PhysicistInstructions = "You are an expert in physics. You answer questions from a physics perspective."; + +const string ChemistName = "ChemistAgent"; +const string ChemistInstructions = "You are a middle school chemistry teacher. You answer questions so that middle school students can understand."; + +AIAgent physicistAgent = client.GetChatClient(deploymentName).CreateAIAgent(PhysicistInstructions, PhysicistName); +AIAgent chemistAgent = client.GetChatClient(deploymentName).CreateAIAgent(ChemistInstructions, ChemistName); + +// Orchestrator function +static async Task RunOrchestratorAsync(TaskOrchestrationContext context, string prompt) +{ + // Get both agents + DurableAIAgent physicist = context.GetAgent(PhysicistName); + DurableAIAgent chemist = context.GetAgent(ChemistName); + + // Start both agent runs concurrently + Task> physicistTask = physicist.RunAsync(prompt); + Task> chemistTask = chemist.RunAsync(prompt); + + // Wait for both tasks to complete using Task.WhenAll + await Task.WhenAll(physicistTask, chemistTask); + + // Get the results + TextResponse physicistResponse = (await physicistTask).Result; + TextResponse chemistResponse = (await chemistTask).Result; + + // Return the result as a structured, anonymous type + return new + { + physicist = physicistResponse.Text, + chemist = chemistResponse.Text, + }; +} + +// Configure the console app to host the AI agents. +IHost host = Host.CreateDefaultBuilder(args) + .ConfigureLogging(loggingBuilder => loggingBuilder.SetMinimumLevel(LogLevel.Warning)) + .ConfigureServices(services => + { + services.ConfigureDurableAgents( + options => + { + options + .AddAIAgent(physicistAgent) + .AddAIAgent(chemistAgent); + }, + workerBuilder: builder => + { + builder.UseDurableTaskScheduler(dtsConnectionString); + builder.AddTasks( + registry => registry.AddOrchestratorFunc(nameof(RunOrchestratorAsync), RunOrchestratorAsync)); + }, + clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString)); + }) + .Build(); + +await host.StartAsync(); + +DurableTaskClient durableTaskClient = host.Services.GetRequiredService(); + +// Console colors for better UX +Console.ForegroundColor = ConsoleColor.Cyan; +Console.WriteLine("=== Multi-Agent Concurrent Orchestration Sample ==="); +Console.ResetColor(); +Console.WriteLine("Enter a question for the agents:"); +Console.WriteLine(); + +// Read prompt from stdin +string? prompt = Console.ReadLine(); +if (string.IsNullOrWhiteSpace(prompt)) +{ + Console.ForegroundColor = ConsoleColor.Red; + Console.Error.WriteLine("Error: Prompt is required."); + Console.ResetColor(); + Environment.Exit(1); + return; +} + +Console.WriteLine(); +Console.ForegroundColor = ConsoleColor.Gray; +Console.WriteLine("Starting orchestration..."); +Console.ResetColor(); + +try +{ + // Start the orchestration + string instanceId = await durableTaskClient.ScheduleNewOrchestrationInstanceAsync( + orchestratorName: nameof(RunOrchestratorAsync), + input: prompt); + + Console.ForegroundColor = ConsoleColor.Gray; + Console.WriteLine($"Orchestration started with instance ID: {instanceId}"); + Console.WriteLine("Waiting for completion..."); + Console.ResetColor(); + + // Wait for orchestration to complete + OrchestrationMetadata status = await durableTaskClient.WaitForInstanceCompletionAsync( + instanceId, + getInputsAndOutputs: true, + CancellationToken.None); + + Console.WriteLine(); + + if (status.RuntimeStatus == OrchestrationRuntimeStatus.Completed) + { + Console.ForegroundColor = ConsoleColor.Green; + Console.WriteLine("✓ Orchestration completed successfully!"); + Console.ResetColor(); + Console.WriteLine(); + + // Parse the output + using JsonDocument doc = JsonDocument.Parse(status.SerializedOutput!); + JsonElement output = doc.RootElement; + + Console.ForegroundColor = ConsoleColor.Yellow; + Console.WriteLine("Physicist's response:"); + Console.ResetColor(); + Console.WriteLine(output.GetProperty("physicist").GetString()); + Console.WriteLine(); + + Console.ForegroundColor = ConsoleColor.Yellow; + Console.WriteLine("Chemist's response:"); + Console.ResetColor(); + Console.WriteLine(output.GetProperty("chemist").GetString()); + } + else if (status.RuntimeStatus == OrchestrationRuntimeStatus.Failed) + { + Console.ForegroundColor = ConsoleColor.Red; + Console.WriteLine("✗ Orchestration failed!"); + Console.ResetColor(); + if (status.FailureDetails != null) + { + Console.WriteLine($"Error: {status.FailureDetails.ErrorMessage}"); + } + Environment.Exit(1); + } + else + { + Console.ForegroundColor = ConsoleColor.Yellow; + Console.WriteLine($"Orchestration status: {status.RuntimeStatus}"); + Console.ResetColor(); + } +} +catch (Exception ex) +{ + Console.ForegroundColor = ConsoleColor.Red; + Console.Error.WriteLine($"Error: {ex.Message}"); + Console.ResetColor(); + Environment.Exit(1); +} +finally +{ + await host.StopAsync(); +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/03_AgentOrchestration_Concurrency/README.md b/dotnet/samples/DurableAgents/ConsoleApps/03_AgentOrchestration_Concurrency/README.md new file mode 100644 index 0000000000..2ac1a504c8 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/03_AgentOrchestration_Concurrency/README.md @@ -0,0 +1,68 @@ +# Multi-Agent Concurrent Orchestration Sample + +This sample demonstrates how to use the durable agents extension to create a console app that orchestrates concurrent execution of multiple AI agents using durable orchestration. + +## Key Concepts Demonstrated + +- Running multiple agents concurrently in a single orchestration +- Using `Task.WhenAll` to wait for concurrent agent executions +- Combining results from multiple agents into a single response +- Waiting for orchestration completion using `WaitForInstanceCompletionAsync` + +## Environment Setup + +See the [README.md](../README.md) file in the parent directory for more information on how to configure the environment, including how to install and run common sample dependencies. + +## Running the Sample + +With the environment setup, you can run the sample: + +```bash +cd dotnet/samples/DurableAgents/ConsoleApps/03_AgentOrchestration_Concurrency +dotnet run --framework net10.0 +``` + +The app will prompt you for a question: + +```text +=== Multi-Agent Concurrent Orchestration Sample === +Enter a question for the agents: + +What is temperature? +``` + +The orchestration will run both agents concurrently and display their responses: + +```text +Orchestration started with instance ID: 86313f1d45fb42eeb50b1852626bf3ff +Waiting for completion... + +✓ Orchestration completed successfully! + +Physicist's response: +Temperature is a measure of the average kinetic energy of particles in a system... + +Chemist's response: +From a chemistry perspective, temperature is crucial for chemical reactions... +``` + +Both agents run in parallel, and the orchestration waits for both to complete before returning the combined results. + +## Viewing Orchestration State + +You can view the state of the orchestration in the Durable Task Scheduler dashboard: + +1. Open your browser and navigate to `http://localhost:8082` +2. In the dashboard, you can see: + - **Orchestrations**: View the orchestration instance, including its runtime status, input, output, and execution history + - **Agents**: View the state of both the PhysicistAgent and ChemistAgent, including their individual conversation histories + +The orchestration instance ID is displayed in the console output. You can use this ID to find the specific orchestration in the dashboard and inspect how the concurrent agent executions were coordinated, including the timing of when each agent started and completed. + +## Scriptable Usage + +You can also pipe input to the app: + +```bash +echo "What is temperature?" | dotnet run +``` diff --git a/dotnet/samples/DurableAgents/ConsoleApps/04_AgentOrchestration_Conditionals/04_AgentOrchestration_Conditionals.csproj b/dotnet/samples/DurableAgents/ConsoleApps/04_AgentOrchestration_Conditionals/04_AgentOrchestration_Conditionals.csproj new file mode 100644 index 0000000000..46e348dfec --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/04_AgentOrchestration_Conditionals/04_AgentOrchestration_Conditionals.csproj @@ -0,0 +1,30 @@ + + + net10.0 + Exe + enable + enable + AgentOrchestration_Conditionals + AgentOrchestration_Conditionals + + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/DurableAgents/ConsoleApps/04_AgentOrchestration_Conditionals/Models.cs b/dotnet/samples/DurableAgents/ConsoleApps/04_AgentOrchestration_Conditionals/Models.cs new file mode 100644 index 0000000000..a39695d7d0 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/04_AgentOrchestration_Conditionals/Models.cs @@ -0,0 +1,38 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Text.Json.Serialization; + +namespace AgentOrchestration_Conditionals; + +/// +/// Represents an email input for spam detection and response generation. +/// +public sealed class Email +{ + [JsonPropertyName("email_id")] + public string EmailId { get; set; } = string.Empty; + + [JsonPropertyName("email_content")] + public string EmailContent { get; set; } = string.Empty; +} + +/// +/// Represents the result of spam detection analysis. +/// +public sealed class DetectionResult +{ + [JsonPropertyName("is_spam")] + public bool IsSpam { get; set; } + + [JsonPropertyName("reason")] + public string Reason { get; set; } = string.Empty; +} + +/// +/// Represents a generated email response. +/// +public sealed class EmailResponse +{ + [JsonPropertyName("response")] + public string Response { get; set; } = string.Empty; +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/04_AgentOrchestration_Conditionals/Program.cs b/dotnet/samples/DurableAgents/ConsoleApps/04_AgentOrchestration_Conditionals/Program.cs new file mode 100644 index 0000000000..2277438755 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/04_AgentOrchestration_Conditionals/Program.cs @@ -0,0 +1,228 @@ +// Copyright (c) Microsoft. All rights reserved. + +using AgentOrchestration_Conditionals; +using Azure; +using Azure.AI.OpenAI; +using Azure.Identity; +using Microsoft.Agents.AI; +using Microsoft.Agents.AI.DurableTask; +using Microsoft.DurableTask; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Client.AzureManaged; +using Microsoft.DurableTask.Worker; +using Microsoft.DurableTask.Worker.AzureManaged; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using OpenAI.Chat; + +// Get the Azure OpenAI endpoint and deployment name from environment variables. +string endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") + ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set."); +string deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT") + ?? throw new InvalidOperationException("AZURE_OPENAI_DEPLOYMENT is not set."); + +// Get DTS connection string from environment variable +string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING") + ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"; + +// Use Azure Key Credential if provided, otherwise use Azure CLI Credential. +string? azureOpenAiKey = Environment.GetEnvironmentVariable("AZURE_OPENAI_KEY"); +AzureOpenAIClient client = !string.IsNullOrEmpty(azureOpenAiKey) + ? new AzureOpenAIClient(new Uri(endpoint), new AzureKeyCredential(azureOpenAiKey)) + : new AzureOpenAIClient(new Uri(endpoint), new AzureCliCredential()); + +// Spam detection agent +const string SpamDetectionAgentName = "SpamDetectionAgent"; +const string SpamDetectionAgentInstructions = + """ + You are an expert email spam detection system. Analyze emails and determine if they are spam. + Return your analysis as JSON with 'is_spam' (boolean) and 'reason' (string) fields. + """; + +// Email assistant agent +const string EmailAssistantAgentName = "EmailAssistantAgent"; +const string EmailAssistantAgentInstructions = + """ + You are a professional email assistant. Draft professional, courteous, and helpful email responses. + Return your response as JSON with a 'response' field containing the reply. + """; + +AIAgent spamDetectionAgent = client.GetChatClient(deploymentName).CreateAIAgent(SpamDetectionAgentInstructions, SpamDetectionAgentName); +AIAgent emailAssistantAgent = client.GetChatClient(deploymentName).CreateAIAgent(EmailAssistantAgentInstructions, EmailAssistantAgentName); + +// Orchestrator function +static async Task RunOrchestratorAsync(TaskOrchestrationContext context, Email email) +{ + // Get the spam detection agent + DurableAIAgent spamDetectionAgent = context.GetAgent(SpamDetectionAgentName); + AgentThread spamThread = spamDetectionAgent.GetNewThread(); + + // Step 1: Check if the email is spam + AgentRunResponse spamDetectionResponse = await spamDetectionAgent.RunAsync( + message: + $""" + Analyze this email for spam content and return a JSON response with 'is_spam' (boolean) and 'reason' (string) fields: + Email ID: {email.EmailId} + Content: {email.EmailContent} + """, + thread: spamThread); + DetectionResult result = spamDetectionResponse.Result; + + // Step 2: Conditional logic based on spam detection result + if (result.IsSpam) + { + // Handle spam email + return await context.CallActivityAsync(nameof(HandleSpamEmail), result.Reason); + } + + // Generate and send response for legitimate email + DurableAIAgent emailAssistantAgent = context.GetAgent(EmailAssistantAgentName); + AgentThread emailThread = emailAssistantAgent.GetNewThread(); + + AgentRunResponse emailAssistantResponse = await emailAssistantAgent.RunAsync( + message: + $""" + Draft a professional response to this email. Return a JSON response with a 'response' field containing the reply: + + Email ID: {email.EmailId} + Content: {email.EmailContent} + """, + thread: emailThread); + + EmailResponse emailResponse = emailAssistantResponse.Result; + + return await context.CallActivityAsync(nameof(SendEmail), emailResponse.Response); +} + +// Activity functions +static void HandleSpamEmail(TaskActivityContext context, string reason) +{ + Console.WriteLine($"Email marked as spam: {reason}"); +} + +static void SendEmail(TaskActivityContext context, string message) +{ + Console.WriteLine($"Email sent: {message}"); +} + +// Configure the console app to host the AI agents. +IHost host = Host.CreateDefaultBuilder(args) + .ConfigureLogging(loggingBuilder => loggingBuilder.SetMinimumLevel(LogLevel.Warning)) + .ConfigureServices(services => + { + services.ConfigureDurableAgents( + options => + { + options + .AddAIAgent(spamDetectionAgent) + .AddAIAgent(emailAssistantAgent); + }, + workerBuilder: builder => + { + builder.UseDurableTaskScheduler(dtsConnectionString); + builder.AddTasks(registry => + { + registry.AddOrchestratorFunc(nameof(RunOrchestratorAsync), RunOrchestratorAsync); + registry.AddActivityFunc(nameof(HandleSpamEmail), HandleSpamEmail); + registry.AddActivityFunc(nameof(SendEmail), SendEmail); + }); + }, + clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString)); + }) + .Build(); + +await host.StartAsync(); + +DurableTaskClient durableTaskClient = host.Services.GetRequiredService(); + +// Console colors for better UX +Console.ForegroundColor = ConsoleColor.Cyan; +Console.WriteLine("=== Multi-Agent Conditional Orchestration Sample ==="); +Console.ResetColor(); +Console.WriteLine("Enter email content:"); +Console.WriteLine(); + +// Read email content from stdin +string? emailContent = Console.ReadLine(); +if (string.IsNullOrWhiteSpace(emailContent)) +{ + Console.ForegroundColor = ConsoleColor.Red; + Console.Error.WriteLine("Error: Email content is required."); + Console.ResetColor(); + Environment.Exit(1); + return; +} + +// Generate email ID automatically +Email email = new() +{ + EmailId = $"email-{Guid.NewGuid():N}", + EmailContent = emailContent +}; + +Console.WriteLine(); +Console.ForegroundColor = ConsoleColor.Gray; +Console.WriteLine("Starting orchestration..."); +Console.ResetColor(); + +try +{ + // Start the orchestration + string instanceId = await durableTaskClient.ScheduleNewOrchestrationInstanceAsync( + orchestratorName: nameof(RunOrchestratorAsync), + input: email); + + Console.ForegroundColor = ConsoleColor.Gray; + Console.WriteLine($"Orchestration started with instance ID: {instanceId}"); + Console.WriteLine("Waiting for completion..."); + Console.ResetColor(); + + // Wait for orchestration to complete + OrchestrationMetadata status = await durableTaskClient.WaitForInstanceCompletionAsync( + instanceId, + getInputsAndOutputs: true, + CancellationToken.None); + + Console.WriteLine(); + + if (status.RuntimeStatus == OrchestrationRuntimeStatus.Completed) + { + Console.ForegroundColor = ConsoleColor.Green; + Console.WriteLine("✓ Orchestration completed successfully!"); + Console.ResetColor(); + Console.WriteLine(); + Console.ForegroundColor = ConsoleColor.Yellow; + Console.Write("Result: "); + Console.ResetColor(); + Console.WriteLine(status.ReadOutputAs()); + } + else if (status.RuntimeStatus == OrchestrationRuntimeStatus.Failed) + { + Console.ForegroundColor = ConsoleColor.Red; + Console.WriteLine("✗ Orchestration failed!"); + Console.ResetColor(); + if (status.FailureDetails != null) + { + Console.WriteLine($"Error: {status.FailureDetails.ErrorMessage}"); + } + Environment.Exit(1); + } + else + { + Console.ForegroundColor = ConsoleColor.Yellow; + Console.WriteLine($"Orchestration status: {status.RuntimeStatus}"); + Console.ResetColor(); + } +} +catch (Exception ex) +{ + Console.ForegroundColor = ConsoleColor.Red; + Console.Error.WriteLine($"Error: {ex.Message}"); + Console.ResetColor(); + Environment.Exit(1); +} +finally +{ + await host.StopAsync(); +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/04_AgentOrchestration_Conditionals/README.md b/dotnet/samples/DurableAgents/ConsoleApps/04_AgentOrchestration_Conditionals/README.md new file mode 100644 index 0000000000..646e5eda4e --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/04_AgentOrchestration_Conditionals/README.md @@ -0,0 +1,95 @@ +# Multi-Agent Conditional Orchestration Sample + +This sample demonstrates how to use the durable agents extension to create a console app that orchestrates multiple AI agents with conditional logic based on the results of previous agent interactions. + +## Key Concepts Demonstrated + +- Multi-agent orchestration with conditional branching +- Using agent responses to determine workflow paths +- Activity functions for non-agent operations +- Waiting for orchestration completion using `WaitForInstanceCompletionAsync` + +## Environment Setup + +See the [README.md](../README.md) file in the parent directory for more information on how to configure the environment, including how to install and run common sample dependencies. + +## Running the Sample + +With the environment setup, you can run the sample: + +```bash +cd dotnet/samples/DurableAgents/ConsoleApps/04_AgentOrchestration_Conditionals +dotnet run --framework net10.0 +``` + +The app will prompt you for email content. You can test both legitimate emails and spam emails: + +### Testing with a Legitimate Email + +```text +=== Multi-Agent Conditional Orchestration Sample === +Enter email content: + +Hi John, I hope you're doing well. I wanted to follow up on our meeting yesterday about the quarterly report. Could you please send me the updated figures by Friday? Thanks! +``` + +The orchestration will analyze the email and display the result: + +```text +Orchestration started with instance ID: 86313f1d45fb42eeb50b1852626bf3ff +Waiting for completion... + +✓ Orchestration completed successfully! + +Result: Email sent: Thank you for your email. I'll prepare the updated figures... +``` + +### Testing with a Spam Email + +```text +=== Multi-Agent Conditional Orchestration Sample === +Enter email content: + +URGENT! You've won $1,000,000! Click here now to claim your prize! Limited time offer! Don't miss out! +``` + +The orchestration will detect it as spam and display: + +```text +Orchestration started with instance ID: 86313f1d45fb42eeb50b1852626bf3ff +Waiting for completion... + +✓ Orchestration completed successfully! + +Result: Email marked as spam: Contains suspicious claims about winning money and urgent action requests... +``` + +## Scriptable Usage + +You can also pipe email content to the app: + +```bash +# Test with a legitimate email +echo "Hi John, I hope you're doing well..." | dotnet run + +# Test with a spam email +echo "URGENT! You've won $1,000,000! Click here now!" | dotnet run +``` + +The orchestration will proceed as follows: + +1. The SpamDetectionAgent analyzes the email to determine if it's spam +2. Based on the result: + - If spam: The orchestration calls the `HandleSpamEmail` activity function + - If not spam: The EmailAssistantAgent drafts a response, then the `SendEmail` activity function is called + +## Viewing Orchestration State + +You can view the state of the orchestration in the Durable Task Scheduler dashboard: + +1. Open your browser and navigate to `http://localhost:8082` +2. In the dashboard, you can see: + - **Orchestrations**: View the orchestration instance, including its runtime status, input, output, and execution history + - **Agents**: View the state of both the SpamDetectionAgent and EmailAssistantAgent + +The orchestration instance ID is displayed in the console output. You can use this ID to find the specific orchestration in the dashboard and inspect the conditional branching logic, including which path was taken based on the spam detection result. diff --git a/dotnet/samples/DurableAgents/ConsoleApps/05_AgentOrchestration_HITL/05_AgentOrchestration_HITL.csproj b/dotnet/samples/DurableAgents/ConsoleApps/05_AgentOrchestration_HITL/05_AgentOrchestration_HITL.csproj new file mode 100644 index 0000000000..21db94a33f --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/05_AgentOrchestration_HITL/05_AgentOrchestration_HITL.csproj @@ -0,0 +1,30 @@ + + + net10.0 + Exe + enable + enable + AgentOrchestration_HITL + AgentOrchestration_HITL + + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/DurableAgents/ConsoleApps/05_AgentOrchestration_HITL/Models.cs b/dotnet/samples/DurableAgents/ConsoleApps/05_AgentOrchestration_HITL/Models.cs new file mode 100644 index 0000000000..1eaf1407eb --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/05_AgentOrchestration_HITL/Models.cs @@ -0,0 +1,44 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Text.Json.Serialization; + +namespace AgentOrchestration_HITL; + +/// +/// Represents the input for the Human-in-the-Loop content generation workflow. +/// +public sealed class ContentGenerationInput +{ + [JsonPropertyName("topic")] + public string Topic { get; set; } = string.Empty; + + [JsonPropertyName("max_review_attempts")] + public int MaxReviewAttempts { get; set; } = 3; + + [JsonPropertyName("approval_timeout_hours")] + public float ApprovalTimeoutHours { get; set; } = 72; +} + +/// +/// Represents the content generated by the writer agent. +/// +public sealed class GeneratedContent +{ + [JsonPropertyName("title")] + public string Title { get; set; } = string.Empty; + + [JsonPropertyName("content")] + public string Content { get; set; } = string.Empty; +} + +/// +/// Represents the human approval response. +/// +public sealed class HumanApprovalResponse +{ + [JsonPropertyName("approved")] + public bool Approved { get; set; } + + [JsonPropertyName("feedback")] + public string Feedback { get; set; } = string.Empty; +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/05_AgentOrchestration_HITL/Program.cs b/dotnet/samples/DurableAgents/ConsoleApps/05_AgentOrchestration_HITL/Program.cs new file mode 100644 index 0000000000..ab24e914ff --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/05_AgentOrchestration_HITL/Program.cs @@ -0,0 +1,333 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Text.Json; +using AgentOrchestration_HITL; +using Azure; +using Azure.AI.OpenAI; +using Azure.Identity; +using Microsoft.Agents.AI; +using Microsoft.Agents.AI.DurableTask; +using Microsoft.DurableTask; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Client.AzureManaged; +using Microsoft.DurableTask.Worker; +using Microsoft.DurableTask.Worker.AzureManaged; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using OpenAI.Chat; + +// Get the Azure OpenAI endpoint and deployment name from environment variables. +string endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") + ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set."); +string deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT") + ?? throw new InvalidOperationException("AZURE_OPENAI_DEPLOYMENT is not set."); + +// Get DTS connection string from environment variable +string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING") + ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"; + +// Use Azure Key Credential if provided, otherwise use Azure CLI Credential. +string? azureOpenAiKey = Environment.GetEnvironmentVariable("AZURE_OPENAI_KEY"); +AzureOpenAIClient client = !string.IsNullOrEmpty(azureOpenAiKey) + ? new AzureOpenAIClient(new Uri(endpoint), new AzureKeyCredential(azureOpenAiKey)) + : new AzureOpenAIClient(new Uri(endpoint), new AzureCliCredential()); + +// Single agent used by the orchestration to demonstrate human-in-the-loop workflow. +const string WriterName = "WriterAgent"; +const string WriterInstructions = + """ + You are a professional content writer who creates high-quality articles on various topics. + You write engaging, informative, and well-structured content that follows best practices for readability and accuracy. + """; + +AIAgent writerAgent = client.GetChatClient(deploymentName).CreateAIAgent(WriterInstructions, WriterName); + +// Orchestrator function +static async Task RunOrchestratorAsync(TaskOrchestrationContext context, ContentGenerationInput input) +{ + // Get the writer agent + DurableAIAgent writerAgent = context.GetAgent("WriterAgent"); + AgentThread writerThread = writerAgent.GetNewThread(); + + // Set initial status + context.SetCustomStatus($"Starting content generation for topic: {input.Topic}"); + + // Step 1: Generate initial content + AgentRunResponse writerResponse = await writerAgent.RunAsync( + message: $"Write a short article about '{input.Topic}' in less than 300 words.", + thread: writerThread); + GeneratedContent content = writerResponse.Result; + + // Human-in-the-loop iteration - we set a maximum number of attempts to avoid infinite loops + int iterationCount = 0; + while (iterationCount++ < input.MaxReviewAttempts) + { + context.SetCustomStatus( + $"Requesting human feedback. Iteration #{iterationCount}. Timeout: {input.ApprovalTimeoutHours} hour(s)."); + + // Step 2: Notify user to review the content + await context.CallActivityAsync(nameof(NotifyUserForApproval), content); + + // Step 3: Wait for human feedback with configurable timeout + HumanApprovalResponse humanResponse; + try + { + humanResponse = await context.WaitForExternalEvent( + eventName: "HumanApproval", + timeout: TimeSpan.FromHours(input.ApprovalTimeoutHours)); + } + catch (OperationCanceledException) + { + // Timeout occurred - treat as rejection + context.SetCustomStatus( + $"Human approval timed out after {input.ApprovalTimeoutHours} hour(s). Treating as rejection."); + throw new TimeoutException($"Human approval timed out after {input.ApprovalTimeoutHours} hour(s)."); + } + + if (humanResponse.Approved) + { + context.SetCustomStatus("Content approved by human reviewer. Publishing content..."); + + // Step 4: Publish the approved content + await context.CallActivityAsync(nameof(PublishContent), content); + + context.SetCustomStatus($"Content published successfully at {context.CurrentUtcDateTime:s}"); + return new { content = content.Content }; + } + + context.SetCustomStatus("Content rejected by human reviewer. Incorporating feedback and regenerating..."); + + // Incorporate human feedback and regenerate + writerResponse = await writerAgent.RunAsync( + message: $""" + The content was rejected by a human reviewer. Please rewrite the article incorporating their feedback. + + Human Feedback: {humanResponse.Feedback} + """, + thread: writerThread); + + content = writerResponse.Result; + } + + // If we reach here, it means we exhausted the maximum number of iterations + throw new InvalidOperationException( + $"Content could not be approved after {input.MaxReviewAttempts} iterations."); +} + +// Activity functions +static void NotifyUserForApproval(TaskActivityContext context, GeneratedContent content) +{ + // In a real implementation, this would send notifications via email, SMS, etc. + Console.WriteLine( + $""" + NOTIFICATION: Please review the following content for approval: + Title: {content.Title} + Content: {content.Content} + Use the approval endpoint to approve or reject this content. + """); +} + +static void PublishContent(TaskActivityContext context, GeneratedContent content) +{ + // In a real implementation, this would publish to a CMS, website, etc. + Console.WriteLine( + $""" + PUBLISHING: Content has been published successfully. + Title: {content.Title} + Content: {content.Content} + """); +} + +// Configure the console app to host the AI agent. +IHost host = Host.CreateDefaultBuilder(args) + .ConfigureLogging(loggingBuilder => loggingBuilder.SetMinimumLevel(LogLevel.Warning)) + .ConfigureServices(services => + { + services.ConfigureDurableAgents( + options => options.AddAIAgent(writerAgent), + workerBuilder: builder => + { + builder.UseDurableTaskScheduler(dtsConnectionString); + builder.AddTasks(registry => + { + registry.AddOrchestratorFunc(nameof(RunOrchestratorAsync), RunOrchestratorAsync); + registry.AddActivityFunc(nameof(NotifyUserForApproval), NotifyUserForApproval); + registry.AddActivityFunc(nameof(PublishContent), PublishContent); + }); + }, + clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString)); + }) + .Build(); + +await host.StartAsync(); + +DurableTaskClient durableTaskClient = host.Services.GetRequiredService(); + +// Console colors for better UX +Console.ForegroundColor = ConsoleColor.Cyan; +Console.WriteLine("=== Human-in-the-Loop Orchestration Sample ==="); +Console.ResetColor(); +Console.WriteLine("Enter topic for content generation:"); +Console.WriteLine(); + +// Read topic from stdin +string? topic = Console.ReadLine(); +if (string.IsNullOrWhiteSpace(topic)) +{ + Console.ForegroundColor = ConsoleColor.Red; + Console.Error.WriteLine("Error: Topic is required."); + Console.ResetColor(); + Environment.Exit(1); + return; +} + +// Prompt for optional parameters with defaults +Console.WriteLine(); +Console.WriteLine("Max review attempts (default: 3):"); +string? maxAttemptsInput = Console.ReadLine(); +int maxReviewAttempts = int.TryParse(maxAttemptsInput, out int maxAttempts) && maxAttempts > 0 + ? maxAttempts + : 3; + +Console.WriteLine("Approval timeout in hours (default: 72):"); +string? timeoutInput = Console.ReadLine(); +float approvalTimeoutHours = float.TryParse(timeoutInput, out float timeout) && timeout > 0 + ? timeout + : 72; + +ContentGenerationInput input = new() +{ + Topic = topic, + MaxReviewAttempts = maxReviewAttempts, + ApprovalTimeoutHours = approvalTimeoutHours +}; + +Console.WriteLine(); +Console.ForegroundColor = ConsoleColor.Gray; +Console.WriteLine("Starting orchestration..."); +Console.ResetColor(); + +try +{ + // Start the orchestration + string instanceId = await durableTaskClient.ScheduleNewOrchestrationInstanceAsync( + orchestratorName: nameof(RunOrchestratorAsync), + input: input); + + Console.ForegroundColor = ConsoleColor.Gray; + Console.WriteLine($"Orchestration started with instance ID: {instanceId}"); + Console.WriteLine("Waiting for human approval..."); + Console.ResetColor(); + Console.WriteLine(); + + // Monitor orchestration status and handle approval prompts + using CancellationTokenSource cts = new(); + Task orchestrationTask = Task.Run(async () => + { + while (!cts.Token.IsCancellationRequested) + { + OrchestrationMetadata? status = await durableTaskClient.GetInstanceAsync( + instanceId, + getInputsAndOutputs: true, + cts.Token); + + if (status == null) + { + await Task.Delay(TimeSpan.FromSeconds(1), cts.Token); + continue; + } + + // Check if we're waiting for approval + if (status.SerializedCustomStatus != null) + { + string? customStatus = status.ReadCustomStatusAs(); + if (customStatus?.StartsWith("Requesting human feedback", StringComparison.OrdinalIgnoreCase) == true) + { + // Prompt user for approval + Console.ForegroundColor = ConsoleColor.Yellow; + Console.WriteLine("Content is ready for review. Check the logs above for details."); + Console.Write("Approve? (y/n): "); + Console.ResetColor(); + + string? approvalInput = Console.ReadLine(); + bool approved = approvalInput?.Trim().Equals("y", StringComparison.OrdinalIgnoreCase) == true; + + Console.Write("Feedback (optional): "); + string? feedback = Console.ReadLine() ?? ""; + + HumanApprovalResponse approvalResponse = new() + { + Approved = approved, + Feedback = feedback + }; + + await durableTaskClient.RaiseEventAsync(instanceId, "HumanApproval", approvalResponse); + } + } + + if (status.RuntimeStatus is OrchestrationRuntimeStatus.Completed or OrchestrationRuntimeStatus.Failed or OrchestrationRuntimeStatus.Terminated) + { + break; + } + + await Task.Delay(TimeSpan.FromSeconds(1), cts.Token); + } + }, cts.Token); + + // Wait for orchestration to complete + OrchestrationMetadata finalStatus = await durableTaskClient.WaitForInstanceCompletionAsync( + instanceId, + getInputsAndOutputs: true, + CancellationToken.None); + + cts.Cancel(); + await orchestrationTask; + + Console.WriteLine(); + + if (finalStatus.RuntimeStatus == OrchestrationRuntimeStatus.Completed) + { + Console.ForegroundColor = ConsoleColor.Green; + Console.WriteLine("✓ Orchestration completed successfully!"); + Console.ResetColor(); + Console.WriteLine(); + + JsonElement output = finalStatus.ReadOutputAs(); + if (output.TryGetProperty("content", out JsonElement contentElement)) + { + Console.ForegroundColor = ConsoleColor.Yellow; + Console.WriteLine("Published content:"); + Console.ResetColor(); + Console.WriteLine(contentElement.GetString()); + } + } + else if (finalStatus.RuntimeStatus == OrchestrationRuntimeStatus.Failed) + { + Console.ForegroundColor = ConsoleColor.Red; + Console.WriteLine("✗ Orchestration failed!"); + Console.ResetColor(); + if (finalStatus.FailureDetails != null) + { + Console.WriteLine($"Error: {finalStatus.FailureDetails.ErrorMessage}"); + } + Environment.Exit(1); + } + else + { + Console.ForegroundColor = ConsoleColor.Yellow; + Console.WriteLine($"Orchestration status: {finalStatus.RuntimeStatus}"); + Console.ResetColor(); + } +} +catch (Exception ex) +{ + Console.ForegroundColor = ConsoleColor.Red; + Console.Error.WriteLine($"Error: {ex.Message}"); + Console.ResetColor(); + Environment.Exit(1); +} +finally +{ + await host.StopAsync(); +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/05_AgentOrchestration_HITL/README.md b/dotnet/samples/DurableAgents/ConsoleApps/05_AgentOrchestration_HITL/README.md new file mode 100644 index 0000000000..1386dfbcb1 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/05_AgentOrchestration_HITL/README.md @@ -0,0 +1,73 @@ +# Human-in-the-Loop Orchestration Sample + +This sample demonstrates how to use the durable agents extension to create a console app that implements a human-in-the-loop workflow using durable orchestration, including interactive approval prompts. + +## Key Concepts Demonstrated + +- Human-in-the-loop workflows with durable orchestration +- External event handling for human approval/rejection +- Timeout handling for approval requests +- Iterative content refinement based on human feedback + +## Environment Setup + +See the [README.md](../README.md) file in the parent directory for more information on how to configure the environment, including how to install and run common sample dependencies. + +## Running the Sample + +With the environment setup, you can run the sample: + +```bash +cd dotnet/samples/DurableAgents/ConsoleApps/05_AgentOrchestration_HITL +dotnet run --framework net10.0 +``` + +The app will prompt you for input: + +```text +=== Human-in-the-Loop Orchestration Sample === +Enter topic for content generation: + +The Future of Artificial Intelligence + +Max review attempts (default: 3): +3 +Approval timeout in hours (default: 72): +72 +``` + +The orchestration will generate content and prompt you for approval: + +```text +Orchestration started with instance ID: 86313f1d45fb42eeb50b1852626bf3ff + +=== NOTIFICATION: Content Ready for Review === +Title: The Future of Artificial Intelligence + +Content: +[Generated content appears here] + +Please review the content above and provide your approval. + +Content is ready for review. Check the logs above for details. +Approve? (y/n): n +Feedback (optional): Please add more details about the ethical implications. +``` + +The orchestration will incorporate your feedback and regenerate the content. Once approved, it will publish and complete. + +## Viewing Orchestration State + +You can view the state of the orchestration in the Durable Task Scheduler dashboard: + +1. Open your browser and navigate to `http://localhost:8082` +2. In the dashboard, you can see: + - **Orchestrations**: View the orchestration instance, including its runtime status, custom status (which shows approval state), input, output, and execution history + - **Agents**: View the state of the WriterAgent, including conversation history + +The orchestration instance ID is displayed in the console output. You can use this ID to find the specific orchestration in the dashboard and inspect: + +- The custom status field, which shows the current state of the approval workflow +- When the orchestration is waiting for external events +- The iteration count and feedback history +- The final published content diff --git a/dotnet/samples/DurableAgents/ConsoleApps/06_LongRunningTools/06_LongRunningTools.csproj b/dotnet/samples/DurableAgents/ConsoleApps/06_LongRunningTools/06_LongRunningTools.csproj new file mode 100644 index 0000000000..d7557dbdfc --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/06_LongRunningTools/06_LongRunningTools.csproj @@ -0,0 +1,30 @@ + + + net10.0 + Exe + enable + enable + LongRunningTools + LongRunningTools + + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/DurableAgents/ConsoleApps/06_LongRunningTools/Models.cs b/dotnet/samples/DurableAgents/ConsoleApps/06_LongRunningTools/Models.cs new file mode 100644 index 0000000000..43ab9d99f8 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/06_LongRunningTools/Models.cs @@ -0,0 +1,44 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Text.Json.Serialization; + +namespace LongRunningTools; + +/// +/// Represents the input for the content generation workflow. +/// +public sealed class ContentGenerationInput +{ + [JsonPropertyName("topic")] + public string Topic { get; set; } = string.Empty; + + [JsonPropertyName("max_review_attempts")] + public int MaxReviewAttempts { get; set; } = 3; + + [JsonPropertyName("approval_timeout_hours")] + public float ApprovalTimeoutHours { get; set; } = 72; +} + +/// +/// Represents the content generated by the writer agent. +/// +public sealed class GeneratedContent +{ + [JsonPropertyName("title")] + public string Title { get; set; } = string.Empty; + + [JsonPropertyName("content")] + public string Content { get; set; } = string.Empty; +} + +/// +/// Represents the human feedback response. +/// +public sealed class HumanFeedbackResponse +{ + [JsonPropertyName("approved")] + public bool Approved { get; set; } + + [JsonPropertyName("feedback")] + public string Feedback { get; set; } = string.Empty; +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/06_LongRunningTools/Program.cs b/dotnet/samples/DurableAgents/ConsoleApps/06_LongRunningTools/Program.cs new file mode 100644 index 0000000000..0e83d56d8f --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/06_LongRunningTools/Program.cs @@ -0,0 +1,351 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.ComponentModel; +using Azure; +using Azure.AI.OpenAI; +using Azure.Identity; +using LongRunningTools; +using Microsoft.Agents.AI; +using Microsoft.Agents.AI.DurableTask; +using Microsoft.DurableTask; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Client.AzureManaged; +using Microsoft.DurableTask.Worker; +using Microsoft.DurableTask.Worker.AzureManaged; +using Microsoft.Extensions.AI; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using OpenAI.Chat; + +// Get the Azure OpenAI endpoint and deployment name from environment variables. +string endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") + ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set."); +string deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT") + ?? throw new InvalidOperationException("AZURE_OPENAI_DEPLOYMENT is not set."); + +// Get DTS connection string from environment variable +string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING") + ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"; + +// Use Azure Key Credential if provided, otherwise use Azure CLI Credential. +string? azureOpenAiKey = Environment.GetEnvironmentVariable("AZURE_OPENAI_KEY"); +AzureOpenAIClient client = !string.IsNullOrEmpty(azureOpenAiKey) + ? new AzureOpenAIClient(new Uri(endpoint), new AzureKeyCredential(azureOpenAiKey)) + : new AzureOpenAIClient(new Uri(endpoint), new AzureCliCredential()); + +// Agent used by the orchestration to write content. +const string WriterAgentName = "Writer"; +const string WriterAgentInstructions = + """ + You are a professional content writer who creates high-quality articles on various topics. + You write engaging, informative, and well-structured content that follows best practices for readability and accuracy. + """; + +AIAgent writerAgent = client.GetChatClient(deploymentName).CreateAIAgent(WriterAgentInstructions, WriterAgentName); + +// Agent that can start content generation workflows using tools +const string PublisherAgentName = "Publisher"; +const string PublisherAgentInstructions = + """ + You are a publishing agent that can manage content generation workflows. + You have access to tools to start, monitor, and raise events for content generation workflows. + """; + +const string HumanFeedbackEventName = "HumanFeedback"; + +// Orchestrator function +static async Task RunOrchestratorAsync(TaskOrchestrationContext context, ContentGenerationInput input) +{ + // Get the writer agent + DurableAIAgent writerAgent = context.GetAgent(WriterAgentName); + AgentThread writerThread = writerAgent.GetNewThread(); + + // Set initial status + context.SetCustomStatus($"Starting content generation for topic: {input.Topic}"); + + // Step 1: Generate initial content + AgentRunResponse writerResponse = await writerAgent.RunAsync( + message: $"Write a short article about '{input.Topic}'.", + thread: writerThread); + GeneratedContent content = writerResponse.Result; + + // Human-in-the-loop iteration - we set a maximum number of attempts to avoid infinite loops + int iterationCount = 0; + while (iterationCount++ < input.MaxReviewAttempts) + { + context.SetCustomStatus( + new + { + message = "Requesting human feedback.", + approvalTimeoutHours = input.ApprovalTimeoutHours, + iterationCount, + content + }); + + // Step 2: Notify user to review the content + await context.CallActivityAsync(nameof(NotifyUserForApproval), content); + + // Step 3: Wait for human feedback with configurable timeout + HumanFeedbackResponse humanResponse; + try + { + humanResponse = await context.WaitForExternalEvent( + eventName: HumanFeedbackEventName, + timeout: TimeSpan.FromHours(input.ApprovalTimeoutHours)); + } + catch (OperationCanceledException) + { + // Timeout occurred - treat as rejection + context.SetCustomStatus( + new + { + message = $"Human approval timed out after {input.ApprovalTimeoutHours} hour(s). Treating as rejection.", + iterationCount, + content + }); + throw new TimeoutException($"Human approval timed out after {input.ApprovalTimeoutHours} hour(s)."); + } + + if (humanResponse.Approved) + { + context.SetCustomStatus(new + { + message = "Content approved by human reviewer. Publishing content...", + content + }); + + // Step 4: Publish the approved content + await context.CallActivityAsync(nameof(PublishContent), content); + + context.SetCustomStatus(new + { + message = $"Content published successfully at {context.CurrentUtcDateTime:s}", + humanFeedback = humanResponse, + content + }); + return new { content = content.Content }; + } + + context.SetCustomStatus(new + { + message = "Content rejected by human reviewer. Incorporating feedback and regenerating...", + humanFeedback = humanResponse, + content + }); + + // Incorporate human feedback and regenerate + writerResponse = await writerAgent.RunAsync( + message: $""" + The content was rejected by a human reviewer. Please rewrite the article incorporating their feedback. + + Human Feedback: {humanResponse.Feedback} + """, + thread: writerThread); + + content = writerResponse.Result; + } + + // If we reach here, it means we exhausted the maximum number of iterations + throw new InvalidOperationException( + $"Content could not be approved after {input.MaxReviewAttempts} iterations."); +} + +// Activity functions +static void NotifyUserForApproval(TaskActivityContext context, GeneratedContent content) +{ + // In a real implementation, this would send notifications via email, SMS, etc. + Console.ForegroundColor = ConsoleColor.DarkMagenta; + Console.WriteLine( + $""" + NOTIFICATION: Please review the following content for approval: + Title: {content.Title} + Content: {content.Content} + """); + Console.ResetColor(); +} + +static void PublishContent(TaskActivityContext context, GeneratedContent content) +{ + // In a real implementation, this would publish to a CMS, website, etc. + Console.ForegroundColor = ConsoleColor.DarkMagenta; + Console.WriteLine( + $""" + PUBLISHING: Content has been published successfully. + Title: {content.Title} + Content: {content.Content} + """); + Console.ResetColor(); +} + +// Tools that demonstrate starting orchestrations from agent tool calls. +[Description("Starts a content generation workflow and returns the instance ID for tracking.")] +static string StartContentGenerationWorkflow([Description("The topic for content generation")] string topic) +{ + const int MaxReviewAttempts = 3; + const float ApprovalTimeoutHours = 72; + + // Schedule the orchestration, which will start running after the tool call completes. + string instanceId = DurableAgentContext.Current.ScheduleNewOrchestration( + name: nameof(RunOrchestratorAsync), + input: new ContentGenerationInput + { + Topic = topic, + MaxReviewAttempts = MaxReviewAttempts, + ApprovalTimeoutHours = ApprovalTimeoutHours + }); + + return $"Workflow started with instance ID: {instanceId}"; +} + +[Description("Gets the status of a workflow orchestration and returns a summary of the workflow's current status.")] +static async Task GetWorkflowStatusAsync( + [Description("The instance ID of the workflow to check")] string instanceId, + [Description("Whether to include detailed information")] bool includeDetails = true) +{ + // Get the current agent context using the thread-static property + OrchestrationMetadata? status = await DurableAgentContext.Current.GetOrchestrationStatusAsync( + instanceId, + includeDetails); + + if (status is null) + { + return new + { + instanceId, + error = $"Workflow instance '{instanceId}' not found.", + }; + } + + return new + { + instanceId = status.InstanceId, + createdAt = status.CreatedAt, + executionStatus = status.RuntimeStatus, + workflowStatus = status.SerializedCustomStatus, + lastUpdatedAt = status.LastUpdatedAt, + failureDetails = status.FailureDetails + }; +} + +[Description( + "Raises a feedback event for the content generation workflow. If approved, the workflow will be published. " + + "If rejected, the workflow will generate new content.")] +static async Task SubmitHumanFeedbackAsync( + [Description("The instance ID of the workflow to submit feedback for")] string instanceId, + [Description("Feedback to submit")] HumanFeedbackResponse feedback) +{ + await DurableAgentContext.Current.RaiseOrchestrationEventAsync(instanceId, HumanFeedbackEventName, feedback); +} + +// Configure the console app to host the AI agents. +IHost host = Host.CreateDefaultBuilder(args) + .ConfigureLogging(loggingBuilder => loggingBuilder.SetMinimumLevel(LogLevel.Warning)) + .ConfigureServices(services => + { + services.ConfigureDurableAgents( + options => + { + // Add the writer agent used by the orchestration + options.AddAIAgent(writerAgent); + + // Define the agent that can start orchestrations from tool calls + options.AddAIAgentFactory(PublisherAgentName, sp => + { + return client.GetChatClient(deploymentName).CreateAIAgent( + instructions: PublisherAgentInstructions, + name: PublisherAgentName, + services: sp, + tools: [ + AIFunctionFactory.Create(StartContentGenerationWorkflow), + AIFunctionFactory.Create(GetWorkflowStatusAsync), + AIFunctionFactory.Create(SubmitHumanFeedbackAsync), + ]); + }); + }, + workerBuilder: builder => + { + builder.UseDurableTaskScheduler(dtsConnectionString); + builder.AddTasks(registry => + { + registry.AddOrchestratorFunc(nameof(RunOrchestratorAsync), RunOrchestratorAsync); + registry.AddActivityFunc(nameof(NotifyUserForApproval), NotifyUserForApproval); + registry.AddActivityFunc(nameof(PublishContent), PublishContent); + }); + }, + clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString)); + }) + .Build(); + +await host.StartAsync(); + +// Get the agent proxy from services +IServiceProvider services = host.Services; +AIAgent? agentProxy = services.GetKeyedService(PublisherAgentName); +if (agentProxy == null) +{ + Console.ForegroundColor = ConsoleColor.Red; + Console.Error.WriteLine("Agent 'Publisher' not found."); + Console.ResetColor(); + Environment.Exit(1); + return; +} + +// Console colors for better UX +Console.ForegroundColor = ConsoleColor.Cyan; +Console.WriteLine("=== Long Running Tools Sample ==="); +Console.ResetColor(); +Console.WriteLine("Enter a topic for the Publisher agent to write about (or 'exit' to quit):"); +Console.WriteLine(); + +// Create a thread for the conversation +AgentThread thread = agentProxy.GetNewThread(); + +using CancellationTokenSource cts = new(); +Console.CancelKeyPress += (sender, e) => +{ + e.Cancel = true; + cts.Cancel(); +}; + +while (!cts.Token.IsCancellationRequested) +{ + // Read input from stdin + Console.ForegroundColor = ConsoleColor.Yellow; + Console.Write("You: "); + Console.ResetColor(); + + string? input = Console.ReadLine(); + if (string.IsNullOrWhiteSpace(input) || input.Equals("exit", StringComparison.OrdinalIgnoreCase)) + { + break; + } + + // Run the agent + Console.ForegroundColor = ConsoleColor.Green; + Console.Write("Publisher: "); + Console.ResetColor(); + + try + { + AgentRunResponse agentResponse = await agentProxy.RunAsync( + message: input, + thread: thread, + cancellationToken: cts.Token); + + Console.WriteLine(agentResponse.Text); + Console.WriteLine(); + } + catch (Exception ex) + { + Console.ForegroundColor = ConsoleColor.Red; + Console.Error.WriteLine($"Error: {ex.Message}"); + Console.ResetColor(); + Console.WriteLine(); + } + + Console.WriteLine("(Press Enter to prompt the Publisher agent again)"); + _ = Console.ReadLine(); +} + +await host.StopAsync(); diff --git a/dotnet/samples/DurableAgents/ConsoleApps/06_LongRunningTools/README.md b/dotnet/samples/DurableAgents/ConsoleApps/06_LongRunningTools/README.md new file mode 100644 index 0000000000..b0dd69b129 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/06_LongRunningTools/README.md @@ -0,0 +1,90 @@ +# Long Running Tools Sample + +This sample demonstrates how to use the durable agents extension to create a console app with agents that have long running tools. This sample builds on the [05_AgentOrchestration_HITL](../05_AgentOrchestration_HITL) sample by adding a publisher agent that can start and manage content generation workflows. A key difference is that the publisher agent knows the IDs of the workflows it starts, so it can check the status of the workflows and approve or reject them without being explicitly given the context (instance IDs, etc). + +## Key Concepts Demonstrated + +The same key concepts as the [05_AgentOrchestration_HITL](../05_AgentOrchestration_HITL) sample are demonstrated, but with the following additional concepts: + +- **Long running tools**: Using `DurableAgentContext.Current` to start orchestrations from tool calls +- **Multi-agent orchestration**: Agents can start and manage workflows that orchestrate other agents +- **Human-in-the-loop (with delegation)**: The agent acts as an intermediary between the human and the workflow. The human remains in the loop, but delegates to the agent to start the workflow and approve or reject the content. + +## Environment Setup + +See the [README.md](../README.md) file in the parent directory for more information on how to configure the environment, including how to install and run common sample dependencies. + +## Running the Sample + +With the environment setup, you can run the sample: + +```bash +cd dotnet/samples/DurableAgents/ConsoleApps/06_LongRunningTools +dotnet run --framework net10.0 +``` + +The app will prompt you for input. You can interact with the Publisher agent: + +```text +=== Long Running Tools Sample === +Enter a topic for the Publisher agent to write about (or 'exit' to quit): + +You: Start a content generation workflow for the topic 'The Future of Artificial Intelligence' +Publisher: The content generation workflow for the topic "The Future of Artificial Intelligence" has been successfully started, and the instance ID is **6a04276e8d824d8d941e1dc4142cc254**. If you need any further assistance or updates on the workflow, feel free to ask! +``` + +Behind the scenes, the publisher agent will: + +1. Start the content generation workflow via a tool call +2. The workflow will generate initial content using the Writer agent and wait for human approval, which will be visible in the terminal + +Once the workflow is waiting for human approval, you can send approval or rejection by prompting the publisher agent accordingly. + +> [!NOTE] +> You must press Enter after each message to continue the conversation. The sample is set up this way because the workflow is running in the background and may write to the console asynchronously. + +To tell the agent to rewrite the content with feedback, you can prompt it to reject the content with feedback. + +```text +You: Reject the content with feedback: The article needs more technical depth and better examples. +Publisher: The content has been successfully rejected with the feedback: "The article needs more technical depth and better examples." The workflow will now generate new content based on this feedback. +``` + +Once you're satisfied with the content, you can approve it for publishing. + +```text +You: Approve the content +Publisher: The content has been successfully approved for publishing. If you need any more assistance or have further requests, feel free to let me know! +``` + +Once the workflow has completed, you can get the status by prompting the publisher agent to give you the status. + +```text +You: Get the status of the workflow you previously started +Publisher: The status of the workflow with instance ID **6a04276e8d824d8d941e1dc4142cc254** is as follows: + +- **Execution Status:** Completed +- **Created At:** December 22, 2025, 23:08:13 UTC +- **Last Updated At:** December 22, 2025, 23:09:59 UTC +- **Workflow Status:** + - Message: Content published successfully at December 22, 2025, 23:09:59 UTC + - Human Feedback: Approved +``` + +## Viewing Agent and Orchestration State + +You can view the state of both the agent and the orchestrations it starts in the Durable Task Scheduler dashboard: + +1. Open your browser and navigate to `http://localhost:8082` +2. In the dashboard, you can see: + - **Agents**: View the state of the Publisher agent, including its conversation history and tool call history + - **Orchestrations**: View the content generation orchestration instances that were started by the agent via tool calls, including their runtime status, custom status, input, output, and execution history + +When the publisher agent starts a workflow, the orchestration instance ID is included in the agent's response. You can use this ID to find the specific orchestration in the dashboard and inspect: + +- The orchestration's execution progress +- When it's waiting for human approval (visible in custom status) +- The content generation workflow state +- The WriterAgent state within the orchestration + +This demonstrates how agents can manage long-running workflows and how you can monitor both the agent's state and the workflows it orchestrates. diff --git a/dotnet/samples/DurableAgents/ConsoleApps/07_ReliableStreaming/07_ReliableStreaming.csproj b/dotnet/samples/DurableAgents/ConsoleApps/07_ReliableStreaming/07_ReliableStreaming.csproj new file mode 100644 index 0000000000..09c6a8cdd3 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/07_ReliableStreaming/07_ReliableStreaming.csproj @@ -0,0 +1,31 @@ + + + net10.0 + Exe + enable + enable + ReliableStreaming + ReliableStreaming + + + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/DurableAgents/ConsoleApps/07_ReliableStreaming/Program.cs b/dotnet/samples/DurableAgents/ConsoleApps/07_ReliableStreaming/Program.cs new file mode 100644 index 0000000000..3dac9604b3 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/07_ReliableStreaming/Program.cs @@ -0,0 +1,352 @@ +// Copyright (c) Microsoft. All rights reserved. + +// This sample demonstrates how to implement reliable streaming for durable agents using Redis Streams. +// It reads prompts from stdin and streams agent responses to stdout in real-time. + +using System.ComponentModel; +using Azure; +using Azure.AI.OpenAI; +using Azure.Identity; +using Microsoft.Agents.AI; +using Microsoft.Agents.AI.DurableTask; +using Microsoft.DurableTask.Client.AzureManaged; +using Microsoft.DurableTask.Worker.AzureManaged; +using Microsoft.Extensions.AI; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using OpenAI.Chat; +using ReliableStreaming; +using StackExchange.Redis; + +// Get the Azure OpenAI endpoint and deployment name from environment variables. +string endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") + ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set."); +string deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT") + ?? throw new InvalidOperationException("AZURE_OPENAI_DEPLOYMENT is not set."); + +// Get Redis connection string from environment variable. +string redisConnectionString = Environment.GetEnvironmentVariable("REDIS_CONNECTION_STRING") + ?? "localhost:6379"; + +// Get the Redis stream TTL from environment variable (default: 10 minutes). +int redisStreamTtlMinutes = int.Parse(Environment.GetEnvironmentVariable("REDIS_STREAM_TTL_MINUTES") ?? "10"); + +// Get DTS connection string from environment variable +string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING") + ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"; + +// Use Azure Key Credential if provided, otherwise use Azure CLI Credential. +string? azureOpenAiKey = Environment.GetEnvironmentVariable("AZURE_OPENAI_KEY"); +AzureOpenAIClient client = !string.IsNullOrEmpty(azureOpenAiKey) + ? new AzureOpenAIClient(new Uri(endpoint), new AzureKeyCredential(azureOpenAiKey)) + : new AzureOpenAIClient(new Uri(endpoint), new AzureCliCredential()); + +// Travel Planner agent instructions - designed to produce longer responses for demonstrating streaming. +const string TravelPlannerName = "TravelPlanner"; +const string TravelPlannerInstructions = + """ + You are an expert travel planner who creates detailed, personalized travel itineraries. + When asked to plan a trip, you should: + 1. Create a comprehensive day-by-day itinerary + 2. Include specific recommendations for activities, restaurants, and attractions + 3. Provide practical tips for each destination + 4. Consider weather and local events when making recommendations + 5. Include estimated times and logistics between activities + + Always use the available tools to get current weather forecasts and local events + for the destination to make your recommendations more relevant and timely. + + Format your response with clear headings for each day and include emoji icons + to make the itinerary easy to scan and visually appealing. + """; + +// Mock travel tools that return hardcoded data for demonstration purposes. +[Description("Gets the weather forecast for a destination on a specific date. Use this to provide weather-aware recommendations in the itinerary.")] +static string GetWeatherForecast(string destination, string date) +{ + Dictionary weatherByRegion = new(StringComparer.OrdinalIgnoreCase) + { + ["Tokyo"] = ("Partly cloudy with a chance of light rain", 58, 45), + ["Paris"] = ("Overcast with occasional drizzle", 52, 41), + ["New York"] = ("Clear and cold", 42, 28), + ["London"] = ("Foggy morning, clearing in afternoon", 48, 38), + ["Sydney"] = ("Sunny and warm", 82, 68), + ["Rome"] = ("Sunny with light breeze", 62, 48), + ["Barcelona"] = ("Partly sunny", 59, 47), + ["Amsterdam"] = ("Cloudy with light rain", 46, 38), + ["Dubai"] = ("Sunny and hot", 85, 72), + ["Singapore"] = ("Tropical thunderstorms in afternoon", 88, 77), + ["Bangkok"] = ("Hot and humid, afternoon showers", 91, 78), + ["Los Angeles"] = ("Sunny and pleasant", 72, 55), + ["San Francisco"] = ("Morning fog, afternoon sun", 62, 52), + ["Seattle"] = ("Rainy with breaks", 48, 40), + ["Miami"] = ("Warm and sunny", 78, 65), + ["Honolulu"] = ("Tropical paradise weather", 82, 72), + }; + + (string condition, int highF, int lowF) forecast = ("Partly cloudy", 65, 50); + foreach (KeyValuePair entry in weatherByRegion) + { + if (destination.Contains(entry.Key, StringComparison.OrdinalIgnoreCase)) + { + forecast = entry.Value; + break; + } + } + + return $""" + Weather forecast for {destination} on {date}: + Conditions: {forecast.condition} + High: {forecast.highF}°F ({(forecast.highF - 32) * 5 / 9}°C) + Low: {forecast.lowF}°F ({(forecast.lowF - 32) * 5 / 9}°C) + + Recommendation: {GetWeatherRecommendation(forecast.condition)} + """; +} + +[Description("Gets local events and activities happening at a destination around a specific date. Use this to suggest timely activities and experiences.")] +static string GetLocalEvents(string destination, string date) +{ + Dictionary eventsByCity = new(StringComparer.OrdinalIgnoreCase) + { + ["Tokyo"] = [ + "🎭 Kabuki Theater Performance at Kabukiza Theatre - Traditional Japanese drama", + "🌸 Winter Illuminations at Yoyogi Park - Spectacular light displays", + "🍜 Ramen Festival at Tokyo Station - Sample ramen from across Japan", + "🎮 Gaming Expo at Tokyo Big Sight - Latest video games and technology", + ], + ["Paris"] = [ + "🎨 Impressionist Exhibition at Musée d'Orsay - Extended evening hours", + "🍷 Wine Tasting Tour in Le Marais - Local sommelier guided", + "🎵 Jazz Night at Le Caveau de la Huchette - Historic jazz club", + "🥐 French Pastry Workshop - Learn from master pâtissiers", + ], + ["New York"] = [ + "🎭 Broadway Show: Hamilton - Limited engagement performances", + "🏀 Knicks vs Lakers at Madison Square Garden", + "🎨 Modern Art Exhibit at MoMA - New installations", + "🍕 Pizza Walking Tour of Brooklyn - Artisan pizzerias", + ], + ["London"] = [ + "👑 Royal Collection Exhibition at Buckingham Palace", + "🎭 West End Musical: The Phantom of the Opera", + "🍺 Craft Beer Festival at Brick Lane", + "🎪 Winter Wonderland at Hyde Park - Rides and markets", + ], + ["Sydney"] = [ + "🏄 Pro Surfing Competition at Bondi Beach", + "🎵 Opera at Sydney Opera House - La Bohème", + "🦘 Wildlife Night Safari at Taronga Zoo", + "🍽️ Harbor Dinner Cruise with fireworks", + ], + ["Rome"] = [ + "🏛️ After-Hours Vatican Tour - Skip the crowds", + "🍝 Pasta Making Class in Trastevere", + "🎵 Classical Concert at Borghese Gallery", + "🍷 Wine Tasting in Roman Cellars", + ], + }; + + string[] events = [ + "🎭 Local theater performance", + "🍽️ Food and wine festival", + "🎨 Art gallery opening", + "🎵 Live music at local venues", + ]; + + foreach (KeyValuePair entry in eventsByCity) + { + if (destination.Contains(entry.Key, StringComparison.OrdinalIgnoreCase)) + { + events = entry.Value; + break; + } + } + + string eventList = string.Join("\n• ", events); + return $""" + Local events in {destination} around {date}: + + • {eventList} + + 💡 Tip: Book popular events in advance as they may sell out quickly! + """; +} + +static string GetWeatherRecommendation(string condition) +{ + return condition switch + { + string c when c.Contains("rain", StringComparison.OrdinalIgnoreCase) || c.Contains("drizzle", StringComparison.OrdinalIgnoreCase) => + "Bring an umbrella and waterproof jacket. Consider indoor activities for backup.", + string c when c.Contains("fog", StringComparison.OrdinalIgnoreCase) => + "Morning visibility may be limited. Plan outdoor sightseeing for afternoon.", + string c when c.Contains("cold", StringComparison.OrdinalIgnoreCase) => + "Layer up with warm clothing. Hot drinks and cozy cafés recommended.", + string c when c.Contains("hot", StringComparison.OrdinalIgnoreCase) || c.Contains("warm", StringComparison.OrdinalIgnoreCase) => + "Stay hydrated and use sunscreen. Plan strenuous activities for cooler morning hours.", + string c when c.Contains("thunder", StringComparison.OrdinalIgnoreCase) || c.Contains("storm", StringComparison.OrdinalIgnoreCase) => + "Keep an eye on weather updates. Have indoor alternatives ready.", + _ => "Pleasant conditions expected. Great day for outdoor exploration!" + }; +} + +// Configure the console app to host the AI agent. +IHost host = Host.CreateDefaultBuilder(args) + .ConfigureLogging(loggingBuilder => loggingBuilder.SetMinimumLevel(LogLevel.Warning)) + .ConfigureServices(services => + { + services.ConfigureDurableAgents( + options => + { + // Define the Travel Planner agent with tools for weather and events + options.AddAIAgentFactory(TravelPlannerName, sp => + { + return client.GetChatClient(deploymentName).CreateAIAgent( + instructions: TravelPlannerInstructions, + name: TravelPlannerName, + services: sp, + tools: [ + AIFunctionFactory.Create(GetWeatherForecast), + AIFunctionFactory.Create(GetLocalEvents), + ]); + }); + }, + workerBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString), + clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString)); + + // Register Redis connection as a singleton + services.AddSingleton(_ => + ConnectionMultiplexer.Connect(redisConnectionString)); + + // Register the Redis stream response handler - this captures agent responses + // and publishes them to Redis Streams for reliable delivery. + services.AddSingleton(sp => + new RedisStreamResponseHandler( + sp.GetRequiredService(), + TimeSpan.FromMinutes(redisStreamTtlMinutes))); + services.AddSingleton(sp => + sp.GetRequiredService()); + }) + .Build(); + +await host.StartAsync(); + +// Get the agent proxy from services +IServiceProvider services = host.Services; +AIAgent? agentProxy = services.GetKeyedService(TravelPlannerName); +RedisStreamResponseHandler streamHandler = services.GetRequiredService(); + +if (agentProxy == null) +{ + Console.ForegroundColor = ConsoleColor.Red; + Console.Error.WriteLine($"Agent '{TravelPlannerName}' not found."); + Console.ResetColor(); + Environment.Exit(1); + return; +} + +// Console colors for better UX +Console.ForegroundColor = ConsoleColor.Cyan; +Console.WriteLine("=== Reliable Streaming Sample ==="); +Console.ResetColor(); +Console.WriteLine("Enter a travel planning request (or 'exit' to quit):"); +Console.WriteLine(); + +string? lastCursor = null; + +async Task ReadStreamTask(string conversationId, string? cursor, CancellationToken cancellationToken) +{ + await foreach (StreamChunk chunk in streamHandler.ReadStreamAsync(conversationId, cursor, cancellationToken)) + { + if (chunk.Error != null) + { + Console.ForegroundColor = ConsoleColor.Red; + Console.Error.WriteLine($"\n[Error: {chunk.Error}]"); + Console.ResetColor(); + break; + } + + if (chunk.IsDone) + { + Console.WriteLine(); + Console.WriteLine(); + break; + } + + if (chunk.Text != null) + { + Console.Write(chunk.Text); + } + + lastCursor = chunk.EntryId; + } +} + +// New conversation: prompt from stdin +Console.ForegroundColor = ConsoleColor.Yellow; +Console.Write("You: "); +Console.ResetColor(); + +string? prompt = Console.ReadLine(); +if (string.IsNullOrWhiteSpace(prompt) || prompt.Equals("exit", StringComparison.OrdinalIgnoreCase)) +{ + return; +} + +// Create a new agent thread +AgentThread thread = agentProxy.GetNewThread(); +AgentThreadMetadata? metadata = thread.GetService(); + +string conversationId = metadata?.ConversationId + ?? throw new InvalidOperationException("Failed to get conversation ID from thread metadata."); + +Console.ForegroundColor = ConsoleColor.Green; +Console.WriteLine($"Conversation ID: {conversationId}"); +Console.WriteLine("Press [Enter] to interrupt the stream."); +Console.ResetColor(); + +// Run the agent in the background +DurableAgentRunOptions options = new() { IsFireAndForget = true }; +await agentProxy.RunAsync(prompt, thread, options, CancellationToken.None); + +bool streamCompleted = false; +while (!streamCompleted) +{ + // On a key press, cancel the cancellation token to stop the stream + using CancellationTokenSource userCancellationSource = new(); + _ = Task.Run(() => + { + _ = Console.ReadLine(); + userCancellationSource.Cancel(); + }); + + try + { + // Start reading the stream and wait for it to complete + await ReadStreamTask(conversationId, lastCursor, userCancellationSource.Token); + streamCompleted = true; + } + catch (OperationCanceledException) + { + Console.ForegroundColor = ConsoleColor.Yellow; + Console.WriteLine("Stream cancelled. Press [Enter] to reconnect and resume the stream from the last cursor."); + Console.WriteLine($"Last cursor: {lastCursor ?? "(n/a)"}"); + Console.ResetColor(); + } + + if (!streamCompleted) + { + Console.ReadLine(); + Console.ForegroundColor = ConsoleColor.Green; + Console.WriteLine($"Resuming conversation: {conversationId} from cursor: {lastCursor ?? "(beginning)"}"); + Console.ResetColor(); + } +} + +Console.ForegroundColor = ConsoleColor.Green; +Console.WriteLine("Conversation completed."); +Console.ResetColor(); + +await host.StopAsync(); diff --git a/dotnet/samples/DurableAgents/ConsoleApps/07_ReliableStreaming/README.md b/dotnet/samples/DurableAgents/ConsoleApps/07_ReliableStreaming/README.md new file mode 100644 index 0000000000..c1956157e8 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/07_ReliableStreaming/README.md @@ -0,0 +1,181 @@ +# Reliable Streaming with Redis + +This sample demonstrates how to implement reliable streaming for durable agents using Redis Streams as a message broker. It enables clients to disconnect and reconnect to ongoing agent responses without losing messages, inspired by [OpenAI's background mode](https://platform.openai.com/docs/guides/background) for the Responses API. + +## Key Concepts Demonstrated + +- **Reliable message delivery**: Agent responses are persisted to Redis Streams, allowing clients to resume from any point +- **Real-time streaming**: Chunks are printed to stdout as they arrive (like `tail -f`) +- **Cursor-based resumption**: Each chunk includes an entry ID that can be used to resume the stream +- **Fire-and-forget agent invocation**: The agent runs in the background while the client streams from Redis + +## Environment Setup + +See the [README.md](../README.md) file in the parent directory for more information on how to configure the environment, including how to install and run common sample dependencies. + +### Additional Requirements: Redis + +This sample requires a Redis instance. Start a local Redis instance using Docker: + +```bash +docker run -d --name redis -p 6379:6379 redis:latest +``` + +To verify Redis is running: + +```bash +docker ps | grep redis +``` + +## Running the Sample + +With the environment setup, you can run the sample: + +```bash +cd dotnet/samples/DurableAgents/ConsoleApps/07_ReliableStreaming +dotnet run --framework net10.0 +``` + +The app will prompt you for a travel planning request: + +```text +=== Reliable Streaming Sample === +Enter a travel planning request (or 'exit' to quit): + +You: Plan a 7-day trip to Tokyo, Japan for next month. Include daily activities, restaurant recommendations, and tips for getting around. +``` + +The agent's response will stream to your console in real-time as chunks arrive from Redis: + +```text +Starting new conversation: @dafx-travelplanner@a1b2c3d4e5f67890abcdef1234567890 +Press [Enter] to interrupt the stream. + +TravelPlanner: # 7-Day Tokyo Adventure + +## Day 1: Arrival and Exploration +... +``` + +### Demonstrating Stream Interruption and Resumption + +This is the key feature of reliable streaming. Follow these steps to see it in action: + +1. **Start a stream**: Run the app and enter a travel planning request +2. **Note the conversation ID**: The conversation ID is displayed at the start of the stream (e.g., `Starting new conversation: @dafx-travelplanner@a1b2c3d4e5f67890abcdef1234567890`) +3. **Interrupt the stream**: While the agent is still generating text, press **`Enter`** to interrupt. The agent continues running in the background - your messages are being saved to Redis. +4. **Resume the stream**: Press **`Enter`** again to reconnect and resume the stream from the last cursor position. The app will automatically resume from where it left off. + +```text +Starting new conversation: @dafx-travelplanner@a1b2c3d4e5f67890abcdef1234567890 +Press [Enter] to interrupt the stream. + +TravelPlanner: # 7-Day Tokyo Adventure + +## Day 1: Arrival and Exploration +[Streaming content...] + +[Press Enter to interrupt] +Stream cancelled. Press [Enter] to reconnect and resume the stream from the last cursor. +Last cursor: 1734567890123-0 + +[Press Enter to resume] +Resuming conversation: @dafx-travelplanner@a1b2c3d4e5f67890abcdef1234567890 from cursor: 1734567890123-0 + +[Stream continues from where it left off...] +``` + +## Viewing Agent State + +You can view the state of the agent in the Durable Task Scheduler dashboard: + +1. Open your browser and navigate to `http://localhost:8082` +2. In the dashboard, you can see: + - **Agents**: View the state of the TravelPlanner agent, including conversation history and current state + - **Orchestrations**: View any orchestrations that may have been triggered by the agent + +The conversation ID displayed in the console output (shown as "Starting new conversation: {conversationId}") corresponds to the agent's conversation thread. You can use this to identify the agent in the dashboard and inspect: + +- The agent's conversation state +- Tool calls made by the agent (weather and events lookups) +- The streaming response state + +Note that while the console app streams responses from Redis, the agent state in DTS shows the underlying durable agent execution, including all tool calls and conversation context. + +## Architecture Overview + +```text +┌─────────────┐ stdin (prompt) ┌─────────────────────┐ +│ Client │ ─────────────────────► │ Console App │ +│ (stdin) │ │ (Program.cs) │ +└─────────────┘ └──────────────┬──────┘ + ▲ │ + │ stdout (chunks) Signal Entity + │ │ + │ ▼ + │ ┌─────────────────────┐ + │ │ AgentEntity │ + │ │ (Durable Entity) │ + │ └──────────┬──────────┘ + │ │ + │ IAgentResponseHandler + │ │ + │ ▼ + │ ┌─────────────────────┐ + │ │ RedisStreamResponse │ + │ │ Handler │ + │ └──────────┬──────────┘ + │ │ + │ XADD (write) + │ │ + │ ▼ + │ ┌─────────────────────┐ + └─────────── XREAD (poll) ────────── │ Redis Streams │ + │ (Durable Log) │ + └─────────────────────┘ +``` + +### Data Flow + +1. **Client sends prompt**: The console app reads the prompt from stdin and generates a new agent thread. + +2. **Agent invoked**: The durable agent is signaled to run the travel planner agent. This is fire-and-forget from the console app's perspective. + +3. **Responses captured**: As the agent generates responses, the `RedisStreamResponseHandler` (implementing `IAgentResponseHandler`) extracts the text from each `AgentRunResponseUpdate` and publishes it to a Redis Stream keyed by the agent session's conversation ID. + +4. **Client polls Redis**: The console app streams events by polling the Redis Stream and printing chunks to stdout as they arrive. + +5. **Resumption**: If the client interrupts the stream (e.g., by pressing Enter in the sample), it can resume from the last cursor position by providing the conversation ID and cursor to the call to resume the stream. + +## Message Delivery Guarantees + +This sample provides **at-least-once delivery** with the following characteristics: + +- **Durability**: Messages are persisted to Redis Streams with configurable TTL (default: 10 minutes). +- **Ordering**: Messages are delivered in order within a session. +- **Real-time**: Chunks are printed as soon as they arrive from Redis. + +### Important Considerations + +- **No exactly-once delivery**: If a client disconnects exactly when receiving a message, it may receive that message again upon resumption. Clients should handle duplicate messages idempotently. +- **TTL expiration**: Streams expire after the configured TTL. Clients cannot resume streams that have expired. +- **Redis guarantees**: Redis streams are backed by Redis persistence mechanisms (RDB/AOF). Ensure your Redis instance is configured for durability as needed. + +## Configuration + +| Environment Variable | Description | Default | +|---------------------|-------------|---------| +| `REDIS_CONNECTION_STRING` | Redis connection string | `localhost:6379` | +| `REDIS_STREAM_TTL_MINUTES` | How long streams are retained after last write | `10` | +| `AZURE_OPENAI_ENDPOINT` | Azure OpenAI endpoint URL | (required) | +| `AZURE_OPENAI_DEPLOYMENT` | Azure OpenAI deployment name | (required) | +| `AZURE_OPENAI_KEY` | API key (optional, uses Azure CLI auth if not set) | (optional) | + +## Cleanup + +To stop and remove the Redis Docker containers: + +```bash +docker stop redis +docker rm redis +``` diff --git a/dotnet/samples/DurableAgents/ConsoleApps/07_ReliableStreaming/RedisStreamResponseHandler.cs b/dotnet/samples/DurableAgents/ConsoleApps/07_ReliableStreaming/RedisStreamResponseHandler.cs new file mode 100644 index 0000000000..b0a95f49f6 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/07_ReliableStreaming/RedisStreamResponseHandler.cs @@ -0,0 +1,213 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Runtime.CompilerServices; +using Microsoft.Agents.AI; +using Microsoft.Agents.AI.DurableTask; +using StackExchange.Redis; + +namespace ReliableStreaming; + +/// +/// Represents a chunk of data read from a Redis stream. +/// +/// The Redis stream entry ID (can be used as a cursor for resumption). +/// The text content of the chunk, or null if this is a completion/error marker. +/// True if this chunk marks the end of the stream. +/// An error message if something went wrong, or null otherwise. +public readonly record struct StreamChunk(string EntryId, string? Text, bool IsDone, string? Error); + +/// +/// An implementation of that publishes agent response updates +/// to Redis Streams for reliable delivery. This enables clients to disconnect and reconnect +/// to ongoing agent responses without losing messages. +/// +/// +/// +/// Redis Streams provide a durable, append-only log that supports consumer groups and message +/// acknowledgment. This implementation uses auto-generated IDs (which are timestamp-based) +/// as sequence numbers, allowing clients to resume from any point in the stream. +/// +/// +/// Each agent session gets its own Redis Stream, keyed by session ID. The stream entries +/// contain text chunks extracted from objects. +/// +/// +public sealed class RedisStreamResponseHandler : IAgentResponseHandler +{ + private const int MaxEmptyReads = 300; // 5 minutes at 1 second intervals + private const int PollIntervalMs = 1000; + + private readonly IConnectionMultiplexer _redis; + private readonly TimeSpan _streamTtl; + + /// + /// Initializes a new instance of the class. + /// + /// The Redis connection multiplexer. + /// The time-to-live for stream entries. Streams will expire after this duration of inactivity. + public RedisStreamResponseHandler(IConnectionMultiplexer redis, TimeSpan streamTtl) + { + this._redis = redis; + this._streamTtl = streamTtl; + } + + /// + public async ValueTask OnStreamingResponseUpdateAsync( + IAsyncEnumerable messageStream, + CancellationToken cancellationToken) + { + // Get the current session ID from the DurableAgentContext + // This is set by the AgentEntity before invoking the response handler + DurableAgentContext? context = DurableAgentContext.Current; + if (context is null) + { + throw new InvalidOperationException( + "DurableAgentContext.Current is not set. This handler must be used within a durable agent context."); + } + + // Get conversation ID from the current thread context, which is only available in the context of + // a durable agent execution. + string conversationId = context.CurrentThread.GetService()?.ConversationId + ?? throw new InvalidOperationException("Unable to determine conversation ID from the current thread."); + string streamKey = GetStreamKey(conversationId); + + IDatabase db = this._redis.GetDatabase(); + int sequenceNumber = 0; + + await foreach (AgentRunResponseUpdate update in messageStream.WithCancellation(cancellationToken)) + { + // Extract just the text content - this avoids serialization round-trip issues + string text = update.Text; + + // Only publish non-empty text chunks + if (!string.IsNullOrEmpty(text)) + { + // Create the stream entry with the text and metadata + NameValueEntry[] entries = + [ + new NameValueEntry("text", text), + new NameValueEntry("sequence", sequenceNumber++), + new NameValueEntry("timestamp", DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()), + ]; + + // Add to the Redis Stream with auto-generated ID (timestamp-based) + await db.StreamAddAsync(streamKey, entries); + + // Refresh the TTL on each write to keep the stream alive during active streaming + await db.KeyExpireAsync(streamKey, this._streamTtl); + } + } + + // Add a sentinel entry to mark the end of the stream + NameValueEntry[] endEntries = + [ + new NameValueEntry("text", ""), + new NameValueEntry("sequence", sequenceNumber), + new NameValueEntry("timestamp", DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()), + new NameValueEntry("done", "true"), + ]; + await db.StreamAddAsync(streamKey, endEntries); + + // Set final TTL - the stream will be cleaned up after this duration + await db.KeyExpireAsync(streamKey, this._streamTtl); + } + + /// + public ValueTask OnAgentResponseAsync(AgentRunResponse message, CancellationToken cancellationToken) + { + // This handler is optimized for streaming responses. + // For non-streaming responses, we don't need to store in Redis since + // the response is returned directly to the caller. + return ValueTask.CompletedTask; + } + + /// + /// Reads chunks from a Redis stream for the given session, yielding them as they become available. + /// + /// The conversation ID to read from. + /// Optional cursor to resume from. If null, reads from the beginning. + /// Cancellation token. + /// An async enumerable of stream chunks. + public async IAsyncEnumerable ReadStreamAsync( + string conversationId, + string? cursor, + [EnumeratorCancellation] CancellationToken cancellationToken) + { + string streamKey = GetStreamKey(conversationId); + + IDatabase db = this._redis.GetDatabase(); + string startId = string.IsNullOrEmpty(cursor) ? "0-0" : cursor; + + int emptyReadCount = 0; + bool hasSeenData = false; + + while (!cancellationToken.IsCancellationRequested) + { + StreamEntry[]? entries = null; + string? errorMessage = null; + + try + { + entries = await db.StreamReadAsync(streamKey, startId, count: 100); + } + catch (Exception ex) + { + errorMessage = ex.Message; + } + + if (errorMessage != null) + { + yield return new StreamChunk(startId, null, false, errorMessage); + yield break; + } + + // entries is guaranteed to be non-null if errorMessage is null + if (entries!.Length == 0) + { + if (!hasSeenData) + { + emptyReadCount++; + if (emptyReadCount >= MaxEmptyReads) + { + yield return new StreamChunk( + startId, + null, + false, + $"Stream not found or timed out after {MaxEmptyReads * PollIntervalMs / 1000} seconds"); + yield break; + } + } + + await Task.Delay(PollIntervalMs, cancellationToken); + continue; + } + + hasSeenData = true; + + foreach (StreamEntry entry in entries) + { + startId = entry.Id.ToString(); + string? text = entry["text"]; + string? done = entry["done"]; + + if (done == "true") + { + yield return new StreamChunk(startId, null, true, null); + yield break; + } + + if (!string.IsNullOrEmpty(text)) + { + yield return new StreamChunk(startId, text, false, null); + } + } + } + } + + /// + /// Gets the Redis Stream key for a given conversation ID. + /// + /// The conversation ID. + /// The Redis Stream key. + internal static string GetStreamKey(string conversationId) => $"agent-stream:{conversationId}"; +} diff --git a/dotnet/samples/DurableAgents/ConsoleApps/README.md b/dotnet/samples/DurableAgents/ConsoleApps/README.md new file mode 100644 index 0000000000..1bd2b0d224 --- /dev/null +++ b/dotnet/samples/DurableAgents/ConsoleApps/README.md @@ -0,0 +1,109 @@ +# Console App Samples + +This directory contains samples for console app hosting of durable agents. These samples use standard I/O (stdin/stdout) for interaction, making them both interactive and scriptable. + +- **[01_SingleAgent](01_SingleAgent)**: A sample that demonstrates how to host a single conversational agent in a console app and interact with it via stdin/stdout. +- **[02_AgentOrchestration_Chaining](02_AgentOrchestration_Chaining)**: A sample that demonstrates how to host a single conversational agent in a console app and invoke it using a durable orchestration. +- **[03_AgentOrchestration_Concurrency](03_AgentOrchestration_Concurrency)**: A sample that demonstrates how to host multiple agents in a console app and run them concurrently using a durable orchestration. +- **[04_AgentOrchestration_Conditionals](04_AgentOrchestration_Conditionals)**: A sample that demonstrates how to host multiple agents in a console app and run them sequentially using a durable orchestration with conditionals. +- **[05_AgentOrchestration_HITL](05_AgentOrchestration_HITL)**: A sample that demonstrates how to implement a human-in-the-loop workflow using durable orchestration, including interactive approval prompts. +- **[06_LongRunningTools](06_LongRunningTools)**: A sample that demonstrates how agents can start and interact with durable orchestrations from tool calls to enable long-running tool scenarios. +- **[07_ReliableStreaming](07_ReliableStreaming)**: A sample that demonstrates how to implement reliable streaming for durable agents using Redis Streams, enabling clients to disconnect and reconnect without losing messages. + +## Running the Samples + +These samples are designed to be run locally in a cloned repository. + +### Prerequisites + +The following prerequisites are required to run the samples: + +- [.NET 10.0 SDK or later](https://dotnet.microsoft.com/download/dotnet) +- [Azure CLI](https://learn.microsoft.com/cli/azure/install-azure-cli) installed and authenticated (`az login`) or an API key for the Azure OpenAI service +- [Azure OpenAI Service](https://learn.microsoft.com/azure/ai-services/openai/how-to/create-resource) with a deployed model (gpt-4o-mini or better is recommended) +- [Durable Task Scheduler](https://learn.microsoft.com/azure/azure-functions/durable/durable-task-scheduler/develop-with-durable-task-scheduler) (local emulator or Azure-hosted) +- [Docker](https://docs.docker.com/get-docker/) installed if running the Durable Task Scheduler emulator locally +- [Redis](https://redis.io/) (for sample 07 only) - can be run locally using Docker + +### Configuring RBAC Permissions for Azure OpenAI + +These samples are configured to use the Azure OpenAI service with RBAC permissions to access the model. You'll need to configure the RBAC permissions for the Azure OpenAI service to allow the console app to access the model. + +Below is an example of how to configure the RBAC permissions for the Azure OpenAI service to allow the current user to access the model. + +Bash (Linux/macOS/WSL): + +```bash +az role assignment create \ + --assignee "yourname@contoso.com" \ + --role "Cognitive Services OpenAI User" \ + --scope /subscriptions//resourceGroups//providers/Microsoft.CognitiveServices/accounts/ +``` + +PowerShell: + +```powershell +az role assignment create ` + --assignee "yourname@contoso.com" ` + --role "Cognitive Services OpenAI User" ` + --scope /subscriptions//resourceGroups//providers/Microsoft.CognitiveServices/accounts/ +``` + +More information on how to configure RBAC permissions for Azure OpenAI can be found in the [Azure OpenAI documentation](https://learn.microsoft.com/azure/ai-services/openai/how-to/create-resource?pivots=cli). + +### Setting an API key for the Azure OpenAI service + +As an alternative to configuring Azure RBAC permissions, you can set an API key for the Azure OpenAI service by setting the `AZURE_OPENAI_KEY` environment variable. + +Bash (Linux/macOS/WSL): + +```bash +export AZURE_OPENAI_KEY="your-api-key" +``` + +PowerShell: + +```powershell +$env:AZURE_OPENAI_KEY="your-api-key" +``` + +### Start Durable Task Scheduler + +Most samples use the Durable Task Scheduler (DTS) to support hosted agents and durable orchestrations. DTS also allows you to view the status of orchestrations and their inputs and outputs from a web UI. + +To run the Durable Task Scheduler locally, you can use the following `docker` command: + +```bash +docker run -d --name dts-emulator -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest +``` + +The DTS dashboard will be available at `http://localhost:8080`. + +### Environment Configuration + +Each sample reads configuration from environment variables. You'll need to set the following environment variables: + +```bash +export AZURE_OPENAI_ENDPOINT="https://your-resource.openai.azure.com/" +export AZURE_OPENAI_DEPLOYMENT="your-deployment-name" +``` + +### Running the Console Apps + +Navigate to the sample directory and run the console app: + +```bash +cd dotnet/samples/DurableAgents/ConsoleApps/01_SingleAgent +dotnet run --framework net10.0 +``` + +> [!NOTE] +> The `--framework` option is required to specify the target framework for the console app because the samples are designed to support multiple target frameworks. If you are using a different target framework, you can specify it with the `--framework` option. + +The app will prompt you for input via stdin. + +### Viewing the sample output + +The console app output is displayed directly in the terminal where you ran `dotnet run`. Agent responses are printed to stdout with subtle color coding for better readability. + +You can also see the state of agents and orchestrations in the Durable Task Scheduler dashboard at `http://localhost:8082`. diff --git a/dotnet/samples/DurableAgents/Directory.Build.props b/dotnet/samples/DurableAgents/Directory.Build.props new file mode 100644 index 0000000000..7c4cb7dea2 --- /dev/null +++ b/dotnet/samples/DurableAgents/Directory.Build.props @@ -0,0 +1,9 @@ + + + + + + + + + \ No newline at end of file diff --git a/dotnet/tests/Microsoft.Agents.AI.DurableTask.IntegrationTests/ConsoleAppSamplesValidation.cs b/dotnet/tests/Microsoft.Agents.AI.DurableTask.IntegrationTests/ConsoleAppSamplesValidation.cs new file mode 100644 index 0000000000..5ef87403d7 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.DurableTask.IntegrationTests/ConsoleAppSamplesValidation.cs @@ -0,0 +1,884 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Collections.Concurrent; +using System.Diagnostics; +using System.Reflection; +using System.Text; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.Logging; +using Xunit.Abstractions; + +namespace Microsoft.Agents.AI.DurableTask.IntegrationTests; + +[Collection("Samples")] +[Trait("Category", "SampleValidation")] +public sealed class ConsoleAppSamplesValidation(ITestOutputHelper outputHelper) : IAsyncLifetime +{ + private const string DtsPort = "8080"; + private const string RedisPort = "6379"; + + private static readonly string s_dotnetTargetFramework = GetTargetFramework(); + private static readonly IConfiguration s_configuration = + new ConfigurationBuilder() + .AddUserSecrets(Assembly.GetExecutingAssembly()) + .AddEnvironmentVariables() + .Build(); + + private static bool s_infrastructureStarted; + private static readonly string s_samplesPath = Path.GetFullPath( + Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "..", "..", "..", "..", "..", "samples", "DurableAgents", "ConsoleApps")); + + private readonly ITestOutputHelper _outputHelper = outputHelper; + + async Task IAsyncLifetime.InitializeAsync() + { + if (!s_infrastructureStarted) + { + await this.StartSharedInfrastructureAsync(); + s_infrastructureStarted = true; + } + } + + async Task IAsyncLifetime.DisposeAsync() + { + // Nothing to clean up + await Task.CompletedTask; + } + + [Fact] + public async Task SingleAgentSampleValidationAsync() + { + using CancellationTokenSource testTimeoutCts = this.CreateTestTimeoutCts(); + string samplePath = Path.Combine(s_samplesPath, "01_SingleAgent"); + await this.RunSampleTestAsync(samplePath, async (process, logs) => + { + string agentResponse = string.Empty; + bool inputSent = false; + + // Read output from logs queue + string? line; + while ((line = this.ReadLogLine(logs, testTimeoutCts.Token)) != null) + { + // Look for the agent's response. Unlike the interactive mode, we won't actually see a line + // that starts with "Joker: ". Instead, we'll see a line that looks like "You: Joker: ..." because + // the standard input is *not* echoed back to standard output. + if (line.Contains("Joker: ", StringComparison.OrdinalIgnoreCase)) + { + // This will give us the first line of the agent's response, which is all we need to verify that the agent is working. + agentResponse = line.Substring("Joker: ".Length).Trim(); + break; + } + else if (!inputSent) + { + // Send input to stdin after we've started seeing output from the app + await this.WriteInputAsync(process, "Tell me a joke about a pirate.", testTimeoutCts.Token); + inputSent = true; + } + } + + Assert.True(inputSent, "Input was not sent to the agent"); + Assert.NotEmpty(agentResponse); + + // Send exit command + await this.WriteInputAsync(process, "exit", testTimeoutCts.Token); + }); + } + + [Fact] + public async Task SingleAgentOrchestrationChainingSampleValidationAsync() + { + using CancellationTokenSource testTimeoutCts = this.CreateTestTimeoutCts(); + string samplePath = Path.Combine(s_samplesPath, "02_AgentOrchestration_Chaining"); + await this.RunSampleTestAsync(samplePath, async (process, logs) => + { + // Console app runs automatically, just wait for completion + string? line; + bool foundSuccess = false; + + while ((line = this.ReadLogLine(logs, testTimeoutCts.Token)) != null) + { + if (line.Contains("Orchestration completed successfully!", StringComparison.OrdinalIgnoreCase)) + { + foundSuccess = true; + } + + if (line.Contains("Result:", StringComparison.OrdinalIgnoreCase)) + { + string result = line.Substring("Result:".Length).Trim(); + Assert.NotEmpty(result); + break; + } + + // Check for failure + if (line.Contains("Orchestration failed!", StringComparison.OrdinalIgnoreCase)) + { + Assert.Fail("Orchestration failed."); + } + } + + Assert.True(foundSuccess, "Orchestration did not complete successfully."); + }); + } + + [Fact] + public async Task MultiAgentConcurrencySampleValidationAsync() + { + using CancellationTokenSource testTimeoutCts = this.CreateTestTimeoutCts(); + string samplePath = Path.Combine(s_samplesPath, "03_AgentOrchestration_Concurrency"); + await this.RunSampleTestAsync(samplePath, async (process, logs) => + { + // Send input to stdin + await this.WriteInputAsync(process, "What is temperature?", testTimeoutCts.Token); + + // Read output from logs queue + StringBuilder output = new(); + string? line; + bool foundSuccess = false; + bool foundPhysicist = false; + bool foundChemist = false; + + while ((line = this.ReadLogLine(logs, testTimeoutCts.Token)) != null) + { + output.AppendLine(line); + + if (line.Contains("Orchestration completed successfully!", StringComparison.OrdinalIgnoreCase)) + { + foundSuccess = true; + } + + if (line.Contains("Physicist's response:", StringComparison.OrdinalIgnoreCase)) + { + foundPhysicist = true; + } + + if (line.Contains("Chemist's response:", StringComparison.OrdinalIgnoreCase)) + { + foundChemist = true; + } + + // Check for failure + if (line.Contains("Orchestration failed!", StringComparison.OrdinalIgnoreCase)) + { + Assert.Fail("Orchestration failed."); + } + + // Stop reading once we have both responses + if (foundSuccess && foundPhysicist && foundChemist) + { + break; + } + } + + Assert.True(foundSuccess, "Orchestration did not complete successfully."); + Assert.True(foundPhysicist, "Physicist response not found."); + Assert.True(foundChemist, "Chemist response not found."); + }); + } + + [Fact] + public async Task MultiAgentConditionalSampleValidationAsync() + { + using CancellationTokenSource testTimeoutCts = this.CreateTestTimeoutCts(); + string samplePath = Path.Combine(s_samplesPath, "04_AgentOrchestration_Conditionals"); + await this.RunSampleTestAsync(samplePath, async (process, logs) => + { + // Test with legitimate email + await this.TestSpamDetectionAsync( + process: process, + logs: logs, + emailId: "email-001", + emailContent: "Hi John. I wanted to follow up on our meeting yesterday about the quarterly report. Could you please send me the updated figures by Friday? Thanks!", + expectedSpam: false, + testTimeoutCts.Token); + + // Restart the process for the second test + await process.WaitForExitAsync(); + }); + + // Run second test with spam email + using CancellationTokenSource testTimeoutCts2 = this.CreateTestTimeoutCts(); + await this.RunSampleTestAsync(samplePath, async (process, logs) => + { + await this.TestSpamDetectionAsync( + process, + logs, + emailId: "email-002", + emailContent: "URGENT! You've won $1,000,000! Click here now to claim your prize! Limited time offer! Don't miss out!", + expectedSpam: true, + testTimeoutCts2.Token); + }); + } + + private async Task TestSpamDetectionAsync( + Process process, + BlockingCollection logs, + string emailId, + string emailContent, + bool expectedSpam, + CancellationToken cancellationToken) + { + // Send email content to stdin + await this.WriteInputAsync(process, emailContent, cancellationToken); + + // Read output from logs queue + string? line; + bool foundSuccess = false; + + while ((line = this.ReadLogLine(logs, cancellationToken)) != null) + { + if (line.Contains("Email sent", StringComparison.OrdinalIgnoreCase)) + { + Assert.False(expectedSpam, "Email was sent, but was expected to be marked as spam."); + } + + if (line.Contains("Email marked as spam", StringComparison.OrdinalIgnoreCase)) + { + Assert.True(expectedSpam, "Email was marked as spam, but was expected to be sent."); + } + + if (line.Contains("Orchestration completed successfully!", StringComparison.OrdinalIgnoreCase)) + { + foundSuccess = true; + break; + } + + // Check for failure + if (line.Contains("Orchestration failed!", StringComparison.OrdinalIgnoreCase)) + { + Assert.Fail("Orchestration failed."); + } + } + + Assert.True(foundSuccess, "Orchestration did not complete successfully."); + } + + [Fact] + public async Task SingleAgentOrchestrationHITLSampleValidationAsync() + { + string samplePath = Path.Combine(s_samplesPath, "05_AgentOrchestration_HITL"); + + await this.RunSampleTestAsync(samplePath, async (process, logs) => + { + using CancellationTokenSource testTimeoutCts = this.CreateTestTimeoutCts(); + + // Start the HITL orchestration following the happy path from README + await this.WriteInputAsync(process, "The Future of Artificial Intelligence", testTimeoutCts.Token); + await this.WriteInputAsync(process, "3", testTimeoutCts.Token); + await this.WriteInputAsync(process, "72", testTimeoutCts.Token); + + // Read output from logs queue + string? line; + bool rejectionSent = false; + bool approvalSent = false; + bool contentPublished = false; + + while ((line = this.ReadLogLine(logs, testTimeoutCts.Token)) != null) + { + // Look for notification that content is ready. The first time we see this, we should send a rejection. + // The second time we see this, we should send approval. + if (line.Contains("Content is ready for review", StringComparison.OrdinalIgnoreCase)) + { + if (!rejectionSent) + { + // Prompt: Approve? (y/n): + await this.WriteInputAsync(process, "n", testTimeoutCts.Token); + + // Prompt: Feedback (optional): + await this.WriteInputAsync( + process, + "The article needs more technical depth and better examples. Rewrite it with less than 300 words.", + testTimeoutCts.Token); + rejectionSent = true; + } + else if (!approvalSent) + { + // Prompt: Approve? (y/n): + await this.WriteInputAsync(process, "y", testTimeoutCts.Token); + + // Prompt: Feedback (optional): + await this.WriteInputAsync(process, "Looks good!", testTimeoutCts.Token); + approvalSent = true; + } + else + { + // This should never happen + Assert.Fail("Unexpected message found."); + } + } + + // Look for success message + if (line.Contains("PUBLISHING: Content has been published", StringComparison.OrdinalIgnoreCase)) + { + contentPublished = true; + break; + } + + // Check for failure + if (line.Contains("Orchestration failed", StringComparison.OrdinalIgnoreCase)) + { + Assert.Fail("Orchestration failed."); + } + } + + Assert.True(rejectionSent, "Wasn't prompted with the first draft."); + Assert.True(approvalSent, "Wasn't prompted with the second draft."); + Assert.True(contentPublished, "Content was not published."); + }); + } + + [Fact] + public async Task LongRunningToolsSampleValidationAsync() + { + string samplePath = Path.Combine(s_samplesPath, "06_LongRunningTools"); + await this.RunSampleTestAsync(samplePath, async (process, logs) => + { + // This test takes a bit longer to run due to the multiple agent interactions and the lengthy content generation. + using CancellationTokenSource testTimeoutCts = this.CreateTestTimeoutCts(TimeSpan.FromSeconds(90)); + + // Test starting an agent that schedules a content generation orchestration + await this.WriteInputAsync( + process, + "Start a content generation workflow for the topic 'The Future of Artificial Intelligence'. Keep it less than 300 words.", + testTimeoutCts.Token); + + // Read output from logs queue + bool rejectionSent = false; + bool approvalSent = false; + bool contentPublished = false; + + string? line; + while ((line = this.ReadLogLine(logs, testTimeoutCts.Token)) != null) + { + // Look for notification that content is ready. The first time we see this, we should send a rejection. + // The second time we see this, we should send approval. + if (line.Contains("NOTIFICATION: Please review the following content for approval", StringComparison.OrdinalIgnoreCase)) + { + // Wait for the notification to be fully written to the console + await Task.Delay(TimeSpan.FromSeconds(1), testTimeoutCts.Token); + + if (!rejectionSent) + { + // Reject the content with feedback. Note that we need to send a newline character to the console first before sending the input. + await this.WriteInputAsync( + process, + "\nReject the content with feedback: Make it even shorter.", + testTimeoutCts.Token); + rejectionSent = true; + } + else if (!approvalSent) + { + // Approve the content. Note that we need to send a newline character to the console first before sending the input. + await this.WriteInputAsync( + process, + "\nApprove the content", + testTimeoutCts.Token); + approvalSent = true; + } + else + { + // This should never happen + Assert.Fail("Unexpected message found."); + } + } + + // Look for success message + if (line.Contains("PUBLISHING: Content has been published successfully", StringComparison.OrdinalIgnoreCase)) + { + contentPublished = true; + + // Ask for the status of the workflow to confirm that it completed successfully. + await Task.Delay(TimeSpan.FromSeconds(1), testTimeoutCts.Token); + await this.WriteInputAsync(process, "\nGet the status of the workflow you previously started", testTimeoutCts.Token); + } + + // Check for workflow completion or failure + if (contentPublished) + { + if (line.Contains("Completed", StringComparison.OrdinalIgnoreCase)) + { + break; + } + else if (line.Contains("Failed", StringComparison.OrdinalIgnoreCase)) + { + Assert.Fail("Workflow failed."); + } + } + } + + Assert.True(rejectionSent, "Wasn't prompted with the first draft."); + Assert.True(approvalSent, "Wasn't prompted with the second draft."); + Assert.True(contentPublished, "Content was not published."); + }); + } + + [Fact] + public async Task ReliableStreamingSampleValidationAsync() + { + string samplePath = Path.Combine(s_samplesPath, "07_ReliableStreaming"); + await this.RunSampleTestAsync(samplePath, async (process, logs) => + { + // This test takes a bit longer to run due to the multiple agent interactions and the lengthy content generation. + using CancellationTokenSource testTimeoutCts = this.CreateTestTimeoutCts(TimeSpan.FromSeconds(90)); + + // Test the agent endpoint with a simple prompt + await this.WriteInputAsync(process, "Plan a 3-day trip to Seattle. Include daily activities.", testTimeoutCts.Token); + + // Read output from stdout - should stream in real-time + // NOTE: The sample uses Console.Write() for streaming chunks, which means content may not be line-buffered. + // We test the interrupt/resume flow by: + // 1. Waiting for at least 10 lines of content + // 2. Sending Enter to interrupt + // 3. Verifying we get "Last cursor" output + // 4. Sending Enter again to resume + // 5. Verifying we get more content and that we're not restarting from the beginning + string? line; + bool foundConversationStart = false; + int contentLinesBeforeInterrupt = 0; + int contentLinesAfterResume = 0; + bool foundLastCursor = false; + bool foundResumeMessage = false; + bool interrupted = false; + bool resumed = false; + + // Read output with a reasonable timeout + using CancellationTokenSource readTimeoutCts = this.CreateTestTimeoutCts(); + try + { + while ((line = this.ReadLogLine(logs, readTimeoutCts.Token)) != null) + { + // Look for the conversation start message (updated format) + if (line.Contains("Conversation ID", StringComparison.OrdinalIgnoreCase)) + { + foundConversationStart = true; + continue; + } + + // Look for completion message (if stream completes naturally) + if (line.Contains("Conversation completed", StringComparison.OrdinalIgnoreCase)) + { + break; + } + + // Check if this is a content line (not prompts or status messages) + bool isContentLine = !string.IsNullOrWhiteSpace(line) && + !line.Contains("Conversation ID", StringComparison.OrdinalIgnoreCase) && + !line.Contains("Press [Enter]", StringComparison.OrdinalIgnoreCase) && + !line.Contains("You:", StringComparison.OrdinalIgnoreCase) && + !line.Contains("exit", StringComparison.OrdinalIgnoreCase) && + !line.Contains("Stream cancelled", StringComparison.OrdinalIgnoreCase) && + !line.Contains("Resuming conversation", StringComparison.OrdinalIgnoreCase) && + !line.Contains("Last cursor", StringComparison.OrdinalIgnoreCase); + + // Phase 1: Collect content before interrupt + if (foundConversationStart && !interrupted && isContentLine) + { + contentLinesBeforeInterrupt++; + } + + // Phase 2: Wait for enough content, then interrupt + if (foundConversationStart && !interrupted && contentLinesBeforeInterrupt >= 5) + { + this._outputHelper.WriteLine($"Interrupting stream after {contentLinesBeforeInterrupt} content lines"); + interrupted = true; + + // Send Enter to interrupt the stream + await this.WriteInputAsync(process, string.Empty, testTimeoutCts.Token); + } + + // Phase 3: Look for "Last cursor" message after interrupt + if (interrupted && !resumed && line.Contains("Last cursor", StringComparison.OrdinalIgnoreCase)) + { + foundLastCursor = true; + + // Send Enter again to resume + this._outputHelper.WriteLine("Resuming stream from last cursor"); + await this.WriteInputAsync(process, string.Empty, testTimeoutCts.Token); + resumed = true; + } + + // Phase 4: Look for resume message + if (resumed && line.Contains("Resuming conversation", StringComparison.OrdinalIgnoreCase)) + { + foundResumeMessage = true; + } + + // Phase 5: Collect content after resume + if (resumed && isContentLine) + { + contentLinesAfterResume++; + } + + // Stop once we've verified the interrupt/resume flow works + if (resumed && foundResumeMessage && contentLinesAfterResume >= 5) + { + this._outputHelper.WriteLine($"Successfully verified interrupt/resume: {contentLinesBeforeInterrupt} lines before, {contentLinesAfterResume} lines after"); + break; + } + } + } + catch (OperationCanceledException) + { + // Timeout - check if we got enough to verify the flow + this._outputHelper.WriteLine($"Read timeout reached. Interrupted: {interrupted}, Resumed: {resumed}, Content before: {contentLinesBeforeInterrupt}, Content after: {contentLinesAfterResume}"); + } + + Assert.True(foundConversationStart, "Conversation start message not found."); + Assert.True(contentLinesBeforeInterrupt >= 5, $"Not enough content before interrupt (got {contentLinesBeforeInterrupt})."); + Assert.True(interrupted, "Stream was not interrupted."); + Assert.True(foundLastCursor, "'Last cursor' message not found after interrupt."); + Assert.True(resumed, "Stream was not resumed."); + Assert.True(foundResumeMessage, "Resume message not found."); + Assert.True(contentLinesAfterResume > 0, "No content received after resume (expected to continue from cursor, not restart)."); + }); + } + + private static string GetTargetFramework() + { + string filePath = new Uri(typeof(ConsoleAppSamplesValidation).Assembly.Location).LocalPath; + string directory = Path.GetDirectoryName(filePath)!; + string tfm = Path.GetFileName(directory); + if (tfm.StartsWith("net", StringComparison.OrdinalIgnoreCase)) + { + return tfm; + } + + throw new InvalidOperationException($"Unable to find target framework in path: {filePath}"); + } + + private async Task StartSharedInfrastructureAsync() + { + this._outputHelper.WriteLine("Starting shared infrastructure for console app samples..."); + + // Start DTS emulator + await this.StartDtsEmulatorAsync(); + + // Start Redis + await this.StartRedisAsync(); + + // Wait for infrastructure to be ready + await Task.Delay(TimeSpan.FromSeconds(5)); + } + + private async Task StartDtsEmulatorAsync() + { + // Start DTS emulator if it's not already running + if (!await this.IsDtsEmulatorRunningAsync()) + { + this._outputHelper.WriteLine("Starting DTS emulator..."); + await this.RunCommandAsync("docker", [ + "run", "-d", + "--name", "dts-emulator", + "-p", $"{DtsPort}:8080", + "mcr.microsoft.com/dts/dts-emulator:latest" + ]); + } + } + + private async Task StartRedisAsync() + { + if (!await this.IsRedisRunningAsync()) + { + this._outputHelper.WriteLine("Starting Redis..."); + await this.RunCommandAsync("docker", [ + "run", "-d", + "--name", "redis", + "-p", $"{RedisPort}:6379", + "redis:latest" + ]); + } + } + + private async Task IsDtsEmulatorRunningAsync() + { + this._outputHelper.WriteLine($"Checking if DTS emulator is running at http://localhost:{DtsPort}/healthz..."); + + // DTS emulator doesn't support HTTP/1.1, so we need to use HTTP/2.0 + using HttpClient http2Client = new() + { + DefaultRequestVersion = new Version(2, 0), + DefaultVersionPolicy = HttpVersionPolicy.RequestVersionExact + }; + + try + { + using CancellationTokenSource timeoutCts = new(TimeSpan.FromSeconds(30)); + using HttpResponseMessage response = await http2Client.GetAsync(new Uri($"http://localhost:{DtsPort}/healthz"), timeoutCts.Token); + if (response.Content.Headers.ContentLength > 0) + { + string content = await response.Content.ReadAsStringAsync(timeoutCts.Token); + this._outputHelper.WriteLine($"DTS emulator health check response: {content}"); + } + + if (response.IsSuccessStatusCode) + { + this._outputHelper.WriteLine("DTS emulator is running"); + return true; + } + + this._outputHelper.WriteLine($"DTS emulator is not running. Status code: {response.StatusCode}"); + return false; + } + catch (HttpRequestException ex) + { + this._outputHelper.WriteLine($"DTS emulator is not running: {ex.Message}"); + return false; + } + } + + private async Task IsRedisRunningAsync() + { + this._outputHelper.WriteLine($"Checking if Redis is running at localhost:{RedisPort}..."); + + try + { + using CancellationTokenSource timeoutCts = new(TimeSpan.FromSeconds(30)); + ProcessStartInfo startInfo = new() + { + FileName = "docker", + Arguments = "exec redis redis-cli ping", + UseShellExecute = false, + RedirectStandardOutput = true, + RedirectStandardError = true, + CreateNoWindow = true + }; + + using Process process = new() { StartInfo = startInfo }; + if (!process.Start()) + { + this._outputHelper.WriteLine("Failed to start docker exec command"); + return false; + } + + string output = await process.StandardOutput.ReadToEndAsync(timeoutCts.Token); + await process.WaitForExitAsync(timeoutCts.Token); + + if (process.ExitCode == 0 && output.Contains("PONG", StringComparison.OrdinalIgnoreCase)) + { + this._outputHelper.WriteLine("Redis is running"); + return true; + } + + this._outputHelper.WriteLine($"Redis is not running. Exit code: {process.ExitCode}, Output: {output}"); + return false; + } + catch (Exception ex) + { + this._outputHelper.WriteLine($"Redis is not running: {ex.Message}"); + return false; + } + } + + private async Task RunSampleTestAsync(string samplePath, Func, Task> testAction) + { + // Start the console app + // Use BlockingCollection to safely read logs asynchronously captured from the process + using BlockingCollection logsContainer = []; + using Process appProcess = this.StartConsoleApp(samplePath, logsContainer); + try + { + // Run the test + await testAction(appProcess, logsContainer); + } + catch (OperationCanceledException e) + { + throw new TimeoutException("Core test logic timed out!", e); + } + finally + { + logsContainer.CompleteAdding(); + await this.StopProcessAsync(appProcess); + } + } + + private sealed record OutputLog(DateTime Timestamp, LogLevel Level, string Message); + + /// + /// Writes a line to the process's stdin and flushes it. + /// Logs the input being sent for debugging purposes. + /// + private async Task WriteInputAsync(Process process, string input, CancellationToken cancellationToken) + { + this._outputHelper.WriteLine($"{DateTime.Now:HH:mm:ss.fff} [{process.ProcessName}(in)]: {input}"); + await process.StandardInput.WriteLineAsync(input); + await process.StandardInput.FlushAsync(cancellationToken); + } + + /// + /// Reads a line from the logs queue, filtering for Information level logs (stdout). + /// Returns null if the collection is completed and empty, or if cancellation is requested. + /// + private string? ReadLogLine(BlockingCollection logs, CancellationToken cancellationToken) + { + try + { + while (!cancellationToken.IsCancellationRequested) + { + // Block until a log entry is available or cancellation is requested + // Take will throw OperationCanceledException if cancelled, or InvalidOperationException if collection is completed + OutputLog log = logs.Take(cancellationToken); + + // Check for unhandled exceptions in the logs, which are never expected (but can happen) + if (log.Message.Contains("Unhandled exception")) + { + Assert.Fail("Console app encountered an unhandled exception."); + } + + // Only return Information level logs (stdout), skip Error logs (stderr) + if (log.Level == LogLevel.Information) + { + return log.Message; + } + } + } + catch (OperationCanceledException) + { + // Cancellation requested + return null; + } + catch (InvalidOperationException) + { + // Collection is completed and empty + return null; + } + + return null; + } + + private Process StartConsoleApp(string samplePath, BlockingCollection logs) + { + ProcessStartInfo startInfo = new() + { + FileName = "dotnet", + Arguments = $"run --framework {s_dotnetTargetFramework}", + WorkingDirectory = samplePath, + UseShellExecute = false, + RedirectStandardOutput = true, + RedirectStandardError = true, + RedirectStandardInput = true, + }; + + string openAiEndpoint = s_configuration["AZURE_OPENAI_ENDPOINT"] ?? + throw new InvalidOperationException("The required AZURE_OPENAI_ENDPOINT env variable is not set."); + string openAiDeployment = s_configuration["AZURE_OPENAI_CHAT_DEPLOYMENT_NAME"] ?? + throw new InvalidOperationException("The required AZURE_OPENAI_CHAT_DEPLOYMENT_NAME env variable is not set."); + + // Set required environment variables for the app + startInfo.EnvironmentVariables["AZURE_OPENAI_ENDPOINT"] = openAiEndpoint; + startInfo.EnvironmentVariables["AZURE_OPENAI_DEPLOYMENT"] = openAiDeployment; + startInfo.EnvironmentVariables["DURABLE_TASK_SCHEDULER_CONNECTION_STRING"] = + $"Endpoint=http://localhost:{DtsPort};TaskHub=default;Authentication=None"; + startInfo.EnvironmentVariables["REDIS_CONNECTION_STRING"] = $"localhost:{RedisPort}"; + + Process process = new() { StartInfo = startInfo }; + + // Capture the output and error streams asynchronously + // These events fire asynchronously, so we add to the blocking collection which is thread-safe + process.ErrorDataReceived += (sender, e) => + { + if (e.Data != null) + { + string logMessage = $"{DateTime.Now:HH:mm:ss.fff} [{startInfo.FileName}(err)]: {e.Data}"; + this._outputHelper.WriteLine(logMessage); + Debug.WriteLine(logMessage); + try + { + logs.Add(new OutputLog(DateTime.Now, LogLevel.Error, e.Data)); + } + catch (InvalidOperationException) + { + // Collection is completed, ignore + } + } + }; + + process.OutputDataReceived += (sender, e) => + { + if (e.Data != null) + { + string logMessage = $"{DateTime.Now:HH:mm:ss.fff} [{startInfo.FileName}(out)]: {e.Data}"; + this._outputHelper.WriteLine(logMessage); + Debug.WriteLine(logMessage); + try + { + logs.Add(new OutputLog(DateTime.Now, LogLevel.Information, e.Data)); + } + catch (InvalidOperationException) + { + // Collection is completed, ignore + } + } + }; + + if (!process.Start()) + { + throw new InvalidOperationException("Failed to start the console app"); + } + + process.BeginErrorReadLine(); + process.BeginOutputReadLine(); + + return process; + } + + private async Task RunCommandAsync(string command, string[] args) + { + await this.RunCommandAsync(command, workingDirectory: null, args: args); + } + + private async Task RunCommandAsync(string command, string? workingDirectory, string[] args) + { + ProcessStartInfo startInfo = new() + { + FileName = command, + Arguments = string.Join(" ", args), + WorkingDirectory = workingDirectory, + UseShellExecute = false, + RedirectStandardOutput = true, + RedirectStandardError = true, + CreateNoWindow = true + }; + + this._outputHelper.WriteLine($"Running command: {command} {string.Join(" ", args)}"); + + using Process process = new() { StartInfo = startInfo }; + process.ErrorDataReceived += (sender, e) => this._outputHelper.WriteLine($"[{command}(err)]: {e.Data}"); + process.OutputDataReceived += (sender, e) => this._outputHelper.WriteLine($"[{command}(out)]: {e.Data}"); + if (!process.Start()) + { + throw new InvalidOperationException("Failed to start the command"); + } + process.BeginErrorReadLine(); + process.BeginOutputReadLine(); + + using CancellationTokenSource cancellationTokenSource = new(TimeSpan.FromMinutes(1)); + await process.WaitForExitAsync(cancellationTokenSource.Token); + + this._outputHelper.WriteLine($"Command completed with exit code: {process.ExitCode}"); + } + + private async Task StopProcessAsync(Process process) + { + try + { + if (!process.HasExited) + { + this._outputHelper.WriteLine($"{DateTime.Now:HH:mm:ss.fff} Killing process {process.ProcessName}#{process.Id}"); + process.Kill(entireProcessTree: true); + + using CancellationTokenSource timeoutCts = new(TimeSpan.FromSeconds(10)); + await process.WaitForExitAsync(timeoutCts.Token); + this._outputHelper.WriteLine($"{DateTime.Now:HH:mm:ss.fff} Process exited: {process.Id}"); + } + } + catch (Exception ex) + { + this._outputHelper.WriteLine($"{DateTime.Now:HH:mm:ss.fff} Failed to stop process: {ex.Message}"); + } + } + + private CancellationTokenSource CreateTestTimeoutCts(TimeSpan? timeout = null) + { + TimeSpan testTimeout = Debugger.IsAttached ? TimeSpan.FromMinutes(5) : timeout ?? TimeSpan.FromSeconds(60); + return new CancellationTokenSource(testTimeout); + } +}