Skip to content

Commit 216ddb9

Browse files
SergeyMenshykhSergeyMenshykh
committed
Merge branch 'update-a2a-agent' of https://github.com/SergeyMenshykh/semantic-kernel into update-a2a-agent
2 parents c9119f8 + 49c671d commit 216ddb9

24 files changed

Lines changed: 266 additions & 164 deletions

‎dotnet/src/Agents/AzureAI/AzureAIAgent.cs‎

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -282,19 +282,29 @@ public async IAsyncEnumerable<AgentResponseItem<StreamingChatMessageContent>> In
282282
cancellationToken);
283283

284284
// Return the chunks to the caller.
285+
int messageIndex = 0;
285286
await foreach (var result in invokeResults.ConfigureAwait(false))
286287
{
288+
// Notify the thread of any messages that were assembled from the streaming response during this iteration.
289+
await NotifyMessagesAsync().ConfigureAwait(false);
290+
287291
yield return new(result, azureAIAgentThread);
288292
}
289293

290-
// Notify the thread of any new messages that were assembled from the streaming response.
291-
foreach (var newMessage in newMessagesReceiver)
292-
{
293-
await this.NotifyThreadOfNewMessage(azureAIAgentThread, newMessage, cancellationToken).ConfigureAwait(false);
294+
// Notify the thread of any remaining messages that were assembled from the streaming response after all iterations are complete.
295+
await NotifyMessagesAsync().ConfigureAwait(false);
294296

295-
if (options?.OnIntermediateMessage is not null)
297+
async Task NotifyMessagesAsync()
298+
{
299+
for (; messageIndex < newMessagesReceiver.Count; messageIndex++)
296300
{
297-
await options.OnIntermediateMessage(newMessage).ConfigureAwait(false);
301+
ChatMessageContent newMessage = newMessagesReceiver[messageIndex];
302+
await this.NotifyThreadOfNewMessage(azureAIAgentThread, newMessage, cancellationToken).ConfigureAwait(false);
303+
304+
if (options?.OnIntermediateMessage is not null)
305+
{
306+
await options.OnIntermediateMessage(newMessage).ConfigureAwait(false);
307+
}
298308
}
299309
}
300310
}

‎dotnet/src/Agents/AzureAI/Internal/AgentThreadActions.cs‎

Lines changed: 62 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -388,7 +388,7 @@ public static async IAsyncEnumerable<StreamingChatMessageContent> InvokeStreamin
388388
// Evaluate status and process steps and messages, as encountered.
389389
HashSet<string> processedStepIds = [];
390390
Dictionary<string, FunctionResultContent[]> stepFunctionResults = [];
391-
List<RunStep> stepsToProcess = [];
391+
List<RunStep> messageCreationStepsToProcess = [];
392392

393393
FunctionCallsProcessor functionProcessor = new(logger);
394394
// This matches current behavior. Will be configurable upon integrating with `FunctionChoice` (#6795/#5200)
@@ -401,7 +401,7 @@ public static async IAsyncEnumerable<StreamingChatMessageContent> InvokeStreamin
401401
// Check for cancellation
402402
cancellationToken.ThrowIfCancellationRequested();
403403

404-
stepsToProcess.Clear();
404+
messageCreationStepsToProcess.Clear();
405405

406406
await foreach (StreamingUpdate update in asyncUpdates.ConfigureAwait(false))
407407
{
@@ -440,9 +440,14 @@ public static async IAsyncEnumerable<StreamingChatMessageContent> InvokeStreamin
440440
{
441441
switch (stepUpdate.UpdateKind)
442442
{
443-
case StreamingUpdateReason.RunStepCompleted:
444-
stepsToProcess.Add(stepUpdate.Value);
443+
case StreamingUpdateReason.RunStepCompleted when stepUpdate.Value.StepDetails is RunStepToolCallDetails toolDetails:
444+
ProcessToolCallStep(stepUpdate.Value, toolDetails, agent, messages, threadId, stepFunctionResults);
445445
break;
446+
447+
case StreamingUpdateReason.RunStepCompleted when stepUpdate.Value.StepDetails is RunStepMessageCreationDetails:
448+
messageCreationStepsToProcess.Add(stepUpdate.Value);
449+
break;
450+
446451
default:
447452
break;
448453
}
@@ -510,55 +515,75 @@ await functionProcessor.InvokeFunctionCallsAsync(
510515
}
511516
}
512517

513-
if (stepsToProcess.Count > 0)
518+
if (messageCreationStepsToProcess.Count > 0)
514519
{
515520
logger.LogAzureAIAgentProcessingRunMessages(nameof(InvokeAsync), run!.Id, threadId);
516521

517-
foreach (RunStep step in stepsToProcess)
522+
foreach (RunStep step in messageCreationStepsToProcess)
518523
{
519524
if (step.StepDetails is RunStepMessageCreationDetails messageDetails)
520525
{
521-
PersistentThreadMessage? message =
522-
await RetrieveMessageAsync(
523-
client,
524-
threadId,
525-
messageDetails.MessageCreation.MessageId,
526-
agent.PollingOptions.MessageSynchronizationDelay,
527-
cancellationToken).ConfigureAwait(false);
528-
529-
if (message != null)
530-
{
531-
ChatMessageContent content = GenerateMessageContent(agent.GetName(), message, step, logger);
532-
messages?.Add(content);
533-
}
534-
}
535-
else if (step.StepDetails is RunStepToolCallDetails toolDetails)
536-
{
537-
foreach (RunStepToolCall toolCall in toolDetails.ToolCalls)
538-
{
539-
if (toolCall is RunStepFunctionToolCall functionCall)
540-
{
541-
messages?.Add(GenerateFunctionResultContent(agent.GetName(), stepFunctionResults[step.Id], step));
542-
stepFunctionResults.Remove(step.Id);
543-
break;
544-
}
545-
546-
if (toolCall is RunStepCodeInterpreterToolCall codeCall)
547-
{
548-
messages?.Add(GenerateCodeInterpreterContent(agent.GetName(), codeCall.Input, step));
549-
}
550-
}
526+
await ProcessMessageCreationStepAsync(step, messageDetails, agent, client, messages, threadId, logger, cancellationToken).ConfigureAwait(false);
551527
}
552528
}
553529

554-
logger.LogAzureAIAgentProcessedRunMessages(nameof(InvokeAsync), stepsToProcess.Count, run!.Id, threadId);
530+
logger.LogAzureAIAgentProcessedRunMessages(nameof(InvokeAsync), messageCreationStepsToProcess.Count, run!.Id, threadId);
555531
}
556532
}
557533
while (run?.Status != RunStatus.Completed);
558534

559535
logger.LogAzureAIAgentCompletedRun(nameof(InvokeAsync), run?.Id ?? "Failed", threadId);
560536
}
561537

538+
private static async Task ProcessMessageCreationStepAsync(
539+
RunStep step,
540+
RunStepMessageCreationDetails messageDetails,
541+
AzureAIAgent agent,
542+
PersistentAgentsClient client,
543+
IList<ChatMessageContent>? messages,
544+
string threadId,
545+
ILogger logger,
546+
CancellationToken cancellationToken)
547+
{
548+
PersistentThreadMessage? message =
549+
await RetrieveMessageAsync(
550+
client,
551+
threadId,
552+
messageDetails.MessageCreation.MessageId,
553+
agent.PollingOptions.MessageSynchronizationDelay,
554+
cancellationToken).ConfigureAwait(false);
555+
556+
if (message != null)
557+
{
558+
ChatMessageContent content = GenerateMessageContent(agent.GetName(), message, step, logger);
559+
messages?.Add(content);
560+
}
561+
}
562+
563+
private static void ProcessToolCallStep(
564+
RunStep step,
565+
RunStepToolCallDetails toolDetails,
566+
AzureAIAgent agent,
567+
IList<ChatMessageContent>? messages,
568+
string threadId,
569+
Dictionary<string, FunctionResultContent[]> stepFunctionResults)
570+
{
571+
foreach (RunStepToolCall toolCall in toolDetails.ToolCalls)
572+
{
573+
if (toolCall is RunStepFunctionToolCall functionCall)
574+
{
575+
messages?.Add(GenerateFunctionResultContent(agent.GetName(), stepFunctionResults[step.Id], step));
576+
stepFunctionResults.Remove(step.Id);
577+
break;
578+
}
579+
580+
if (toolCall is RunStepCodeInterpreterToolCall codeCall)
581+
{
582+
messages?.Add(GenerateCodeInterpreterContent(agent.GetName(), codeCall.Input, step));
583+
}
584+
}
585+
}
586+
562587
private static ChatMessageContent GenerateMessageContent(string? assistantName, PersistentThreadMessage message, RunStep? completedStep = null, ILogger? logger = null)
563588
{
564589
AuthorRole role = new(message.Role.ToString());

‎dotnet/src/Agents/OpenAI/Internal/AssistantThreadActions.cs‎

Lines changed: 61 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -376,7 +376,7 @@ public static async IAsyncEnumerable<StreamingChatMessageContent> InvokeStreamin
376376
// Evaluate status and process steps and messages, as encountered.
377377
HashSet<string> processedStepIds = [];
378378
Dictionary<string, FunctionResultContent[]> stepFunctionResults = [];
379-
List<RunStep> stepsToProcess = [];
379+
List<RunStep> messageCreationStepsToProcess = [];
380380
ThreadRun? run = null;
381381

382382
FunctionCallsProcessor functionProcessor = new(logger);
@@ -389,7 +389,7 @@ public static async IAsyncEnumerable<StreamingChatMessageContent> InvokeStreamin
389389
// Check for cancellation
390390
cancellationToken.ThrowIfCancellationRequested();
391391

392-
stepsToProcess.Clear();
392+
messageCreationStepsToProcess.Clear();
393393

394394
await foreach (StreamingUpdate update in asyncUpdates.ConfigureAwait(false))
395395
{
@@ -436,7 +436,15 @@ public static async IAsyncEnumerable<StreamingChatMessageContent> InvokeStreamin
436436
switch (stepUpdate.UpdateKind)
437437
{
438438
case StreamingUpdateReason.RunStepCompleted:
439-
stepsToProcess.Add(stepUpdate.Value);
439+
if (!string.IsNullOrEmpty(stepUpdate.Value.Details.CreatedMessageId))
440+
{
441+
messageCreationStepsToProcess.Add(stepUpdate.Value);
442+
}
443+
else
444+
{
445+
ProcessToolCallStep(stepUpdate.Value, agent, messages, threadId, stepFunctionResults);
446+
}
447+
440448
break;
441449
default:
442450
break;
@@ -504,55 +512,69 @@ await functionProcessor.InvokeFunctionCallsAsync(
504512
}
505513
}
506514

507-
if (stepsToProcess.Count > 0)
515+
if (messageCreationStepsToProcess.Count > 0)
508516
{
509517
logger.LogOpenAIAssistantProcessingRunMessages(nameof(InvokeAsync), run!.Id, threadId);
510518

511-
foreach (RunStep step in stepsToProcess)
519+
foreach (RunStep step in messageCreationStepsToProcess)
512520
{
513-
if (!string.IsNullOrEmpty(step.Details.CreatedMessageId))
514-
{
515-
ThreadMessage? message =
516-
await RetrieveMessageAsync(
517-
client,
518-
threadId,
519-
step.Details.CreatedMessageId,
520-
agent.PollingOptions.MessageSynchronizationDelay,
521-
cancellationToken).ConfigureAwait(false);
522-
523-
if (message != null)
524-
{
525-
ChatMessageContent content = GenerateMessageContent(agent.GetName(), message, step);
526-
messages?.Add(content);
527-
}
528-
}
529-
else
530-
{
531-
foreach (RunStepToolCall toolCall in step.Details.ToolCalls)
532-
{
533-
if (toolCall.Kind == RunStepToolCallKind.Function)
534-
{
535-
messages?.Add(GenerateFunctionResultContent(agent.GetName(), stepFunctionResults[step.Id], step));
536-
stepFunctionResults.Remove(step.Id);
537-
break;
538-
}
539-
540-
if (toolCall.Kind == RunStepToolCallKind.CodeInterpreter)
541-
{
542-
messages?.Add(GenerateCodeInterpreterContent(agent.GetName(), toolCall.CodeInterpreterInput, step));
543-
}
544-
}
545-
}
521+
await ProcessMessageCreationStepAsync(step, agent, client, messages, threadId, cancellationToken).ConfigureAwait(false);
546522
}
547523

548-
logger.LogOpenAIAssistantProcessedRunMessages(nameof(InvokeAsync), stepsToProcess.Count, run!.Id, threadId);
524+
logger.LogOpenAIAssistantProcessedRunMessages(nameof(InvokeAsync), messageCreationStepsToProcess.Count, run!.Id, threadId);
549525
}
550526
}
551527
while (run?.Status != RunStatus.Completed);
552528

553529
logger.LogOpenAIAssistantCompletedRun(nameof(InvokeAsync), run?.Id ?? "Failed", threadId);
554530
}
555531

532+
private static async Task ProcessMessageCreationStepAsync(
533+
RunStep step,
534+
OpenAIAssistantAgent agent,
535+
AssistantClient client,
536+
IList<ChatMessageContent>? messages,
537+
string threadId,
538+
CancellationToken cancellationToken)
539+
{
540+
ThreadMessage? message =
541+
await RetrieveMessageAsync(
542+
client,
543+
threadId,
544+
step.Details.CreatedMessageId,
545+
agent.PollingOptions.MessageSynchronizationDelay,
546+
cancellationToken).ConfigureAwait(false);
547+
548+
if (message != null)
549+
{
550+
ChatMessageContent content = GenerateMessageContent(agent.GetName(), message, step);
551+
messages?.Add(content);
552+
}
553+
}
554+
555+
private static void ProcessToolCallStep(
556+
RunStep step,
557+
OpenAIAssistantAgent agent,
558+
IList<ChatMessageContent>? messages,
559+
string threadId,
560+
Dictionary<string, FunctionResultContent[]> stepFunctionResults)
561+
{
562+
foreach (RunStepToolCall toolCall in step.Details.ToolCalls)
563+
{
564+
if (toolCall.Kind == RunStepToolCallKind.Function)
565+
{
566+
messages?.Add(GenerateFunctionResultContent(agent.GetName(), stepFunctionResults[step.Id], step));
567+
stepFunctionResults.Remove(step.Id);
568+
break;
569+
}
570+
571+
if (toolCall.Kind == RunStepToolCallKind.CodeInterpreter)
572+
{
573+
messages?.Add(GenerateCodeInterpreterContent(agent.GetName(), toolCall.CodeInterpreterInput, step));
574+
}
575+
}
576+
}
577+
556578
private static ChatMessageContent GenerateMessageContent(string? assistantName, ThreadMessage message, RunStep? completedStep = null, ILogger? logger = null)
557579
{
558580
AuthorRole role = new(message.Role.ToString());

‎dotnet/src/Agents/OpenAI/OpenAIAssistantAgent.cs‎

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -289,19 +289,29 @@ IAsyncEnumerable<StreamingChatMessageContent> InternalInvokeStreamingAsync()
289289
}
290290

291291
// Return the chunks to the caller.
292+
int messageIndex = 0;
292293
await foreach (var result in invokeResults.ConfigureAwait(false))
293294
{
295+
// Notify the thread of any messages that were assembled from the streaming response during this iteration.
296+
await NotifyMessagesAsync().ConfigureAwait(false);
297+
294298
yield return new(result, openAIAssistantAgentThread);
295299
}
296300

297-
// Notify the thread of any new messages that were assembled from the streaming response.
298-
foreach (var newMessage in newMessagesReceiver)
299-
{
300-
await this.NotifyThreadOfNewMessage(openAIAssistantAgentThread, newMessage, cancellationToken).ConfigureAwait(false);
301+
// Notify the thread of any remaining messages that were assembled from the streaming response after all iterations are complete.
302+
await NotifyMessagesAsync().ConfigureAwait(false);
301303

302-
if (options?.OnIntermediateMessage is not null)
304+
async Task NotifyMessagesAsync()
305+
{
306+
for (; messageIndex < newMessagesReceiver.Count; messageIndex++)
303307
{
304-
await options.OnIntermediateMessage(newMessage).ConfigureAwait(false);
308+
ChatMessageContent newMessage = newMessagesReceiver[messageIndex];
309+
await this.NotifyThreadOfNewMessage(openAIAssistantAgentThread, newMessage, cancellationToken).ConfigureAwait(false);
310+
311+
if (options?.OnIntermediateMessage is not null)
312+
{
313+
await options.OnIntermediateMessage(newMessage).ConfigureAwait(false);
314+
}
305315
}
306316
}
307317
}

‎dotnet/src/Agents/OpenAI/OpenAIResponseAgent.cs‎

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,7 @@ public override async IAsyncEnumerable<AgentResponseItem<StreamingChatMessageCon
7979
// Invoke responses with the updated chat history.
8080
ChatHistory chatHistory = [.. messages];
8181
int messageCount = chatHistory.Count;
82+
int messageIndex = chatHistory.Count;
8283
var invokeResults = ResponseThreadActions.InvokeStreamingAsync(
8384
this,
8485
chatHistory,
@@ -89,17 +90,26 @@ public override async IAsyncEnumerable<AgentResponseItem<StreamingChatMessageCon
8990
// Return streaming chat message content to the caller.
9091
await foreach (var result in invokeResults.ConfigureAwait(false))
9192
{
93+
// Notify the thread of any messages that were assembled from the streaming response during this iteration.
94+
await NotifyMessagesAsync().ConfigureAwait(false);
95+
9296
yield return new(result, agentThread);
9397
}
9498

95-
// Notify the thread of new messages
96-
for (int i = messageCount; i < chatHistory.Count; i++)
97-
{
98-
await this.NotifyThreadOfNewMessage(agentThread, chatHistory[i], cancellationToken).ConfigureAwait(false);
99+
// Notify the thread of any remaining messages that were assembled from the streaming response after all iterations are complete.
100+
await NotifyMessagesAsync().ConfigureAwait(false);
99101

100-
if (options?.OnIntermediateMessage is not null)
102+
async Task NotifyMessagesAsync()
103+
{
104+
for (; messageIndex < chatHistory.Count; messageIndex++)
101105
{
102-
await options.OnIntermediateMessage(chatHistory[i]).ConfigureAwait(false);
106+
ChatMessageContent newMessage = chatHistory[messageIndex];
107+
await this.NotifyThreadOfNewMessage(agentThread, newMessage, cancellationToken).ConfigureAwait(false);
108+
109+
if (options?.OnIntermediateMessage is not null)
110+
{
111+
await options.OnIntermediateMessage(newMessage).ConfigureAwait(false);
112+
}
103113
}
104114
}
105115
}

0 commit comments

Comments
 (0)